urllib3,一个牢靠的 Python 库

大家好，我是木木。

今天给大家分享一个牢靠的 Python 库，urllib3。

urllib3

如果你平时更多用 requests 或 httpx，可能不会经常直接写 urllib3。但很多 Python HTTP 生态其实都绕不开它：连接池、重试、超时、TLS、代理、压缩和流式响应这些底层能力，urllib3 都打磨得很稳。它适合那些你想更直接掌控 HTTP 连接行为的场景。

项目地址：https://github.com/urllib3/urllib3

官方文档：https://urllib3.readthedocs.io

三大特点

连接池稳

PoolManager 会复用连接，适合服务端脚本、采集任务和内部网关这种频繁请求同一批主机的场景。

重试清楚

Retry 可以明确控制哪些状态码、哪些方法、重试几次，不用把重试逻辑散在业务代码里。

底层可控

超时、TLS、代理、流式读取和连接释放都能细调，更适合需要掌控网络边界的项目。

最佳实践

安装方式：python -m pip install urllib3

这篇全部用本地 HTTP 服务演示，不依赖外部网络。重点看三个实践：连接池如何复用、服务短暂 503 时如何自动重试，以及大响应或持续输出时如何流式读取并释放连接。

功能一：用 PoolManager 复用连接和统一请求配置

这段代码解决什么问题：真实任务里，经常会连续请求同一个 API 域名。如果每次都重新建连接，不仅慢，也更容易把服务端连接数打高。PoolManager 可以统一 headers、timeout，并复用底层连接。

importjsonimportthreadingimporttimefromhttp.serverimportBaseHTTPRequestHandler,ThreadingHTTPServerimporturllib3PORT=18951classHandler(BaseHTTPRequestHandler):defdo_GET(self):ifself.path.startswith("/items"):body=json.dumps({"path":self.path,"agent":self.headers.get("User-Agent")}).encode()self.send_response(200)self.send_header("Content-Type","application/json")self.send_header("Content-Length",str(len(body)))self.end_headers()self.wfile.write(body)else:self.send_response(404)self.end_headers()deflog_message(self,fmt,*args):passserver=ThreadingHTTPServer(("127.0.0.1",PORT),Handler)thread=threading.Thread(target=server.serve_forever,daemon=True)thread.start()time.sleep(0.1)try:http=urllib3.PoolManager(num_pools=2,maxsize=2,headers={"User-Agent":"urllib3-demo"},timeout=urllib3.Timeout(connect=1.0,read=2.0),)base=f"http://127.0.0.1:{PORT}"first=http.request("GET",f"{base}/items",fields=[("tag","pool"),("tag","http")])second=http.request("GET",f"{base}/items",fields={"tag":"reuse"})pool=http.connection_from_url(base)print("first :",first.status,first.json())print("second:",second.status,second.json())print("pool  :",pool.num_connections,"conn /",pool.num_requests,"requests")finally:server.shutdown()server.server_close()

这个例子里两次请求都走同一个连接池，公共请求头和超时也集中在 PoolManager 上。做内部 SDK、批量采集或后端任务时，这种集中配置会比到处裸调更容易维护。

功能二：用 Retry 把短暂失败收敛成明确规则

这段代码解决什么问题：接口偶发返回 503、网关短暂抖动、服务正在重启，这些情况不一定要直接失败。urllib3.util.retry.Retry 能把“哪些状态码要重试、哪些方法允许重试、最多试几次”写得很明确。

importjsonimportthreadingimporttimefromhttp.serverimportBaseHTTPRequestHandler,ThreadingHTTPServerimporturllib3fromurllib3.util.retryimportRetryPORT=18952attempts={"count":0}classHandler(BaseHTTPRequestHandler):defdo_GET(self):ifself.path=="/unstable":attempts["count"]+=1ifattempts["count"]<3:self.send_response(503)self.send_header("Content-Length","0")self.end_headers()returnbody=json.dumps({"ok":True,"attempt":attempts["count"]}).encode()self.send_response(200)self.send_header("Content-Type","application/json")self.send_header("Content-Length",str(len(body)))self.end_headers()self.wfile.write(body)else:self.send_response(404)self.end_headers()deflog_message(self,fmt,*args):passserver=ThreadingHTTPServer(("127.0.0.1",PORT),Handler)thread=threading.Thread(target=server.serve_forever,daemon=True)thread.start()time.sleep(0.1)try:retry=Retry(total=3,status_forcelist=[503],allowed_methods={"GET"},backoff_factor=0)http=urllib3.PoolManager(retries=retry)response=http.request("GET",f"http://127.0.0.1:{PORT}/unstable")print("status :",response.status)print("body   :",response.json())print("tries  :",attempts["count"])print("history:",len(response.retries.history))finally:server.shutdown()server.server_close()

这里前两次返回 503，第三次才成功。好处是业务层只拿到最终响应，同时还能从 response.retries.history 看到中间发生过几次重试，排查问题时不会完全丢证据。

环境与版本信息

• Demo 环境：Windows 11，Python 3.11
• 本文安装的 urllib3 版本：2.2.1
• 当前仓库 pyproject.toml 要求 Python >=3.10
• 可选能力包括：brotli、zstd、socks、h2
• GitHub 最近一次推送时间：2026-04-21T18:44:42Z

高级功能

这段代码解决什么问题：下载大文件、读取日志流或者处理持续输出时，不应该一次性把响应体全部放进内存。preload_content=False 可以让你边读边处理，最后再显式释放连接。

importthreadingimporttimefromhttp.serverimportBaseHTTPRequestHandler,ThreadingHTTPServerimporturllib3PORT=18953classHandler(BaseHTTPRequestHandler):defdo_GET(self):ifself.path=="/events":chunks=[b"alpha\n",b"beta!\n",b"gamma\n"]self.send_response(200)self.send_header("Content-Type","text/plain")self.send_header("Content-Length",str(sum(len(c)forcinchunks)))self.end_headers()forchunkinchunks:self.wfile.write(chunk)self.wfile.flush()time.sleep(0.02)else:self.send_response(404)self.end_headers()deflog_message(self,fmt,*args):passserver=ThreadingHTTPServer(("127.0.0.1",PORT),Handler)thread=threading.Thread(target=server.serve_forever,daemon=True)thread.start()time.sleep(0.1)try:http=urllib3.PoolManager()response=http.request("GET",f"http://127.0.0.1:{PORT}/events",preload_content=False)print("status:",response.status)foridx,chunkinenumerate(response.stream(6),1):print(f"chunk {idx}:",chunk.decode().strip())response.release_conn()print("closed:",response.closed)finally:server.shutdown()server.server_close()

这个模式适合大文件下载、导出任务和日志流。关键点是：既然你选择了手动读取响应，就要在读完后释放连接，让连接池可以继续健康复用。

适用场景

• 你需要更直接地控制连接池、超时、重试、代理或 TLS 行为
• 你在写底层 SDK、采集任务、网关探测或内部网络工具
• 你希望 HTTP 请求层足够稳定，而且能清楚记录失败与重试过程

不适用场景

• 你只是写少量普通业务请求，更想要更高层、手感更接近应用代码的客户端
• 你需要同步和异步 API 都在同一个库里统一管理
• 你主要关注浏览器自动化、页面渲染或前端交互测试

上线检查

1. 根据服务特性明确连接池大小、连接超时和读取超时，不要直接沿用演示值
2. 重试只放在幂等或可接受重复执行的请求上，尤其要谨慎处理 POST、支付和写入接口
3. 使用流式响应时，确认异常路径也会关闭或释放连接，避免连接池被慢慢耗尽

总结

如果你想更贴近 HTTP 底层，把连接池、重试、超时和流式读取这些边界控制得更可靠，urllib3 依然是 Python 生态里很值得信任的一块基石。