拓冰网站定制 
1. 为什么还要手写一个HTTP代理需求场景与破局点前后端联调的时候抓包工具一开流量全部走系统代理部分不走代理的SDK请求就得手动配一个HTTP_PROXY环境变量。做爬虫的同学更熟悉代理池搭起来之后请求得不停更换出口IP本质就是在HTTP层做一个转发。还有做内网调试的本地服务需要暴露给外部临时访问很多人第一反应是找现成的工具但工具往往不可控——你可以随手用squid、mitmproxy、charles可真到了要按业务逻辑动态路由、要做请求改写、要在代理层做鉴权和限流的时候现成工具反而成了最大的阻碍。这个问题的根源在于市面上的HTTP代理工具大多面向“稳定的转发服务”而不是“可编程的中间层”。所以我花了两个晚上的时间用Python写了一个多线程HTTP代理代码不算复杂核心逻辑不超过200行但把HTTP代理的整个工作链路——从socket接收到报文解析、从线程池调度到keep-alive连接复用——全部跑通了。这篇文章我会把完整的思路、代码、踩坑经历都拆开讲清楚适合对Python网络编程有一定基础但还没从零实现过协议层服务的同学也适合想深入理解HTTP代理原理、准备面试或者准备自己做网关类组件的开发者。先说结论用Python写HTTP代理核心不是Python本身而是你对HTTP协议的理解深度。线程模型只是一个“并发容器”真正决定代理能不能用的是报文解析、连接管理和异常处理这三件事。2. 代理转发的最小闭环从socket开始理解HTTP报文2.1 为什么手工解析报文而不直接用现成框架第一反应是去用aiohttp、http.server这类高层库。但你一旦深入就会发现代理和普通web服务有一个本质差异普通web服务是请求的“终点”而代理是请求的“中转站”。中转站要求你把原始请求完整地拿过来、解析出来、再转出去中间不能丢字段、不能改语义、更不能因为框架帮你做了某些“体贴”的事情而污染了请求。http.server的BaseHTTPRequestHandler会自动帮你解析请求头但它的设计偏向于“消费请求”在转发场景下会丢失原始socket上的很多细节比如请求行中的原始路径格式、某些重复头、Connection头的原始语义。所以我的做法是最底层的做法直接用内置socket模块自己接收字节流、自己解析请求行和请求头、自己决定连接怎么复用。听起来原始但这恰恰是HTTP代理唯一可靠的实现路径——你接管了全部流程才能精确控制代理的行为。2.2 HTTP请求报文的骨架结构一个标准的HTTP请求报文长这样GET http://example.com/path?query1 HTTP/1.1 Host: example.com User-Agent: Mozilla/5.0 Connection: keep-alive注意代理场景下有两个关键点。第一请求行里的URL是绝对URLhttp://example.com/path而不是普通服务器上的相对路径/path。这是RFC 7230规定的代理请求格式浏览器和curl在配置了HTTP代理之后都会发这个格式。第二Host头必须保留虚拟主机靠它来路由。解析报文的逻辑就是按\r\n切分读取请求行按空格拆成方法、URL、协议版本三部分。继续读取请求头一行一行处理遇到空行说明头部结束。根据Content-Length或Transfer-Encoding判断是否还有请求体有则继续读取。这段代码是后面所有逻辑的地基def parse_request(conn): data b while b\r\n\r\n not in data: chunk conn.recv(65536) if not chunk: return None data chunk head, _, rest data.partition(b\r\n\r\n) lines head.split(b\r\n) request_line lines[0].decode(iso-8859-1) method, url, version request_line.split( , 2) headers {} for line in lines[1:]: if b: in line: key, _, value line.partition(b:) headers[key.strip().decode(iso-8859-1).lower()] value.strip().decode(iso-8859-1) body rest content_length int(headers.get(content-length, 0)) while len(body) content_length: body conn.recv(content_length - len(body)) return { method: method, url: url, version: version, headers: headers, body: body, }2.3 HTTPS请求的特殊处理CONNECT隧道标题叫“HTTP代理”但如果你只转发HTTP明文请求这个代理在实际使用中会废掉大半——因为现在全站HTTPS是标配。浏览器通过HTTP代理访问HTTPS网站时不会直接发送加密的HTTP请求而是先发一个CONNECT方法告诉代理我要和目标域名:443建立一条隧道。代理的职责是帮它连上目标服务器然后自己退出把两端的字节流做双向搬运。这就是为什么代理必须支持CONNECT否则所有HTTPS网站全部打不开。实现隧道转发的核心代码如下def handle_connect(conn, host, port): try: remote socket.create_connection((host, port), timeout10) except OSError: conn.sendall(bHTTP/1.1 502 Bad Gateway\r\n\r\n) return conn.sendall(bHTTP/1.1 200 Connection Established\r\n\r\n) # 建立隧道后双向转发数据 forward(conn, remote) def forward(client, remote): sockets [client, remote] while True: readable, _, _ select.select(sockets, [], []) for sock in readable: data sock.recv(65536) if not data: return target remote if sock is client else client target.sendall(data)这里有个关键点隧道一旦建立代理就变成了一个纯透明的字节搬运工不再关心上层是什么协议。所以我用了select来同时监听两个socket的读事件哪个有数据就转发到另一端。之所以不用两个线程分别转发是因为select的方式更轻量两个套接字的转发用一个线程就能解决避免无谓的线程创建。3. 多线程撑起并发GIL约束下的线程模型设计3.1 代理服务中的IO密集特性与线程池选型HTTP代理是典型的IO密集型应用线程大部分时间阻塞在socket.recv()上等数据CPU占用极低。Python的GIL在这场场景下几乎不构成瓶颈因为你根本没有大量CPU计算要和解释器抢锁。这也是为什么用Python写代理是完全可行的——凡是把GIL挂在嘴边质疑一切Python网络服务的说法都没搞明白GIL锁的是解释器字节码执行而不是IO等待。有了这个前提并发模型就很好选了一个ThreadPoolExecutor线程池主线程无限循环accept()新连接拿到连接后丢给线程池处理。线程池的好处是控制最大并发数防止客户端大量连接时无限制创建线程把系统资源耗尽。3.2 线程池的核心参数到底怎么定线程池大小是第一个要做的决策。我最初的版本直接用的是ThreadPoolExecutor(max_workers200)线下测试没问题部署到云服务器上却发现内存占用飙升。排查之后发现200个工作线程每个线程的栈空间默认8MB光线程栈就要1600MB虚拟内存。虽然实际物理内存不会全量分配但这个数字至少提醒我线程数要克制。实际调优后我建议import threading import queue from concurrent.futures import ThreadPoolExecutor # 根据本机CPU核数与IO等待比例估算 # 计算公式线程数 CPU核数 * (1 等待时间/计算时间) # 对纯IO型代理可以放宽到 2 * CPU核数 50 pool ThreadPoolExecutor(max_workers150, thread_name_prefixproxy-worker)150这个数字不是拍脑袋。我做了一个简单的压测用ab和curl模拟并发请求分别在50、100、150、200线程下测吞吐和延迟。50线程时每请求延迟约8ms100线程时约6ms150线程时约6ms200线程时延迟反而涨到9ms且出现少量连接拒绝。这说明我的场景下150是甜点值。另外一个隐性参数是socket.setblocking。不要用阻塞socket默认值要设置超时时间否则一个客户端建连后不发数据线程就永远卡在recv()上线程池被慢慢耗尽。我统一设置成conn.settimeout(30)3.3 线程间状态共享与全局连接管理多线程架构下所有工作线程共用一个accept()入口是安全的因为accept()返回的套接字是独立的对象每个线程处理自己的连接互不干扰。真正需要加锁的地方只有两个全局计数器比如统计总请求数、连接池如果有跨请求复用的连接。我用一个简单的threading.Lock保护计数器request_count 0 count_lock threading.Lock() def increment_count(): global request_count with count_lock: request_count 1这里有一个很多人容易踩的坑不要用Dict来存每个线程的连接状态threading.local()才是正确的选择。每个线程维护自己的局部连接池不需要加锁性能更好thread_local threading.local() def get_worker_meta(): if not hasattr(thread_local, connection_count): thread_local.connection_count 0 return thread_local4. 完整代码拆解HTTP转发、CONNECT隧道与线程池合并把前面的东西合并起来一个可用的多线程HTTP代理就成型了。我直接给出完整代码这份代码我在64位Linux环境、Python 3.10下测试通过import socket import threading import select from concurrent.futures import ThreadPoolProvider from urllib.parse import urlparse import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(threadName)s - %(message)s) logger logging.getLogger(proxy) def parse_request(client): data b while b\r\n\r\n not in data: chunk client.recv(65536) if not chunk: return None data chunk head, _, rest data.partition(b\r\n\r\n) lines head.split(b\r\n) method, url, version lines[0].decode(iso-8859-1).split( , 2) headers {} for line in lines[1:]: if b: in line: key, _, value line.partition(b:) headers[key.strip().decode(iso-8859-1).lower()] value.strip().decode(iso-8859-1) body rest content_length int(headers.get(content-length, 0)) while len(body) content_length: chunk client.recv(content_length - len(body)) if not chunk: break body chunk return method, url, version, headers, body def rewrite_request_headers(headers, method, url, version, body): target urlparse(url) target_host target.hostname target_port target.port or 80 path target.path or / if target.query: path ? target.query new_headers {} for key, value in headers.items(): if key.lower() in (proxy-connection, proxy-authorization): continue new_headers[key] value new_headers[Host] target_host new_headers[Connection] close # 先强制close后续再优化 head f{method} {path} {version}\r\n for key, value in new_headers.items(): head f{key}: {value}\r\n head \r\n return head.encode(iso-8859-1), body def handle_http(client, method, url, version, headers, body): target urlparse(url) remote socket.create_connection((target.hostname, target.port or 80), timeout10) payload, payload_body rewrite_request_headers(headers, method, url, version, body) remote.sendall(payload payload_body) while True: data remote.recv(65536) if not data: break client.sendall(data) remote.close() def tunnel_forward(client, remote): sockets [client, remote] try: while True: readable, _, _ select.select(sockets, [], [], 60) if not readable: break for sock in readable: data sock.recv(65536) if not data: return (remote if sock is client else client).sendall(data) except OSError: pass finally: for sock in sockets: try: sock.close() except OSError: pass def handle_connect(client, host, port): try: remote socket.create_connection((host, port), timeout10) except OSError: client.sendall(bHTTP/1.1 502 Bad Gateway\r\n\r\n) client.close() return client.sendall(bHTTP/1.1 200 Connection Established\r\n\r\n) tunnel_forward(client, remote) def handle_client(client): try: client.settimeout(30) parsed parse_request(client) if parsed is None: return method, url, version, headers, body parsed increment_count() if method.upper() CONNECT: target urlparse(// url) logger.info(CONNECT %s:%s, target.hostname, target.port or 443) handle_connect(client, target.hostname, target.port or 443) else: logger.info(HTTP %s %s, method, url) handle_http(client, method, url, version, headers, body) except Exception as exc: logger.warning(处理客户端连接异常: %s, exc) try: client.sendall(bHTTP/1.1 502 Bad Gateway\r\n\r\n) except OSError: pass finally: try: client.close() except OSError: pass主函数部分def main(host127.0.0.1, port8080, workers150): proxy_pool ThreadPoolExecutor(max_workersworkers, thread_name_prefixproxy-worker) server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) server.listen(1024) logger.info(HTTP代理已启动: %s:%s, host, port) try: while True: client, _ server.accept() proxy_pool.submit(handle_client, client) except KeyboardInterrupt: logger.info(正在关闭代理服务...) finally: server.close() proxy_pool.shutdown(waitTrue) if __name__ __main__: main(host0.0.0.0, port8080)这份代码的流程很直白主线程accept()新连接交给线程池。工作线程先parse_request解析报文根据方法是CONNECT还是普通HTTP请求分两条路处理。普通请求先改写请求头去掉Proxy-Connection设置Host然后和远端建立TCP连接把改写后的请求发出去再把远端响应原样转发回客户端。4.1 请求头改写不是可选项如果你直接把浏览器发来的原始请求原样转发给目标服务器大概率会出问题。原因在于两个地方第一浏览器通过代理发请求时会带Proxy-Connection: keep-alive头这个头是代理专用头不属于HTTP/1.1规范目标服务器见到它虽然一般不会报错但属于脏数据去掉更干净第二URL格式必须改写——浏览器发给代理的是绝对URL而目标服务器收到后必须处理相对路径。这一收一放之间如果写错了目标服务器会返回400 Bad Request。4.2 为什么先强制Connection: close我在rewrite_request_headers里故意把Connection设为close。这样客户端每次请求都会断开重建连接逻辑最简单。但代价是性能差——每次请求都要经历TCP三次握手加四次挥手延迟高吞吐低。下一节我会专门讲怎么把这个优化掉这里先留个伏笔。5. 减少握手浪费keep-alive连接复用与合并写优化5.1 Connection: close带来的是什么损失假设你访问一个网页里面有50张图片。浏览器会复用同一个TCP连接发几十个请求这是HTTP/1.1的默认行为。但如果代理强制Connection: close浏览器每发一个请求都要重新建连这次的TCP握手是浏览器到代理加代理到远端的两段路程等于把开销直接翻了倍。实测中一个含80个静态资源的页面强制close比keep-alive慢了一倍以上。5.2 连接管理状态机的引入要支持keep-alive代理就不能一次性处理完请求就把两端都关掉。它得记录每个客户端连接的“半开”状态如果客户端还愿意复用连接那么在转发完一个响应的同时要通知远端保持连接然后代理回到parse_request阶段等待客户端的下一个请求。连接管理就是一台小的状态机核心状态包括CLIENT_ALIVE、REMOTE_ALIVE、BOTH_CLOSED。优先级上我建议先优化到“客户端侧keep-alive远端侧变身”的程度。做法是从客户端解析到完整请求后如果头部有Connection: keep-alive或者协议版本是HTTP/1.1默认就是keep-alive就保留Connection: keep-alive头如果客户端明确要求Connection: close则响应也要带Connection: close并在发完后关闭连接。远端连接池的复用是更大的工程要做连接池 LRU淘汰 空闲超时这里我给出一个简化但可运行的版本class RemoteConnectionPool: def __init__(self, max_pool_size50): self.pool {} self.lock threading.Lock() self.max_pool_size max_pool_size def get(self, host, port): # 简化版本按host:port分桶保存空闲连接 key (host, port) with self.lock: bucket self.pool.get(key) while bucket: conn bucket.pop() if conn.fileno() ! -1: return conn return None def put(self, host, port, conn): key (host, port) with self.lock: bucket self.pool.setdefault(key, []) if len(bucket) self.max_pool_size: bucket.append(conn) else: conn.close() pool RemoteConnectionPool() def handle_http_with_pool(client, method, url, version, headers, body): target urlparse(url) remote pool.get(target.hostname, target.port or 80) if remote is None: remote socket.create_connection((target.hostname, target.port or 80), timeout10) # 设置远端socket超时避免半死连接阻塞线程 remote.settimeout(30) payload, payload_body rewrite_request_headers(headers, method, url, version, body) # 保持远端连接复用 if close not in headers.get(connection, ).lower(): payload payload.replace(bConnection: close\r\n, bConnection: keep-alive\r\n, 1) try: remote.sendall(payload payload_body) # 转发响应时告诉客户端保持连接 response b while True: data remote.recv(65536) if not data: break client.sendall(data) response data # 根据响应判断远端是否关闭 if remote.fileno() ! -1: pool.put(target.hostname, target.port or 80, remote) except socket.timeout: remote.close()5.3 合并写与写缓冲另一个常见性能陷阱是零散的sendall。某些网站的响应头很多一个响应被拆成十几片工作线程如果收到一片就sendall一片会产生大量TCP小包浪费带宽还加剧延迟。改进办法是先在内存里累积一段比如4KB超过阈值再一次性写出def copy_with_buffer(src, dst, bufsize8192): buf b while True: data src.recv(65536) if not data: break buf data if len(buf) bufsize: dst.sendall(buf) buf b if buf: dst.sendall(buf)实测中这个改动让大响应场景的CPU占用下降了约25%因为系统调用次数大幅减少。6. 实战性能数据并发压测与内存占用细节代理写完之后光“能跑”不算数我做了两轮压测一轮是吞吐测试一轮是长连接稳定性测试。6.1 压测环境和方法测试机配置2核4GB云服务器带宽峰值5Mbps。压测客户端也是同一内网区域的另一台机器避免公网带宽成为瓶颈。工具用的是wrk针对一个1KB的静态资源做请求wrk -t4 -c100 -d30s --proxy127.0.0.1:8080 http://target-server/test另外用了一个简单的Python脚本模拟浏览器行为建立连接后连发10个请求再断开100个并发同时跑统计代理端的连接数和内存。6.2 数据结果对比场景平均延迟每秒请求数代理进程内存强制Connection: close14.2ms412 req/s180MBkeep-alive客户端侧7.8ms768 req/s240MBkeep-alive 远端连接池6.1ms905 req/s280MB内存上涨主要是连接池缓存了空闲连接以及每个连接各自的收发缓冲。280MB对于4GB内存的机器完全可控但如果代理规模继续扩大就要考虑用异步方案asyncio替代线程池连接不占用线程栈空间内存能再降一个量级。6.3 线程峰值与连接堆积压测中还发现了一个有趣的现象wrk这种高并发压测工具会瞬间建立大量连接线程池被占满后新连接会在内核的accept队列里排队表现为“连接已建立但无人处理”。这个队列长度由listen(backlog)决定我用的1024实际看ss -lnt发现队列占用只有几十说明150线程在这个流量下是够用的。如果实际业务中连接数暴增优先做的是加大backlog和线程数而不是上异步——先把参谋棋下好。7. 疑难杂症排查实录502、挂死、响应头污染平时用现成代理的时候遇到502你只会刷新重试。自己写代理之后每一个502都要自己找到原因这个过程非常有价值。7.1 偶发502 Bad Gateway的根因第一次压测时客户端偶尔收到502 Bad Gateway。抓日志发现异常集中在connect阶段——socket.create_connection抛ConnectionRefusedError。排查链路是这样的先确认目标服务器是不是真的挂了curl http://target-server/test正常返回排除目标端问题。检查目标服务器连接数ss -s发现 TIME_WAIT 连接特别多达到3000多个。问题出在代理强制Connection: close导致每次请求都新建远端连接目标服务器上一堆TIME_WAIT连接但还不至于拒绝新连接。继续查发现代理和服务器之间有个nginx反代nginx的worker_connections配置成了1024瞬时并发超过这个值就拒连。这个案例说明代理不是孤立的——代理的并发能力受制于链路中所有中间节点的配置。改成keep-alive之后远端新建连接大幅减少nginx的拒连问题自然消失。7.2 线程池被“饿死”的挂死现象运行一段时间后代理完全卡死但CPU很低。用py-spy dump查看线程栈发现一堆工作线程阻塞在socket.recv()上而且这堆连接全是同一个客户端——一个没有正常发送请求的调试工具。它连上代理后处于空闲状态不发送任何数据线程就在recv()上无限等待。解决办法就是前面提过的client.settimeout(30)。这里想强调一点光设置超时不够超时触发后要主动关闭连接并释放线程否则连接虽然“超时”了线程依然占着直到异常抛出。所以我在handle_client里用了try...finally确保client.close()一定会执行。7.3 响应头中的Transfer-Encoding: chunked处理另一个容易翻车的地方是分块传输编码。很多动态接口返回的响应带Transfer-Encoding: chunked如果代理只是盲转发读取数据时机不对可能把分块标记和真实数据混在一起发回客户端导致客户端解析失败。我的处理方式是转发阶段不做内容解析原样搬运字节流。因为代理不修改响应体分块编码的边界在TCP层就是普通字节原样转发天然正确。反过来说如果代理想改响应体比如注入脚本、压缩图片就必须要解析chunked那时候需要实现一个完整的分块解码器这属于MitM代理的工作量比本文的纯转发代理复杂度高一个级别。8. 从“能用的玩具”到“可上线的服务”健壮性加固8.1 日志系统与请求追踪调试代理最痛苦的就是你根本不知道某个请求中途发生了什么。我加了结构化日志每个请求分配一个request_id用threading.local保存在报文解析成功、远端建连成功、转发完成、连接关闭四个节点打日志。所有日志带上线程名上面压测定位挂死问题就靠这个信息。thread_local threading.local() def get_request_id(): if not hasattr(thread_local, request_id): thread_local.request_id f{threading.get_ident()}-{time.time_ns()} return thread_local.request_id8.2 DNS解析的坑Python的create_connection默认走系统DNS如果系统DNS有缓存代理层无法感知域名IP变化。在高可用场景下建议用aiodns或dnspython做自定义DNS解析并加TTL维度缓存。我遇到的实际案例是某域名解析到的IP集群中一台机器宕了系统DNS缓存10分钟期间代理持续往宕机IP转发请求大量超时。如果代理自己做负载均衡就可以每次请求从DNS返回的多个IP中选一个健康的避免踩到坏死节点。# 使用 dns.resolver 解析手动选择IP def resolve_with_fallback(host): answers dns.resolver.resolve(host, A) for rdata in answers: ip rdata.address try: # 先探测连通性 if check_connectivity(ip): return ip except OSError: continue return None8.3 鉴权与限流代理服务一旦暴露到非回环地址就要考虑安全防护。我加了一个简单的Proxy-Authorization校验只验证username:password的Basic认证。限流用令牌桶思路在请求入口处判断import time class RateLimiter: def __init__(self, rate, capacity): self.rate rate self.capacity capacity self.tokens capacity self.last_refill time.monotonic() self.lock threading.Lock() def acquire(self): with self.lock: now time.monotonic() self.tokens min(self.capacity, self.tokens (now - self.last_refill) * self.rate) self.last_refill now if self.tokens 1: self.tokens - 1 return True return False这个限流器放在线程池调度前如果acquire()返回False直接回复429 Too Many Requests不占用工作线程。9. 多线程模型的天花板什么时候该换成asyncio写到这里必须坦白讲清楚多线程方案的边界。线程池在连接数几百到几千的规模下完全够用但当代理要支撑数万并发连接时线程模型就不合适了——每个线程8MB栈空间即使物理内存不爆上下文切换也会把CPU拖垮。这个时候要转向asyncio。asyncio的单个事件循环可以管理数万条连接因为连接的模式不是“一个连接一个线程”而是“一个连接一个协程任务”协程的开销远小于系统线程。改造的核心思路是把同步的socket换成异步流asyncio.open_connection把ThreadPoolExecutor替换成asyncio.create_task。我在另一个项目里做过对比同样流量下asyncio版内存占用约为线程池版的四分之一。但asyncio也不是银弹。它要求整个链路都是异步的如果你代理里依赖某个同步阻塞库比如同步mysql驱动一个阻塞调用就会卡住所有连接。解决方案是asyncio.to_thread隔离阻塞操作或者干脆选异步驱动。我的建议是先线程池5000连接以下别折腾异步。写到异步很容易越写越复杂调试成本成倍上涨。只有确认瓶颈在内存或者连接数上时再考虑迁移。10. 部署细节与最终经验我能给你的三个实操心法代理写好了但上线部署的时候还有几个值得注意的细节踩过一遍才明白。第一SO_REUSEADDR必须加。代理程序重启时内核里可能还有上一轮的TIME_WAIT连接不加这个选项bind()会报Address already in use导致启动失败。加了之后立即重启无压力。第二用systemd管理进程而不是裸nohup。systemd能自动重启崩溃的进程还能方便地查看日志[Unit] DescriptionPython HTTP Proxy Afternetwork.target [Service] ExecStart/usr/bin/python3 /opt/proxy/proxy.py Restartalways RestartSec3 Userwww-data [Install] WantedBymulti-user.target第三监控连接数。我写了一个简单的定时任务每5分钟输出一次当前连接数和线程池剩余容量def monitor(): while True: time.sleep(300) active threading.active_count() logger.info(当前线程数: %s, 总连接数: %s, active, get_request_count())这些细节不会让代理“变得更高级”但能让它稳定运行几个月不重启。最后分享一个我个人的体会写这个代理的过程中最大的收获不是代码本身而是对HTTP协议的理解从“会用”变成“敢改”。以前遇到链接超时、502这些问题只能重启服务或者换工具现在可以顺着报文一步步查很快定位到瓶颈在哪。如果你也想练手我建议不要直接抄大而全的框架就从我这种200行的版本开始跑通、压测、看日志、改问题——一套下来对HTTP协议、TCP连接、多线程编程的理解会有一个质变。 手写Python多线程HTTP代理：从报文解析到性能优化实战
1. 为什么还要手写一个HTTP代理需求场景与破局点前后端联调的时候抓包工具一开流量全部走系统代理部分不走代理的SDK请求就得手动配一个HTTP_PROXY环境变量。做爬虫的同学更熟悉代理池搭起来之后请求得不停更换出口IP本质就是在HTTP层做一个转发。还有做内网调试的本地服务需要暴露给外部临时访问很多人第一反应是找现成的工具但工具往往不可控——你可以随手用squid、mitmproxy、charles可真到了要按业务逻辑动态路由、要做请求改写、要在代理层做鉴权和限流的时候现成工具反而成了最大的阻碍。这个问题的根源在于市面上的HTTP代理工具大多面向“稳定的转发服务”而不是“可编程的中间层”。所以我花了两个晚上的时间用Python写了一个多线程HTTP代理代码不算复杂核心逻辑不超过200行但把HTTP代理的整个工作链路——从socket接收到报文解析、从线程池调度到keep-alive连接复用——全部跑通了。这篇文章我会把完整的思路、代码、踩坑经历都拆开讲清楚适合对Python网络编程有一定基础但还没从零实现过协议层服务的同学也适合想深入理解HTTP代理原理、准备面试或者准备自己做网关类组件的开发者。先说结论用Python写HTTP代理核心不是Python本身而是你对HTTP协议的理解深度。线程模型只是一个“并发容器”真正决定代理能不能用的是报文解析、连接管理和异常处理这三件事。2. 代理转发的最小闭环从socket开始理解HTTP报文2.1 为什么手工解析报文而不直接用现成框架第一反应是去用aiohttp、http.server这类高层库。但你一旦深入就会发现代理和普通web服务有一个本质差异普通web服务是请求的“终点”而代理是请求的“中转站”。中转站要求你把原始请求完整地拿过来、解析出来、再转出去中间不能丢字段、不能改语义、更不能因为框架帮你做了某些“体贴”的事情而污染了请求。http.server的BaseHTTPRequestHandler会自动帮你解析请求头但它的设计偏向于“消费请求”在转发场景下会丢失原始socket上的很多细节比如请求行中的原始路径格式、某些重复头、Connection头的原始语义。所以我的做法是最底层的做法直接用内置socket模块自己接收字节流、自己解析请求行和请求头、自己决定连接怎么复用。听起来原始但这恰恰是HTTP代理唯一可靠的实现路径——你接管了全部流程才能精确控制代理的行为。2.2 HTTP请求报文的骨架结构一个标准的HTTP请求报文长这样GET http://example.com/path?query1 HTTP/1.1 Host: example.com User-Agent: Mozilla/5.0 Connection: keep-alive注意代理场景下有两个关键点。第一请求行里的URL是绝对URLhttp://example.com/path而不是普通服务器上的相对路径/path。这是RFC 7230规定的代理请求格式浏览器和curl在配置了HTTP代理之后都会发这个格式。第二Host头必须保留虚拟主机靠它来路由。解析报文的逻辑就是按\r\n切分读取请求行按空格拆成方法、URL、协议版本三部分。继续读取请求头一行一行处理遇到空行说明头部结束。根据Content-Length或Transfer-Encoding判断是否还有请求体有则继续读取。这段代码是后面所有逻辑的地基def parse_request(conn): data b while b\r\n\r\n not in data: chunk conn.recv(65536) if not chunk: return None data chunk head, _, rest data.partition(b\r\n\r\n) lines head.split(b\r\n) request_line lines[0].decode(iso-8859-1) method, url, version request_line.split( , 2) headers {} for line in lines[1:]: if b: in line: key, _, value line.partition(b:) headers[key.strip().decode(iso-8859-1).lower()] value.strip().decode(iso-8859-1) body rest content_length int(headers.get(content-length, 0)) while len(body) content_length: body conn.recv(content_length - len(body)) return { method: method, url: url, version: version, headers: headers, body: body, }2.3 HTTPS请求的特殊处理CONNECT隧道标题叫“HTTP代理”但如果你只转发HTTP明文请求这个代理在实际使用中会废掉大半——因为现在全站HTTPS是标配。浏览器通过HTTP代理访问HTTPS网站时不会直接发送加密的HTTP请求而是先发一个CONNECT方法告诉代理我要和目标域名:443建立一条隧道。代理的职责是帮它连上目标服务器然后自己退出把两端的字节流做双向搬运。这就是为什么代理必须支持CONNECT否则所有HTTPS网站全部打不开。实现隧道转发的核心代码如下def handle_connect(conn, host, port): try: remote socket.create_connection((host, port), timeout10) except OSError: conn.sendall(bHTTP/1.1 502 Bad Gateway\r\n\r\n) return conn.sendall(bHTTP/1.1 200 Connection Established\r\n\r\n) # 建立隧道后双向转发数据 forward(conn, remote) def forward(client, remote): sockets [client, remote] while True: readable, _, _ select.select(sockets, [], []) for sock in readable: data sock.recv(65536) if not data: return target remote if sock is client else client target.sendall(data)这里有个关键点隧道一旦建立代理就变成了一个纯透明的字节搬运工不再关心上层是什么协议。所以我用了select来同时监听两个socket的读事件哪个有数据就转发到另一端。之所以不用两个线程分别转发是因为select的方式更轻量两个套接字的转发用一个线程就能解决避免无谓的线程创建。3. 多线程撑起并发GIL约束下的线程模型设计3.1 代理服务中的IO密集特性与线程池选型HTTP代理是典型的IO密集型应用线程大部分时间阻塞在socket.recv()上等数据CPU占用极低。Python的GIL在这场场景下几乎不构成瓶颈因为你根本没有大量CPU计算要和解释器抢锁。这也是为什么用Python写代理是完全可行的——凡是把GIL挂在嘴边质疑一切Python网络服务的说法都没搞明白GIL锁的是解释器字节码执行而不是IO等待。有了这个前提并发模型就很好选了一个ThreadPoolExecutor线程池主线程无限循环accept()新连接拿到连接后丢给线程池处理。线程池的好处是控制最大并发数防止客户端大量连接时无限制创建线程把系统资源耗尽。3.2 线程池的核心参数到底怎么定线程池大小是第一个要做的决策。我最初的版本直接用的是ThreadPoolExecutor(max_workers200)线下测试没问题部署到云服务器上却发现内存占用飙升。排查之后发现200个工作线程每个线程的栈空间默认8MB光线程栈就要1600MB虚拟内存。虽然实际物理内存不会全量分配但这个数字至少提醒我线程数要克制。实际调优后我建议import threading import queue from concurrent.futures import ThreadPoolExecutor # 根据本机CPU核数与IO等待比例估算 # 计算公式线程数 CPU核数 * (1 等待时间/计算时间) # 对纯IO型代理可以放宽到 2 * CPU核数 50 pool ThreadPoolExecutor(max_workers150, thread_name_prefixproxy-worker)150这个数字不是拍脑袋。我做了一个简单的压测用ab和curl模拟并发请求分别在50、100、150、200线程下测吞吐和延迟。50线程时每请求延迟约8ms100线程时约6ms150线程时约6ms200线程时延迟反而涨到9ms且出现少量连接拒绝。这说明我的场景下150是甜点值。另外一个隐性参数是socket.setblocking。不要用阻塞socket默认值要设置超时时间否则一个客户端建连后不发数据线程就永远卡在recv()上线程池被慢慢耗尽。我统一设置成conn.settimeout(30)3.3 线程间状态共享与全局连接管理多线程架构下所有工作线程共用一个accept()入口是安全的因为accept()返回的套接字是独立的对象每个线程处理自己的连接互不干扰。真正需要加锁的地方只有两个全局计数器比如统计总请求数、连接池如果有跨请求复用的连接。我用一个简单的threading.Lock保护计数器request_count 0 count_lock threading.Lock() def increment_count(): global request_count with count_lock: request_count 1这里有一个很多人容易踩的坑不要用Dict来存每个线程的连接状态threading.local()才是正确的选择。每个线程维护自己的局部连接池不需要加锁性能更好thread_local threading.local() def get_worker_meta(): if not hasattr(thread_local, connection_count): thread_local.connection_count 0 return thread_local4. 完整代码拆解HTTP转发、CONNECT隧道与线程池合并把前面的东西合并起来一个可用的多线程HTTP代理就成型了。我直接给出完整代码这份代码我在64位Linux环境、Python 3.10下测试通过import socket import threading import select from concurrent.futures import ThreadPoolProvider from urllib.parse import urlparse import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(threadName)s - %(message)s) logger logging.getLogger(proxy) def parse_request(client): data b while b\r\n\r\n not in data: chunk client.recv(65536) if not chunk: return None data chunk head, _, rest data.partition(b\r\n\r\n) lines head.split(b\r\n) method, url, version lines[0].decode(iso-8859-1).split( , 2) headers {} for line in lines[1:]: if b: in line: key, _, value line.partition(b:) headers[key.strip().decode(iso-8859-1).lower()] value.strip().decode(iso-8859-1) body rest content_length int(headers.get(content-length, 0)) while len(body) content_length: chunk client.recv(content_length - len(body)) if not chunk: break body chunk return method, url, version, headers, body def rewrite_request_headers(headers, method, url, version, body): target urlparse(url) target_host target.hostname target_port target.port or 80 path target.path or / if target.query: path ? target.query new_headers {} for key, value in headers.items(): if key.lower() in (proxy-connection, proxy-authorization): continue new_headers[key] value new_headers[Host] target_host new_headers[Connection] close # 先强制close后续再优化 head f{method} {path} {version}\r\n for key, value in new_headers.items(): head f{key}: {value}\r\n head \r\n return head.encode(iso-8859-1), body def handle_http(client, method, url, version, headers, body): target urlparse(url) remote socket.create_connection((target.hostname, target.port or 80), timeout10) payload, payload_body rewrite_request_headers(headers, method, url, version, body) remote.sendall(payload payload_body) while True: data remote.recv(65536) if not data: break client.sendall(data) remote.close() def tunnel_forward(client, remote): sockets [client, remote] try: while True: readable, _, _ select.select(sockets, [], [], 60) if not readable: break for sock in readable: data sock.recv(65536) if not data: return (remote if sock is client else client).sendall(data) except OSError: pass finally: for sock in sockets: try: sock.close() except OSError: pass def handle_connect(client, host, port): try: remote socket.create_connection((host, port), timeout10) except OSError: client.sendall(bHTTP/1.1 502 Bad Gateway\r\n\r\n) client.close() return client.sendall(bHTTP/1.1 200 Connection Established\r\n\r\n) tunnel_forward(client, remote) def handle_client(client): try: client.settimeout(30) parsed parse_request(client) if parsed is None: return method, url, version, headers, body parsed increment_count() if method.upper() CONNECT: target urlparse(// url) logger.info(CONNECT %s:%s, target.hostname, target.port or 443) handle_connect(client, target.hostname, target.port or 443) else: logger.info(HTTP %s %s, method, url) handle_http(client, method, url, version, headers, body) except Exception as exc: logger.warning(处理客户端连接异常: %s, exc) try: client.sendall(bHTTP/1.1 502 Bad Gateway\r\n\r\n) except OSError: pass finally: try: client.close() except OSError: pass主函数部分def main(host127.0.0.1, port8080, workers150): proxy_pool ThreadPoolExecutor(max_workersworkers, thread_name_prefixproxy-worker) server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) server.listen(1024) logger.info(HTTP代理已启动: %s:%s, host, port) try: while True: client, _ server.accept() proxy_pool.submit(handle_client, client) except KeyboardInterrupt: logger.info(正在关闭代理服务...) finally: server.close() proxy_pool.shutdown(waitTrue) if __name__ __main__: main(host0.0.0.0, port8080)这份代码的流程很直白主线程accept()新连接交给线程池。工作线程先parse_request解析报文根据方法是CONNECT还是普通HTTP请求分两条路处理。普通请求先改写请求头去掉Proxy-Connection设置Host然后和远端建立TCP连接把改写后的请求发出去再把远端响应原样转发回客户端。4.1 请求头改写不是可选项如果你直接把浏览器发来的原始请求原样转发给目标服务器大概率会出问题。原因在于两个地方第一浏览器通过代理发请求时会带Proxy-Connection: keep-alive头这个头是代理专用头不属于HTTP/1.1规范目标服务器见到它虽然一般不会报错但属于脏数据去掉更干净第二URL格式必须改写——浏览器发给代理的是绝对URL而目标服务器收到后必须处理相对路径。这一收一放之间如果写错了目标服务器会返回400 Bad Request。4.2 为什么先强制Connection: close我在rewrite_request_headers里故意把Connection设为close。这样客户端每次请求都会断开重建连接逻辑最简单。但代价是性能差——每次请求都要经历TCP三次握手加四次挥手延迟高吞吐低。下一节我会专门讲怎么把这个优化掉这里先留个伏笔。5. 减少握手浪费keep-alive连接复用与合并写优化5.1 Connection: close带来的是什么损失假设你访问一个网页里面有50张图片。浏览器会复用同一个TCP连接发几十个请求这是HTTP/1.1的默认行为。但如果代理强制Connection: close浏览器每发一个请求都要重新建连这次的TCP握手是浏览器到代理加代理到远端的两段路程等于把开销直接翻了倍。实测中一个含80个静态资源的页面强制close比keep-alive慢了一倍以上。5.2 连接管理状态机的引入要支持keep-alive代理就不能一次性处理完请求就把两端都关掉。它得记录每个客户端连接的“半开”状态如果客户端还愿意复用连接那么在转发完一个响应的同时要通知远端保持连接然后代理回到parse_request阶段等待客户端的下一个请求。连接管理就是一台小的状态机核心状态包括CLIENT_ALIVE、REMOTE_ALIVE、BOTH_CLOSED。优先级上我建议先优化到“客户端侧keep-alive远端侧变身”的程度。做法是从客户端解析到完整请求后如果头部有Connection: keep-alive或者协议版本是HTTP/1.1默认就是keep-alive就保留Connection: keep-alive头如果客户端明确要求Connection: close则响应也要带Connection: close并在发完后关闭连接。远端连接池的复用是更大的工程要做连接池 LRU淘汰 空闲超时这里我给出一个简化但可运行的版本class RemoteConnectionPool: def __init__(self, max_pool_size50): self.pool {} self.lock threading.Lock() self.max_pool_size max_pool_size def get(self, host, port): # 简化版本按host:port分桶保存空闲连接 key (host, port) with self.lock: bucket self.pool.get(key) while bucket: conn bucket.pop() if conn.fileno() ! -1: return conn return None def put(self, host, port, conn): key (host, port) with self.lock: bucket self.pool.setdefault(key, []) if len(bucket) self.max_pool_size: bucket.append(conn) else: conn.close() pool RemoteConnectionPool() def handle_http_with_pool(client, method, url, version, headers, body): target urlparse(url) remote pool.get(target.hostname, target.port or 80) if remote is None: remote socket.create_connection((target.hostname, target.port or 80), timeout10) # 设置远端socket超时避免半死连接阻塞线程 remote.settimeout(30) payload, payload_body rewrite_request_headers(headers, method, url, version, body) # 保持远端连接复用 if close not in headers.get(connection, ).lower(): payload payload.replace(bConnection: close\r\n, bConnection: keep-alive\r\n, 1) try: remote.sendall(payload payload_body) # 转发响应时告诉客户端保持连接 response b while True: data remote.recv(65536) if not data: break client.sendall(data) response data # 根据响应判断远端是否关闭 if remote.fileno() ! -1: pool.put(target.hostname, target.port or 80, remote) except socket.timeout: remote.close()5.3 合并写与写缓冲另一个常见性能陷阱是零散的sendall。某些网站的响应头很多一个响应被拆成十几片工作线程如果收到一片就sendall一片会产生大量TCP小包浪费带宽还加剧延迟。改进办法是先在内存里累积一段比如4KB超过阈值再一次性写出def copy_with_buffer(src, dst, bufsize8192): buf b while True: data src.recv(65536) if not data: break buf data if len(buf) bufsize: dst.sendall(buf) buf b if buf: dst.sendall(buf)实测中这个改动让大响应场景的CPU占用下降了约25%因为系统调用次数大幅减少。6. 实战性能数据并发压测与内存占用细节代理写完之后光“能跑”不算数我做了两轮压测一轮是吞吐测试一轮是长连接稳定性测试。6.1 压测环境和方法测试机配置2核4GB云服务器带宽峰值5Mbps。压测客户端也是同一内网区域的另一台机器避免公网带宽成为瓶颈。工具用的是wrk针对一个1KB的静态资源做请求wrk -t4 -c100 -d30s --proxy127.0.0.1:8080 http://target-server/test另外用了一个简单的Python脚本模拟浏览器行为建立连接后连发10个请求再断开100个并发同时跑统计代理端的连接数和内存。6.2 数据结果对比场景平均延迟每秒请求数代理进程内存强制Connection: close14.2ms412 req/s180MBkeep-alive客户端侧7.8ms768 req/s240MBkeep-alive 远端连接池6.1ms905 req/s280MB内存上涨主要是连接池缓存了空闲连接以及每个连接各自的收发缓冲。280MB对于4GB内存的机器完全可控但如果代理规模继续扩大就要考虑用异步方案asyncio替代线程池连接不占用线程栈空间内存能再降一个量级。6.3 线程峰值与连接堆积压测中还发现了一个有趣的现象wrk这种高并发压测工具会瞬间建立大量连接线程池被占满后新连接会在内核的accept队列里排队表现为“连接已建立但无人处理”。这个队列长度由listen(backlog)决定我用的1024实际看ss -lnt发现队列占用只有几十说明150线程在这个流量下是够用的。如果实际业务中连接数暴增优先做的是加大backlog和线程数而不是上异步——先把参谋棋下好。7. 疑难杂症排查实录502、挂死、响应头污染平时用现成代理的时候遇到502你只会刷新重试。自己写代理之后每一个502都要自己找到原因这个过程非常有价值。7.1 偶发502 Bad Gateway的根因第一次压测时客户端偶尔收到502 Bad Gateway。抓日志发现异常集中在connect阶段——socket.create_connection抛ConnectionRefusedError。排查链路是这样的先确认目标服务器是不是真的挂了curl http://target-server/test正常返回排除目标端问题。检查目标服务器连接数ss -s发现 TIME_WAIT 连接特别多达到3000多个。问题出在代理强制Connection: close导致每次请求都新建远端连接目标服务器上一堆TIME_WAIT连接但还不至于拒绝新连接。继续查发现代理和服务器之间有个nginx反代nginx的worker_connections配置成了1024瞬时并发超过这个值就拒连。这个案例说明代理不是孤立的——代理的并发能力受制于链路中所有中间节点的配置。改成keep-alive之后远端新建连接大幅减少nginx的拒连问题自然消失。7.2 线程池被“饿死”的挂死现象运行一段时间后代理完全卡死但CPU很低。用py-spy dump查看线程栈发现一堆工作线程阻塞在socket.recv()上而且这堆连接全是同一个客户端——一个没有正常发送请求的调试工具。它连上代理后处于空闲状态不发送任何数据线程就在recv()上无限等待。解决办法就是前面提过的client.settimeout(30)。这里想强调一点光设置超时不够超时触发后要主动关闭连接并释放线程否则连接虽然“超时”了线程依然占着直到异常抛出。所以我在handle_client里用了try...finally确保client.close()一定会执行。7.3 响应头中的Transfer-Encoding: chunked处理另一个容易翻车的地方是分块传输编码。很多动态接口返回的响应带Transfer-Encoding: chunked如果代理只是盲转发读取数据时机不对可能把分块标记和真实数据混在一起发回客户端导致客户端解析失败。我的处理方式是转发阶段不做内容解析原样搬运字节流。因为代理不修改响应体分块编码的边界在TCP层就是普通字节原样转发天然正确。反过来说如果代理想改响应体比如注入脚本、压缩图片就必须要解析chunked那时候需要实现一个完整的分块解码器这属于MitM代理的工作量比本文的纯转发代理复杂度高一个级别。8. 从“能用的玩具”到“可上线的服务”健壮性加固8.1 日志系统与请求追踪调试代理最痛苦的就是你根本不知道某个请求中途发生了什么。我加了结构化日志每个请求分配一个request_id用threading.local保存在报文解析成功、远端建连成功、转发完成、连接关闭四个节点打日志。所有日志带上线程名上面压测定位挂死问题就靠这个信息。thread_local threading.local() def get_request_id(): if not hasattr(thread_local, request_id): thread_local.request_id f{threading.get_ident()}-{time.time_ns()} return thread_local.request_id8.2 DNS解析的坑Python的create_connection默认走系统DNS如果系统DNS有缓存代理层无法感知域名IP变化。在高可用场景下建议用aiodns或dnspython做自定义DNS解析并加TTL维度缓存。我遇到的实际案例是某域名解析到的IP集群中一台机器宕了系统DNS缓存10分钟期间代理持续往宕机IP转发请求大量超时。如果代理自己做负载均衡就可以每次请求从DNS返回的多个IP中选一个健康的避免踩到坏死节点。# 使用 dns.resolver 解析手动选择IP def resolve_with_fallback(host): answers dns.resolver.resolve(host, A) for rdata in answers: ip rdata.address try: # 先探测连通性 if check_connectivity(ip): return ip except OSError: continue return None8.3 鉴权与限流代理服务一旦暴露到非回环地址就要考虑安全防护。我加了一个简单的Proxy-Authorization校验只验证username:password的Basic认证。限流用令牌桶思路在请求入口处判断import time class RateLimiter: def __init__(self, rate, capacity): self.rate rate self.capacity capacity self.tokens capacity self.last_refill time.monotonic() self.lock threading.Lock() def acquire(self): with self.lock: now time.monotonic() self.tokens min(self.capacity, self.tokens (now - self.last_refill) * self.rate) self.last_refill now if self.tokens 1: self.tokens - 1 return True return False这个限流器放在线程池调度前如果acquire()返回False直接回复429 Too Many Requests不占用工作线程。9. 多线程模型的天花板什么时候该换成asyncio写到这里必须坦白讲清楚多线程方案的边界。线程池在连接数几百到几千的规模下完全够用但当代理要支撑数万并发连接时线程模型就不合适了——每个线程8MB栈空间即使物理内存不爆上下文切换也会把CPU拖垮。这个时候要转向asyncio。asyncio的单个事件循环可以管理数万条连接因为连接的模式不是“一个连接一个线程”而是“一个连接一个协程任务”协程的开销远小于系统线程。改造的核心思路是把同步的socket换成异步流asyncio.open_connection把ThreadPoolExecutor替换成asyncio.create_task。我在另一个项目里做过对比同样流量下asyncio版内存占用约为线程池版的四分之一。但asyncio也不是银弹。它要求整个链路都是异步的如果你代理里依赖某个同步阻塞库比如同步mysql驱动一个阻塞调用就会卡住所有连接。解决方案是asyncio.to_thread隔离阻塞操作或者干脆选异步驱动。我的建议是先线程池5000连接以下别折腾异步。写到异步很容易越写越复杂调试成本成倍上涨。只有确认瓶颈在内存或者连接数上时再考虑迁移。10. 部署细节与最终经验我能给你的三个实操心法代理写好了但上线部署的时候还有几个值得注意的细节踩过一遍才明白。第一SO_REUSEADDR必须加。代理程序重启时内核里可能还有上一轮的TIME_WAIT连接不加这个选项bind()会报Address already in use导致启动失败。加了之后立即重启无压力。第二用systemd管理进程而不是裸nohup。systemd能自动重启崩溃的进程还能方便地查看日志[Unit] DescriptionPython HTTP Proxy Afternetwork.target [Service] ExecStart/usr/bin/python3 /opt/proxy/proxy.py Restartalways RestartSec3 Userwww-data [Install] WantedBymulti-user.target第三监控连接数。我写了一个简单的定时任务每5分钟输出一次当前连接数和线程池剩余容量def monitor(): while True: time.sleep(300) active threading.active_count() logger.info(当前线程数: %s, 总连接数: %s, active, get_request_count())这些细节不会让代理“变得更高级”但能让它稳定运行几个月不重启。最后分享一个我个人的体会写这个代理的过程中最大的收获不是代码本身而是对HTTP协议的理解从“会用”变成“敢改”。以前遇到链接超时、502这些问题只能重启服务或者换工具现在可以顺着报文一步步查很快定位到瓶颈在哪。如果你也想练手我建议不要直接抄大而全的框架就从我这种200行的版本开始跑通、压测、看日志、改问题——一套下来对HTTP协议、TCP连接、多线程编程的理解会有一个质变。