做开发这些年我越来越确信一件事很多看似不起眼的“小模块”反而是项目稳定性的关键底座。前阵子整理代码仓库发现一个被我命名为“req”的通用请求模块已经悄悄支撑了好几个内部系统上亿次的调用量。这个模块没什么花哨功能就是把HTTP请求这件“小事”做到了极致该超时的地方超时该重试的地方重试该复用的连接绝不浪费。这篇文章就把我从零开始打磨“req”这个项目的过程完整写出来包括需求拆解、代码实现、参数取舍以及那些踩过坑之后才明白的细节。不管你是刚学Python、第一个爬虫还没写利索的新手还是已经写了几年接口调用、被各种超时重试折磨过的老手这篇文章应该都能给你一点启发。1. 项目解剖“req”到底是什么凭什么值得做成一个项目1.1 从缩写到实体一个HTTP请求模块的诞生背景“req”这个名字毫不避讳地说就是从英文request里砍出来的。它既是指HTTP协议里那个“请求”动作也指我落地的这一套围绕请求动作构建的代码集合。为什么要专门做一个模块而不是每次现写requests.get因为这个需求太高频了拉取订单数据要请求接口同步用户状态要请求接口调第三方服务要请求接口。如果每个地方都独立写一遍请求逻辑结果是灾难性的——有人忘了设超时有人把重试写成了死循环有人拿着裸的response.text就去解析最后线上冒出成堆的异常日志排查起来想骂人。把“req”当成一个正经项目来做目的就是把所有请求相关的“脏活累活”收拢到一处。这个模块对外暴露一两个方法就够了但内部要处理好连接复用、超时控制、重试策略、异常分类、日志埋点这些细节。好处是什么下游调用方代码极其干净出问题时又能顺着统一日志迅速定位。说白了这就是典型的“把复杂度关进笼子里”的思路。1.2 需求拆解一个通用请求模块该具备哪些能力动手之前我先给自己列了一份能力清单。这份清单不是拍脑袋出来的而是基于以往项目里真实摔过的跟头提炼的稳定的超时控制连接超时必须和读取超时分开设置。很多新手只设一个timeout以为够了实际上下游服务慢吞吞地返回数据时你的请求会一直挂在那边白白占着线程。合理的重试机制遇到连接错误、5xx状态码时允许重试但不能一遇到4xx就重试——客户端参数传错了重试一百次结果也一样只会放大错误。会话保持与连接复用同一个服务多次请求时复用底层TCP连接能显著降低延迟和资源消耗。规范化的响应处理直接返回解析好的JSON或者文本而不是把response对象丢给调用方让调用方自己去折腾。同时要能把HTTP状态码异常和网络异常统一成一套异常体系。日志与可观测性每个请求都能查到耗时、状态码、目标地址、异常原因方便排查。1.3 方案选型为什么是requests而不是urllib或httpx选型过程其实很短暂。Python标准库里的urllib功能上勉强够用但API设计还停留在上个时代处理Cookie、处理重定向、处理HTTPS证书都要额外写不少代码。用urllib做过一次复杂请求你就知道什么叫“把简单的事情搞复杂”。httpx这两年很火也支持异步但当时项目里全是同步逻辑引入异步能力属于过度设计。而且团队里其他人更熟悉requests的API生态资料多出问题随便一搜就有解。更重要的是requests作为老牌库经过了极其充分的生产环境验证各种边界情况基本都被踩平了。所以最终定的方案就是基于requests封装核心库只做薄薄一层可靠封装不搞魔改。这个选择看似保守但实际交付效率很高。2. 动手编码前先想明白的四个细节2.1 超时一个不设超时的请求就是一颗定时炸弹我见过太多线上事故的起点就是“少写一个timeout参数”。有一次排查一个定时任务卡死一看日志进程停在下游接口调用上整整三小时因为没有超时时间那个请求就一直在等待。等网络环境抖动恢复任务早过了调度周期后面全部堆积。超时需要分两层理解连接超时 connect timeout建立TCP连接的最长等待时间。如果目标IP不可达、端口被防火墙丢弃这个阶段会很快暴露问题。一般设置3到5秒足够。读取超时 read timeout建立连接后等待响应数据的最长时间。这个值要结合业务场景判断纯查询接口建议3秒内部批量接口可以放到10秒但要设置上限。在requests里超时用元组表示写成timeout(3.05, 10)。第一个是连接超时第二个是读取超时。注意这里的默认单位是秒但支持小数。requests的文档里特别提了一句连接超时建议用3.05秒这种带0.05的数因为很多服务器的SYN队列处理有时间粒度整数秒容易和别的请求撞在一起。这个技巧我实际验证过靠谱。注意永远不要省略timeout参数。如果实在拿不准先写一个保守值比如(5, 15)之后再根据监控数据校准。一个不设超时的请求本质上是在用系统稳定性赌运气。2.2 重试指数退避比“无脑重发三轮”靠谱得多重试机制听起来简单做起来全是坑。最简单的写法是for i in range(3): try: resp requests.get(url, timeout5) return resp except Exception: continue这种代码最大的问题是没有退避失败三次会在极短时间内全部重放。如果下游服务已经处于过载状态这种“三次闪电连打”会直接把它打挂。正确做法是指数退避加随机抖动第一次失败等1秒第二次等2秒第三次等4秒然后加上一个随机的0到1秒抖动避免多个客户端同时重试造成“惊群效应”。具体逻辑放到requests的HTTPAdapter里配置更优雅。给Session挂一个自定义Adapter指定max_retries和backoff_factor底层urllib3会自动完成退避重试不需要业务代码里写循环。还要注意一个原则只对连接错误和5xx重试对4xx一律不重试。4xx表示请求本身有问题重试是无效动作。请求体必须是可重放的——如果你POST了一个流式上传的文件重试时文件指针已经走到末尾这时候重试就会发一个空body出去坑死人。2.3 Session管理连接池和Cookie的复用是一门学问很多初学者用requests都是每次独立调用requests.get(...) requests.post(...)这种写法本身没错但如果每次请求都要执行TCP握手的完整流程DNS解析、TCP三次握手、TLS握手性能开销非常大。尤其在HTTPS接口上TLS握手消耗的时间可能比业务处理还长。改成Session之后就完全不同session requests.Session() session.get(...) session.post(...)Session内部维护一个连接池同一个主机名下的多个请求会复用底层TCP连接。不仅如此Session还会自动保存服务端返回的Cookie后续请求自动携带这在处理需要登录态的接口时非常方便。连接池本身也有效果参数可调。HTTPAdapter里的pool_connections控制不同主机可缓存的连接数pool_maxsize控制同一主机下最大连接数。常规场景用默认值就行高并发场景要结合线程数调大。我曾经在一个消费者的线程池里开了32个线程结果某台机器的连接池上限还是默认10高峰期大量请求排队等连接延迟飙升。后来定位到是适配器参数没配把pool_maxsize调到64之后立竿见影。提示Session用完必须调用session.close()或者用上下文管理器写法。别小看这一步不关闭的Session会一直占着连接池里的连接长时间运行的程序迟早把端口资源耗尽。我见过因为漏关Session程序跑了两天以后所有连接全部变成TIME_WAIT状态的情况。2.4 响应处理不要拿着response.text就从先干为敬requests返回的response对象里有个text属性很多人习惯直接用了。问题在于text是requests根据HTTP头猜测编码后解码出来的字符串。如果服务端没返回Content-Type头里的charset信息或者返回的charset和实际编码不一致中文就会变成乱码。更稳妥的做法是先明确预期响应格式。调用JSON接口时直接resp.json()让requests自己处理编码。如果是下载文件或抓取原始二进制用resp.content永远不要去碰text。对于接口响应先判断resp.status_code是否在预期范围再解析body不要只依赖异常机制。尽量设置resp.raise_for_status()这样一旦响应状态码是4xx或5xx直接抛出HTTPError异常不用自己手写状态码判断。还有一个小技巧永远先从响应里读取内容再判断响应对象是否被关闭。如果你在异常处理里访问response.text而response已经被释放会抛出一个ConnectionError导致真正的问题被掩盖。正确顺序是先拿到响应数据再决定抛什么异常。3. 核心代码实现一个能直接抄作业的请求模块3.1 模块骨架与初始化配置我最终实现的“req”模块对外只暴露一个类RequestClient内部封装了session管理、超时、重试、异常处理和日志。初始化时可以传入一组默认配置不同业务场景可以各建各的实例。import logging import socket from typing import Any, Dict, Optional, Tuple import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry logger logging.getLogger(req) class RequestClient: 通用HTTP请求客户端封装超时、重试、会话管理。 def __init__( self, base_url: str , timeout: Tuple[float, float] (3.05, 10), max_retries: int 3, backoff_factor: float 1.0, pool_connections: int 16, pool_maxsize: int 32, default_headers: Optional[Dict[str, str]] None, ): self.base_url base_url.rstrip(/) self.timeout timeout self.max_retries max_retries self.default_headers { User-Agent: req-client/1.0, Accept: application/json, text/plain, */*, Content-Type: application/json, **(default_headers or {}), } # 通过Session统一管理连接池和Cookie self.session requests.Session() self.session.headers.update(self.default_headers) # 配置重试策略连接错误、5xx状态码会触发重试 retry_strategy Retry( totalmax_retries, connectmax_retries, readmax_retries, statusmax_retries, backoff_factorbackoff_factor, status_forcelist[500, 502, 503, 504], allowed_methods[GET, POST, PUT, DELETE, HEAD, OPTIONS], raise_on_statusFalse, ) # 配置连接池大小 adapter HTTPAdapter( max_retriesretry_strategy, pool_connectionspool_connections, pool_maxsizepool_maxsize, ) self.session.mount(http://, adapter) self.session.mount(https://, adapter)3.2 核心请求方法完整实现请求方法的核心逻辑是拼URL、发请求、按状态码和响应格式做统一处理、抛自定义异常。我建议把GET和POST拆成两个方法语义更清晰但内部都落到一个私有方法_request里避免重复代码。class RequestError(Exception): 自定义请求异常基类 class RequestTimeoutError(RequestError): 请求超时异常 class RequestConnectionError(RequestError): 网络连接异常 class HttpResponseError(RequestError): HTTP状态码异常 def __init__(self, status_code: int, body: Any None): self.status_code status_code self.body body super().__init__(fHTTP {status_code}: {body})然后是实现请求发送的部分def _request( self, method: str, url: str, params: Optional[Dict[str, Any]] None, json: Optional[Dict[str, Any]] None, data: Optional[Any] None, headers: Optional[Dict[str, str]] None, timeout: Optional[Tuple[float, float]] None, stream: bool False, retries: Optional[int] None, ) - Dict[str, Any]: full_url f{self.base_url}{url} if url.startswith(/) else url request_headers {**self.default_headers, **(headers or {})} effective_timeout timeout or self.timeout # 每次请求可以临时覆盖重试次数 if retries is not None and retries ! self.max_retries: self._rebuild_adapter_retries(max_retriesretries) try: response self.session.request( methodmethod.upper(), urlfull_url, paramsparams, jsonjson, datadata, headersrequest_headers, timeouteffective_timeout, streamstream, ) except requests.exceptions.Timeout as exc: raise RequestTimeoutError(f请求超时: {exc}) from exc except requests.exceptions.ConnectionError as exc: raise RequestConnectionError(f连接失败: {exc}) from exc except requests.exceptions.RequestException as exc: raise RequestError(f请求异常: {exc}) from exc # 必须先读取body内容再判断状态码避免流被关闭 body response.content if stream else response.content if response.status_code 400: raise HttpResponseError( status_coderesponse.status_code, bodyself._safe_decode(body), ) return { status_code: response.status_code, headers: dict(response.headers), body: response.content, text: self._safe_decode(response.content), url: response.url, } staticmethod def _safe_decode(content: bytes) - str: for encoding in (utf-8, gbk, latin1): try: return content.decode(encoding) except UnicodeDecodeError: continue return content.decode(utf-8, errorsreplace) def get( self, url: str, params: Optional[Dict[str, Any]] None, headers: Optional[Dict[str, str]] None, timeout: Optional[Tuple[float, float]] None, stream: bool False, retries: Optional[int] None, ) - Dict[str, Any]: return self._request(GET, url, paramsparams, headersheaders, timeouttimeout, streamstream, retriesretries) def post( self, url: str, json: Optional[Dict[str, Any]] None, data: Optional[Any] None, headers: Optional[Dict[str, str]] None, timeout: Optional[Tuple[float, float]] None, retries: Optional[int] None, ) - Dict[str, Any]: return self._request(POST, url, jsonjson, datadata, headersheaders, timeouttimeout, retriesretries)针对临时调整重试次数的需求专门写一个内部方法来重建Adapter。这里有一个取舍如果频繁改重试参数会导致重复创建Adapter所以只允许在“确实和默认值不同”的时候才重建。大部分场景下调用方不会传retries这里的逻辑根本不会触发。def _rebuild_adapter_retries(self, max_retries: int) - None: retry_strategy Retry( totalmax_retries, connectmax_retries, readmax_retries, statusmax_retries, backoff_factor1.0, status_forcelist[500, 502, 503, 504], allowed_methods[GET, POST, PUT, DELETE, HEAD, OPTIONS], raise_on_statusFalse, ) adapter HTTPAdapter(max_retriesretry_strategy) self.session.mount(http://, adapter) self.session.mount(https://, adapter) def close(self) - None: self.session.close() def __enter__(self): return self def __exit__(self, *args): self.close()3.3 用模拟API完成一次完整验证代码写完必须实测。我本地起了一个简单的测试接口来做验证确认模块行为符不符合预期。先搭建一个最简陋的测试服务from flask import Flask, jsonify, request import time app Flask(__name__) app.route(/health) def health(): return jsonify({status: ok}) app.route(/delay/int:seconds) def delay(seconds: int): time.sleep(seconds) return jsonify({delayed: seconds}) app.route(/error/int:status) def error(status: int): return jsonify({error: 模拟错误}), status if __name__ __main__: app.run(port5000, debugFalse)然后用模块的客户端去请求with RequestClient(base_urlhttp://127.0.0.1:5000, timeout(2, 5), max_retries2) as client: # 正常请求 resp client.get(/health) print(resp[status_code], resp[text]) # 200 {status: ok} # 服务端延迟触发读取超时 try: client.get(/delay/8) # 超过读取超时5秒 except RequestTimeoutError as exc: print(捕获超时异常:, exc) # 服务端返回500触发重试 try: client.get(/error/500) except HttpResponseError as exc: print(捕获HTTP异常, 状态码:, exc.status_code)实测输出基本符合预期1秒内正常返回延迟8秒的请求在5秒后被判定超时返回500的请求经历了两轮重试后其实是3次总请求数最终抛出了HttpResponseError。日志里能清楚看到重试的时间间隔差不多是1秒、2秒符合指数退避的设置。注意测试重试逻辑时一定要观察重试的时间间隔。如果三次重试间隔基本为0说明退避没有生效多半是HTTPAdapter没有正确挂到Session上。另外在本地测试时如果服务端故意保持连接不返回数据不同操作系统的Socket行为有差异建议在超时参数里把read timeout调小到2秒左右更容易触发验证。4. 线上运行半年后我总结的常见问题与排查技巧4.1 连接池耗尽请求变慢往往从连接开始模块上线运行一段时间后我接到过反馈某个服务的接口响应忽然变慢从平均10毫秒涨到500毫秒而且持续时间不定。第一反应是下游服务出了问题但排查后发现下游一切正常瓶颈出在客户端自己身上——线程池不够用连接池连接数量不足请求在排队等待空闲连接导致响应时间被大幅拉长。连接池耗尽的典型症状是慢请求的时间分布呈“均匀锯齿状”CPU占用不高Thread Dump里大量线程卡在获取连接的逻辑上。解决办法是调大pool_maxsize但更重要的是搞清楚并发上限。如果你开了32个工作线程pool_maxsize至少要等于32最好留点余量调到48或64。同时pool_connections也要大于不同目标主机的数量否则连接池会频繁淘汰重建。4.2 重试反而加重了服务端雪崩重试是一把双刃剑。有一次下游服务因为数据库锁冲突整体响应时间拉长但错误率并没有明显上升。此时我们的请求由于设置了超时10秒大量请求先后触发读取超时再走重试机制等于把原来的一倍流量变成了三倍流量。下游越来越慢上游越来越急直接导致服务端连接数被打到上限。这次事故后我把重试策略调整成两个原则重试总量封顶单次请求最多重试2次增加随机抖动保障同一批请求不会在同一时刻重试。核心代码里Retry配置的backoff_factor1.0已经带了基础的退避但客户端之间还是可能同步所以我在外层再加了一层随机睡眠import random import time # 请求前的兜底如果距离上次请求时间极短添加微小的随机延迟 def _random_jitter_before_retry(retry_count: int) - None: time.sleep(min(2 ** retry_count, 8) random.uniform(0, 1))这个函数一般在请求异常处理中手动调用不要在HTTPAdapter层强行覆盖。原因在于HTTPAdapter的退避机制对连接错误的处理已经比较完善手工介入反而容易破坏原有逻辑。4.3 日志与监控请求模块也要有体检报告没有日志的请求模块等于裸奔。我第一次把req模块接入核心系统时只给调用方返回了结果没有任何日志输出。后来排查一个线上问题对方问“你这个请求到底发到哪了参数是什么耗时多久”我一时语塞。现在的实现里所有的请求无论成功失败都会输出结构化日志。关键字段包括目标URL、请求方法、状态码、耗时毫秒、异常类型、重试次数。def _log_request(method: str, url: str, status: int, elapsed_ms: float, err: str ): logger.info( method%s url%s status%s elapsed_ms%s err%s, method, url, status, elapsed_ms, err, )在_request的返回前和异常抛出前都调用这个函数。它的价值在故障复盘时尤其突出——对一份日志可以精确还原“请求从哪个实例发出、经过了多久、遇到了什么错误、重试了几次”。为了监控数据链路的完整性这里还会把请求跟踪ID写入Header下游服务也能看到同一笔请求的完整路径。4.4 一套可参考的常见问题速查表把半年多来线上和团队反馈的典型问题整理成一张速查表省得每次排查都从头看代码现象可能原因排查与解决请求偶尔变慢延迟呈锯齿状连接池大小不够线程在排队调大pool_maxsize观察线程数和连接数匹配度服务端报错但客户端收不到响应读取超时设置太短读取阶段被判定失败调大read timeout或检查服务端是否存在长轮询重试导致下游压力骤增重试策略过激进无抖动降低重试总次数增加随机抖动只对安全方法重试中文返回乱码响应头未指定charset直接用了text使用content解码或启用_safe_decode方式大量TIME_WAIT连接Session未关闭或者连接池参数不合理使用上下文管理器管理Session避免频繁创建请求Hung住不返回未设超时或者超时时间过长给所有请求强行设置connect/read超时4.5 性能与并发场景的进一步扩展“req”模块在同步场景下已经够用但如果你要处理大量并行请求只有一个Session实例还是不够的。线程安全方面requests的Session并非完全线程安全但实际项目中多个线程共用同一个Session的读写操作大多数情况下不会立即崩只是可能出现Header互相覆盖的问题。更稳妥的做法是使用threading.local()为每个线程单独维护Session。import threading class ThreadLocalSessionManager: def __init__(self, client_factory): self._local threading.local() self._client_factory client_factory property def client(self): if not hasattr(self._local, client): self._local.client self._client_factory() return self._local.client如果你需要的是更高吞吐那就直接把核心方法改成协程版本底层用httpx的AsyncClient但接口签名保持一致。这样调用方无感知性能又能上一大截。不过这一步我目前还没有大规模落地因为同步方案在现有流量下已经足够稳定改异步带来的心智负担反而是新的风险。对了还有一个细节值得说raise_on_status参数一定要设为False。我在第一次实现时没有设置这个参数urllib3默认情况下遇到5xx状态码会直接抛异常导致我自己定义的HttpResponseError根本不会触发。设置成False之后状态码判断的逻辑完全由自己掌控重试行为也更可预期。最后再分享一个小技巧给每次请求都打上请求ID。不管是放在Header里透传给下游还是记录在日志里这个ID在排查链路问题时是救命稻草。我自己会在客户端初始化时生成一个UUID前缀每个请求拼接一个自增序号效果非常理想。如果你正在设计自己的请求模块这个习惯越早养成越好。