核心内容摘要
阿里小云KWS模型与ROS系统的集成：智能机器人语音控制
FastAPI 接口限流具体代码示例以下是基于slowapi库的 FastAPI 接口限流完整实现包含基础限流、自定义配置、分布式适配等场景可直接复制到项目中使用适配单机/分布式部署需求。
基础依赖安装先安装核心依赖单机限流仅需基础包分布式需额外安装 Redis 依赖# 基础依赖单机限流pipinstallfastapi uvicorn slowapi# 分布式限流额外依赖Redispipinstalllimits[redis]redis
核心代码示例场景1单机基础限流基于IP维度适用于单机部署基于客户端IP限流支持全局规则与局部规则结合包含自定义限流响应。
fromfastapiimportFastAPI,Request,Dependsfromfastapi.responsesimportJSONResponsefromslowapiimportLimiter,_rate_limit_exceeded_handlerfromslowapi.utilimportget_remote_address# 基于IP限流的关键函数fromslowapi.errorsimportRateLimitExceeded#
初始化FastAPI应用appFastAPI(title基础接口限流示例,version
1.
#
初始化限流器# key_funcget_remote_address以客户端IP作为限流标识# default_limits全局默认限流规则所有接口适用局部可覆盖limiterLimiter(key_funcget_remote_address,default_limits[20/minute]# 全局默认每分钟最多20次请求)#
绑定限流器与异常处理器app.state.limiterlimiter# 注册自定义限流响应处理器替代默认响应返回标准化JSONasyncdefcustom_rate_limit_handler(request:Request,exc:RateLimitExceeded):returnJSONResponse(status_code429,content{code:429,message:f请求过于频繁请{exc.retry_after}秒后重试,retry_after:exc.retry_after,request_path:request.url.path},headers{Retry-After:str(exc.retry_after)}# 规范响应头告知重试时间)app.add_exception_handler(RateLimitExceeded,custom_rate_limit_handler)#
接口实现局部限流规则覆盖全局app.get(/api/public,summary公开接口使用全局限流规则)asyncdefpublic_api():无需认证的公开接口使用全局默认限流20次/分钟return{msg:公开接口响应,status:success}app.get(/api/auth/me,summary需认证接口自定义局部限流)limiter.limit(10/minute)# 局部规则每分钟10次覆盖全局asyncdefauth_api():需认证接口设置更严格的限流规则return{msg:认证接口响应,user:current_user}app.post(/api/submit,summary写操作接口严格限流)limiter.limit(5/minute)# 写操作限流更严格5次/分钟asyncdefsubmit_data():创建/提交类接口防止恶意提交return{msg:数据提交成功,status:success}# 启动服务if__name____main__:importuvicorn uvicorn.run(main:app,host
0.
0.
0,port8000,reloadTrue)场景2基于用户ID限流已认证接口适用于已实现JWT认证的项目以用户ID作为限流标识避免同一用户多IP绕过限流与原文档JWT认证逻辑无缝衔接。
fromfastapiimportFastAPI,Depends,HTTPExceptionfromfastapi.securityimportOAuth2PasswordBearerfromjoseimportJWTError,jwtfromslowapiimportLimiterfromslowapi.utilimportget_remote_addressfromslowapi.errorsimportRateLimitExceeded# -------------------------- 原有JWT认证逻辑复用原项目代码 --------------------------SECRET_KEYyour-secure-secret-key# 生产环境用openssl rand -hex 32生成ALGORITHMHS256oauth2_schemeOAuth2PasswordBearer(tokenUrltoken)# 模拟用户数据库实际替换为真实数据库查询fake_users_db{admin:{id:1,username:admin,role:admin},testuser:{id:2,username:testuser,role:user}}asyncdefget_current_user(token:strDepends(oauth2_scheme)):获取当前登录用户原项目依赖项credentials_exceptionHTTPException(status_code401,detail无法验证凭据,headers{WWW-Authenticate:Bearer})try:payloadjwt.decode(token,SECRET_KEY,algorithms[ALGORITHM])username:strpayload.get(sub)ifusernamenotinfake_users_db:raisecredentials_exceptionreturnfake_users_db[username]exceptJWTError:raisecredentials_exception# -------------------------- 自定义用户ID限流逻辑 --------------------------appFastAPI(title基于用户ID的接口限流示例)# 自定义key_func以用户ID作为限流标识已认证接口专用asyncdefget_user_id(current_user:dictDepends(get_current_user)):returncurrent_user[id]# 用用户ID替代IP精准限流单个用户# 初始化限流器区分公开接口与认证接口# 公开接口基于IP限流ip_limiterLimiter(key_funcget_remote_address,default_limits[15/minute])# 认证接口基于用户ID限流user_limiterLimiter(key_funcget_user_id,default_limits[10/minute])# 绑定限流器与异常处理器app.state.ip_limiterip_limiter app.state.user_limiteruser_limiter app.add_exception_handler(RateLimitExceeded,_rate_limit_exceeded_handler)# -------------------------- 接口实现 --------------------------app.post(/token,summary获取JWT令牌公开接口IP限流)ip_limiter.limit(5/minute)# 登录接口严格限流防止暴力破解asyncdeflogin_for_access_token(form_data:OAuth2PasswordRequestFormDepends()):# 此处省略原有认证逻辑直接返回模拟Tokenreturn{access_token:mock-token,token_type:bearer}app.get(/api/user/profile,summary用户个人中心基于用户ID限流)user_limiter.limit(8/minute)# 单个用户每分钟8次请求asyncdefuser_profile(current_user:dictDepends(get_current_user)):return{user:current_user,msg:个人中心数据}app.put(/api/user/info,summary更新用户信息基于用户ID限流)user_limiter.limit(3/minute)# 更新操作更严格asyncdefupdate_user_info(current_user:dictDepends(get_current_user)):return{msg:用户信息更新成功,user_id:current_user[id]}场景3分布式限流Redis支持适用于多实例部署通过Redis共享限流计数确保跨节点限流一致性解决单机限流计数隔离问题。
fromfastapiimportFastAPIfromslowapiimportLimiterfromslowapi.utilimportget_remote_addressfromslowapi.errorsimportRateLimitExceededfromlimits.storageimportRedisStorage# Redis存储依赖fromlimits.strategiesimportSlidingWindowRateLimiter# 滑动窗口算法importredis#
连接Redis生产环境需配置Redis集群/哨兵redis_clientredis.Redis(hostlocalhost,# Redis服务地址port6379,# 端口db0,# 数据库编号password,# 若有密码需填写decode_responsesTrue)#
初始化Redis存储与限流器# 存储使用Redis共享限流计数# 算法滑动窗口算法避免固定窗口边界突发流量问题limiterLimiter(key_funcget_remote_address,storageRedisStorage(redis_client),# 绑定Redis存储strategySlidingWindowRateLimiter(),# 选择限流算法default_limits[30/minute]# 全局默认规则)#
初始化应用并绑定配置appFastAPI(title分布式接口限流示例Redis)app.state.limiterlimiter app.add_exception_handler(RateLimitExceeded,_rate_limit_exceeded_handler)#
测试接口app.get(/api/distributed/test,summary分布式限流测试接口)limiter.limit(15/minute)# 局部规则覆盖全局asyncdefdistributed_test():return{msg:分布式限流接口响应,storage:redis}app.post(/api/distributed/submit,summary分布式写操作接口)limiter.limit(5/minute)asyncdefdistributed_submit():return{msg:数据提交成功分布式环境安全限流}if__name____main__:importuvicorn uvicorn.run(main:app,host
0.
0.
0,port8000,reloadTrue)
关键配置说明
限流规则格式通过limiter.limit()指定规则格式为「请求次数/时间单位」支持的时间单位second秒如5/second每秒5次minute分如20/minute每分钟20次hour时如100/hour每小时100次day天如500/day每天500次
限流算法选择slowapi支持3种核心算法根据业务需求选择固定窗口FixedWindowRateLimiter默认算法实现简单可能存在窗口边界突发流量问题。
滑动窗口SlidingWindowRateLimiter精准限流避免边界问题适合对流量控制要求高的场景如支付接口。
滑动窗口计数器SlidingWindowCounterRateLimiter性能与精准度均衡适合高并发场景。
生产环境
注意事项分布式部署必须使用 Redis 存储否则各实例独立计数限流失效。
Redis 需配置持久化避免重启后限流计数丢失。
针对高频接口如查询接口可放宽限流写操作/敏感接口登录、支付需严格限流。
监控 429 状态码频率若突发激增需排查是否存在恶意攻击及时调整规则。
测试方法启动服务后通过 Swagger UIhttp://localhost:8000/docs快速重复调用接口。
当请求次数超过限流规则时接口返回 429 状态码携带自定义提示信息。
分布式场景可启动多个服务实例用同一IP调用接口验证 Redis 共享计数是否生效。