1. 项目背景业务场景某开放平台每天处理 200 万次第三方 API 调用。平台团队需要在请求链路中实现以下横切关注点QPS 限流每个租户每秒不超过 100 次、慢请求告警 500ms、默认超时注入防止客户端忘记设 timeout、灰度路由根据用户 ID 将 10% 流量导到新版本服务。最初方案是在每个 API handler 中手动添加这些逻辑——但 50 个 handler × 4 个关注点 200 处代码重复。维护噩梦当 QPS 阈值需要调整时需要修改 50 个文件。团队决定实现一个可组合的中间件链——像 Express.js 或 Django Middleware 那样每个中间件只关心一件事中间件链按顺序执行。痛点问题一横切关注点散落在各处。限流、日志、超时、灰度路由——这些关注点与具体业务逻辑无关它们属于基础设施但因为 requests 没有原生中间件机制它们散落在每个requests.get()调用前后。当需要新增一个关注点时比如安全团队要求所有请求添加签名需要在所有调用点修改。问题二装饰器地狱Decorator Hell。为了解决重复代码有些团队用装饰器来包装请求函数。但装饰器的嵌套顺序难以管理——timeout应该在retry外层还是内层当有 5 个装饰器时代码变成了不可读的千层饼。问题三requests 的扩展点不足以覆盖所有场景。Session hooks 只在响应后触发、Adapter 在传输层太底层、AuthBase 只处理认证——没有一个扩展点能在请求发出前这个黄金位置注入逻辑比如修改 URL、修改请求体、切换代理。需要一种更灵活的中间件模式。中间件洋葱模型请求 ──────────────────────────────────────── 响应 │ │ ├─ Middleware 1: 限流检查 (pre) │ │ ├─ Middleware 2: 超时注入 (pre) │ │ │ ├─ Middleware 3: 灰度路由 (pre) │ │ │ │ ├─ Middleware 4: 日志记录 (pre) │ │ │ │ │ ├── HTTP 请求 ── │ │ │ │ │ ├── HTTP 响应 ── │ │ │ │ ├─ Middleware 4: 日志记录 (post) │ │ │ ├─ Middleware 3: 灰度路由 (post) │ │ ├─ Middleware 2: 超时注入 (post) │ ├─ Middleware 1: 限流检查 (post) │ │ │ 返回给调用者 ────────────────────────────────┘2. 项目设计小胖在 50 个文件中搜索替换限流阈值“大师救命安全团队要求把 QPS 阈值从 100 改成 50——但这个阈值写在 50 个不同的 API handler 里我改到第 15 个就改错了漏了一个。能不能像 Django Middleware 那样只需要在一个地方配置就全局生效”大师“这就是中间件模式要解决的问题。虽然 requests 本身没有像 Django 那样的 middleware 机制但你可以自己实现一个——用装饰器链或类包装。本质是把请求处理过程拆成多个独立的’切面’每个切面只做一件事。”小胖“那跟装饰器有什么区别我试过用retry、log、timeout装饰器但很快就变成了十层嵌套。”大师“装饰器的痛点在于嵌套顺序不透明——你很难一眼看出retry(log(func))和log(retry(func))的区别。中间件模式通过一个显式的列表来管理执行顺序每个中间件有pre和post两个钩子”classMiddlewareChain:def__init__(self):self.middlewares[]defadd(self,middleware):self.middlewares.append(middleware)defexecute(self,request_func,*args,**kwargs):# 洋葱模型——从外到内执行 pre从内到外执行 postdefcall(i):ifilen(self.middlewares):returnrequest_func(*args,**kwargs)mwself.middlewares[i]# pre 处理mw.before_request(*args,**kwargs)try:resultcall(i1)# post 处理returnmw.after_response(result)exceptExceptionase:returnmw.on_error(e)returncall(0)小白追问“那怎么实现一个具体的中间件比如限流器——我要限制每个租户每秒不超过 50 次请求。”大师“每个中间件实现三个方法before_request请求前、after_response响应后、on_error异常时。限流器在before_request中检查令牌桶”importtimefromcollectionsimportdefaultdictclassRateLimitMiddleware:def__init__(self,max_qps50):self.max_qpsmax_qps self._bucketsdefaultdict(list)defbefore_request(self,tenant_id,*args,**kwargs):nowtime.time()bucketself._buckets[tenant_id]# 清理超过 1 秒的旧请求bucket[:][tfortinbucketifnow-t1.0]iflen(bucket)self.max_qps:raiseRateLimitExceeded(tenant_id)bucket.append(now)defafter_response(self,response):returnresponse小胖眼睛放光“那就可以把所有横切关注点放在一个配置文件中按顺序排列——想看执行顺序就一目了然”大师“正是。而且中间件之间可以共享上下文——比如before_request记录了请求耗时起始点after_response可以从上下文中读取起始点来计算耗时。这比装饰器优雅得多。”生活比喻技术映射机场安检流水线中间件链多个检查点每个关卡只查一件事单一职责中间件安检-海关-登机口有顺序中间件的洋葱执行顺序护照检查盖章记录信息上下文在中间件间传递VIP 快速通道特定条件跳过某个中间件3. 项目实战环境准备pipinstallrequests分步实现步骤一中间件引擎——洋葱模型核心目标实现一个通用的中间件链引擎。importrequestsimporttimeimportfunctoolsfromtypingimportCallable,Any,List,OptionalclassMiddlewareContext:在中间件间传递的上下文对象def__init__(self):self.data{}self.start_timeNoneclassBaseMiddleware:中间件基类defbefore(self,ctx:MiddlewareContext,method:str,url:str,**kwargs)-None:passdefafter(self,ctx:MiddlewareContext,response:requests.Response)-requests.Response:returnresponsedefon_error(self,ctx:MiddlewareContext,error:Exception):raiseerrorclassMiddlewareChain:可组合的中间件链——洋葱模型def__init__(self):self._middlewares:List[BaseMiddleware][]defuse(self,middleware:BaseMiddleware)-MiddlewareChain:添加一个中间件到链尾self._middlewares.append(middleware)returnselfdefexecute(self,session:requests.Session,method:str,url:str,**kwargs)-requests.Response:执行请求——穿过所有中间件ctxMiddlewareContext()ctx.start_timetime.perf_counter()def_call_next(index:int)-requests.Response:ifindexlen(self._middlewares):# 没有更多中间件了——执行真实的 HTTP 请求returnsession.request(method,url,**kwargs)mwself._middlewares[index]try:# pre 阶段mw.before(ctx,method,url,**kwargs)# 进入下一个中间件或真实请求response_call_next(index1)# post 阶段returnmw.after(ctx,response)exceptExceptionase:# 错误处理阶段try:returnmw.on_error(ctx,e)exceptException:raise# 中间件无法处理继续向上抛return_call_next(0)# 内置中间件 classLoggingMiddleware(BaseMiddleware):日志中间件——记录请求耗时和状态码defbefore(self,ctx,method,url,**kwargs):ctx.data[req_start]time.perf_counter()defafter(self,ctx,response):elapsedtime.perf_counter()-ctx.data.get(req_start,ctx.start_time)sizelen(response.content)print(f[LOG]{response.request.method}{response.url}f-{response.status_code}({elapsed:.3f}s,{size}B))ifelapsed1.0:print(f [慢请求] 耗时{elapsed:.2f}s 1s!)returnresponseclassTimeoutMiddleware(BaseMiddleware):超时注入中间件——强制设置请求超时def__init__(self,default_timeout:tuple(3,30)):self.default_timeoutdefault_timeoutdefbefore(self,ctx,method,url,**kwargs):iftimeoutnotinkwargsorkwargs[timeout]isNone:kwargs[timeout]self.default_timeout ctx.data[timeout_injected]TrueclassRateLimitMiddleware(BaseMiddleware):限流中间件——基于令牌桶算法def__init__(self,max_qps:int10):self.max_qpsmax_qps self._tokensmax_qps self._last_refilltime.time()defbefore(self,ctx,method,url,**kwargs):# 令牌桶补充nowtime.time()elapsednow-self._last_refill self._tokensmin(self.max_qps,self._tokenselapsed*self.max_qps)self._last_refillnowifself._tokens1:raiseRuntimeError(fQPS 限流! 当前上限:{self.max_qps}/s)self._tokens-1# 组装使用 chainMiddlewareChain()chain.use(RateLimitMiddleware(max_qps10))chain.use(TimeoutMiddleware(default_timeout(3,10)))chain.use(LoggingMiddleware())sessionrequests.Session()print( 中间件链演示 )foriinrange(5):respchain.execute(session,GET,https://httpbin.org/get,params{i:i})print(f #{i}:{resp.status_code})session.close()步骤二生产级中间件——灰度路由 熔断目标实现灰度路由和简单熔断中间件。importhashlibimporttimeimportrandomclassCanaryMiddleware(BaseMiddleware):灰度路由中间件——按用户 ID 哈希将部分流量导向 canary 服务 用法: CanaryMiddleware(canary_hosthttps://canary.api.com, traffic_pct10) # 10% 流量到 canary def__init__(self,canary_host:str,traffic_pct:int10,key_header:strX-User-Id):self.canary_hostcanary_host self.traffic_pcttraffic_pct self.key_headerkey_headerdefbefore(self,ctx,method,url,**kwargs):# 从请求头中获取用户标识headerskwargs.get(headers,{})user_idheaders.get(self.key_header,headers.get(self.key_header.lower(),))ifnotuser_id:return# 一致性哈希——同一用户始终路由到同一版本hash_valint(hashlib.md5(user_id.encode()).hexdigest()[:8],16)buckethash_val%100ifbucketself.traffic_pct:# 切换到 canary 主机importre new_urlre.sub(rhttps?://[^/],self.canary_host,kwargs.get(url,url),)kwargs[url]new_url ctx.data[canary_routed]Trueprint(f [灰度]{user_id}- canary ({bucket}{self.traffic_pct}))classCircuitBreakerMiddleware(BaseMiddleware):熔断器中间件——连续失败 N 次后短路 状态机: CLOSED - OPEN - HALF_OPEN - CLOSED (或 OPEN) def__init__(self,failure_threshold:int5,recovery_timeout:int30):self.thresholdfailure_threshold self.recoveryrecovery_timeout self._failures0self._last_failure0self._stateCLOSED# CLOSED | OPEN | HALF_OPENdefbefore(self,ctx,method,url,**kwargs):ifself._stateOPEN:iftime.time()-self._last_failureself.recovery:self._stateHALF_OPENprint(f [熔断] 进入 HALF_OPEN 状态尝试探测请求)else:raiseRuntimeError(f熔断器 OPEN——服务不可用{self.recovery}s 后再试)defafter(self,ctx,response):ifself._stateHALF_OPENandresponse.ok:self._stateCLOSEDself._failures0print(f [熔断] 探测成功熔断器恢复 CLOSED)returnresponsedefon_error(self,ctx,error):self._failures1self._last_failuretime.time()ifself._failuresself.threshold:self._stateOPENprint(f [熔断] 连续失败{self._failures}次熔断器 OPEN)raiseerror步骤三基于装饰器的轻量中间件目标对于不需要完整中间件链的场景提供装饰器方案。importfunctoolsimportrequestsdefwith_timeout(connect:float3,read:float30):装饰器强制注入超时配置 用法: with_timeout(connect3, read10) defdecorator(func):functools.wraps(func)defwrapper(*args,**kwargs):kwargs.setdefault(timeout,(connect,read))returnfunc(*args,**kwargs)returnwrapperreturndecoratordefwith_retry(max_retries:int3,backoff:float0.5):装饰器自动重试 用法: with_retry(max_retries3, backoff0.5) importtimeas_timedefdecorator(func):functools.wraps(func)defwrapper(*args,**kwargs):forattemptinrange(max_retries1):try:returnfunc(*args,**kwargs)except(requests.exceptions.Timeout,requests.exceptions.ConnectionError)ase:ifattemptmax_retries:raisedelaybackoff*(2**attempt)_time.sleep(delay)returnNonereturnwrapperreturndecoratorwith_timeout(connect3,read10)with_retry(max_retries2,backoff0.3)deffetch_data(url:str,**kwargs)-requests.Response:自动带超时和重试的请求函数returnrequests.get(url,**kwargs)可能遇到的坑及解决方法坑1中间件链中的异常被吞掉# 如果某个中间件的 on_error 没有 re-raise异常会被静默吞掉# 默认实现应该 re-raise显式处理时才截获坑2中间件顺序至关重要# 错误的顺序: 限流 - 日志 - 超时注入# 问题: 超时还没注入日志已记录了 start time限流已扣了令牌# 正确的顺序: 超时注入 - 限流 - 日志坑3kwargs 在中间件链中是可变的# 中间件的 before() 可以修改 kwargs如注入 timeout# 后续中间件和真实请求看到的是修改后的 kwargs# 这既是功能也是陷阱——需要文档说明哪些中间件会修改 kwargs测试验证importpytestimportrequestsclassTestMiddlewareChain:验证中间件链deftest_logging_middleware(self):chainMiddlewareChain().use(LoggingMiddleware())sessionrequests.Session()respchain.execute(session,GET,https://httpbin.org/get)assertresp.status_code200session.close()deftest_timeout_middleware_injects(self):chainMiddlewareChain().use(TimeoutMiddleware((1,5)))sessionrequests.Session()respchain.execute(session,GET,https://httpbin.org/get)assertresp.status_code200session.close()deftest_rate_limit_allows_then_blocks(self):mwRateLimitMiddleware(max_qps5)ctxMiddlewareContext()# 前 5 次应该通过for_inrange(5):mw.before(ctx,GET,https://example.com)# 第 6 次应该被限流withpytest.raises(RuntimeError,match限流):mw.before(ctx,GET,https://example.com)4. 项目总结核心知识点方案复杂度灵活性适用场景装饰器低中简单请求包装1-3 个关注点中间件链中高多个横切关注点4自定义 Adapter中高传输层拦截Session Hooks低低仅响应后处理优点 缺点对比维度中间件链装饰器嵌套自定义 Adapter执行顺序可见性优显式列表差嵌套顺序不直观优pre post 支持优良仅 send动态启用/禁用优中中学习成本中低中适用场景多横切关注点限流 超时 日志 灰度 熔断可观测性注入自动为所有请求添加追踪和指标API 网关路由、认证、限流、日志的中间件流水线注意事项中间件顺序决定执行逻辑——按依赖关系排列before()中抛异常会中断整个请求链on_error()默认应 re-raise除非中间件的职责就是容错中间件间通过MiddlewareContext传递数据常见踩坑经验案例一限流中间件放在超时注入前面。请求限流被扣除令牌后下一个中间件设置了更短的 timeout——请求超时失败。但令牌已经被扣了用户发现什么都没做却被告知已限流。根因限流应该在所有请求修改完成后、真实发送前执行。修复限流放在中间件链最内层贴近真实请求。案例二灰度路由修改 URL 后日志记录的是原 URL。灰度中间件把 10% 流量导向了 canary 环境。但日志中间件在灰度中间件之前执行日志记录的是原 URL——运维排查时找不到 canary 环境的请求日志。根因日志在修改前记录了 URL。修复日志放在中间件链的最外层最先执行 before、最后执行 after确保记录到最终状态。案例三熔断器恢复后流量过载。熔断器 OPEN 30 秒后进入 HALF_OPEN但恢复瞬间大量积压请求同时到达导致再次 OPEN。根因HALF_OPEN 没有限制探测流量。修复HALF_OPEN 状态下只允许少量请求通过如每秒 1 个。思考题设计题设计一个分布式限流中间件——使用 Redis 作为计数器存储实现跨进程/跨 Pod 的全局 QPS 限流。需要考虑 Redis 不可用时的降级策略本地限流 fallback。链式题你实现的中间件链是同步的。如果未来需要支持异步 HTTP 请求httpx/asyncio中间件链需要如何改造考虑async def before()和async def after()的兼容性。延伸阅读与资源Milvus向量数据库实战修炼从 0 到 1精通向量检索与生产落地后端工程师的 AI 转型第一课Ollama 与私有化大模型实战10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析