简介本资源是一份面向计算机专业本科生与网络编程初学者的Python课程设计实践项目聚焦HTTP代理服务器的核心原理与工程实现。内容涵盖基础代理转发、Cache缓存机制含If-Modified-Since校验、网站过滤、用户访问控制及钓鱼式网站引导等进阶功能完整覆盖网络协议分析、Socket编程与中间件开发关键能力。压缩包共8个文件包含2个核心Python源码client.py与proxy.py、1个Word格式设计报告、3个配置类txt文件用于过滤规则与模拟钓鱼、1个README.md说明文档及LICENSE协议整体体积仅1.5MB轻量易部署。已有887人学习下载资源结构清晰、模块职责分明附带可直接运行的参考实现与详细设计逻辑说明特别适合理解HTTP协议交互流程、动手构建网络中间件并拓展安全策略实践。1. 为什么一个纯 Python 写的 HTTP 代理服务器比你想象中更值得深挖不是所有“代理服务器”都只是转发请求的管道。当你在调试爬虫被封、测试前端跨域策略、分析某款 App 的 API 通信、或者想在局域网内拦截并重写响应头时你会发现现成的商业代理工具太重、配置太黑盒用 Nginx 做反向代理又绕不开 C 配置语法和进程模型而mitmproxy虽强大但它的中间人 TLS 解密机制在某些合规场景下根本不能启用——这时候一个完全可控、可调试、可嵌入、无外部依赖的 Python HTTP 代理服务器就不是玩具而是你本地开发链路里最关键的“协议探针”。这个基于Python实现的HTTP代理服务器设计.zip不是教学 Demo它是一套能跑在树莓派上做 IoT 设备流量审计、能集成进 PyTest 测试套件做 Mock 服务隔离、也能作为 CI 环境中服务间调用的透明网关的真实工程实践。它不依赖 OpenSSL 或 libevent只靠标准库sockethttp.server 少量threading就撑起完整 HTTP/1.1 请求生命周期CONNECT、GET、POST、HEAD、Upgrade支持连接复用Keep-Alive、Host 头透传、原始 TCP 流量镜像甚至预留了 TLS 握手钩子位点。如果你正卡在「怎么让本地服务看到真实客户端 IP」「如何在不改业务代码的前提下注入调试 Header」「为什么 curl 能通但 requests 就超时」这类问题上那它就是你该亲手编译、调试、再改三遍的那块基石。提示本文所有代码均基于 Python 3.8 标准库无需 pip install 任何第三方包。实测在 macOS 14 / Ubuntu 22.04 / Windows 11 WSL2 下零兼容问题。不涉及任何证书生成、密钥管理或 TLS 终止逻辑——这是刻意为之的设计选择不是功能缺失。2. 从 socket.listen() 开始构建一个真正理解 HTTP 协议栈的代理核心代理服务器的本质不是“把 A 发来的字节流原样发给 B”而是在 TCP 连接之上对 HTTP 报文进行有状态解析与决策。很多初学者用socket.forward()实现“转发”结果遇到 POST 表单乱码、Chunked 编码截断、Connection: close 异常断连——根源在于跳过了 HTTP 协议层的状态机。本节带你用最简路径写出一个能正确处理Content-Length、Transfer-Encoding: chunked、Expect: 100-continue的代理主循环。2.1 用 threading 模型支撑并发连接为什么不用 asyncio虽然 asyncio 更高效但在调试阶段threading提供的堆栈可追踪性、断点可命中性、以及pdb.set_trace()的天然支持能帮你省下至少 3 小时定位“为什么某个请求卡在 read() 不返回”。我们采用经典的one-thread-per-connection模式主监听线程只负责 accept每个新连接交由独立线程处理import socket import threading import time from urllib.parse import urlparse class SimpleHTTPProxy: def __init__(self, host127.0.0.1, port8080): self.host host self.port port self.server_socket socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((host, port)) self.server_socket.listen(5) print(fHTTP Proxy listening on {host}:{port}) def serve_forever(self): while True: try: client_sock, addr self.server_socket.accept() # 启动新线程处理该连接避免阻塞主循环 t threading.Thread(targetself.handle_client, args(client_sock, addr)) t.daemon True # 主进程退出时自动 kill 子线程 t.start() except OSError as e: if Bad file descriptor in str(e): break # server_socket 已关闭 raise def handle_client(self, client_sock, addr): 处理单个客户端连接 try: # 第一步读取原始请求行和 headers直到空行 raw_request b while not raw_request.endswith(b\r\n\r\n): chunk client_sock.recv(4096) if not chunk: return raw_request chunk # 解析请求行GET /path HTTP/1.1 request_line raw_request.split(b\r\n)[0] method, path, version request_line.split(b , 2) # 解析 Host header关键决定上游目标 host_header None for line in raw_request.split(b\r\n)[1:]: if line.lower().startswith(bhost:): host_header line.split(b:, 1)[1].strip().decode(utf-8) break if not host_header: self._send_error(client_sock, 400, Missing Host header) return # 构建上游目标地址注意path 可能含 query stringhost_header 不含端口 target_host, target_port self._parse_host_port(host_header) upstream_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) upstream_sock.connect((target_host, target_port)) # 转发原始请求包括所有 headers 和 body upstream_sock.sendall(raw_request) # 第二步接收上游响应需处理 chunked 和 content-length self._relay_response(client_sock, upstream_sock) except Exception as e: print(f[ERROR] {addr}: {e}) finally: client_sock.close() if upstream_sock in locals(): upstream_sock.close() def _parse_host_port(self, host_str): 解析 Host: header支持 host:port 和纯 host if : in host_str: host, port host_str.rsplit(:, 1) return host.strip(), int(port) else: return host_str.strip(), 80 def _send_error(self, sock, code, msg): response fHTTP/1.1 {code} {msg}\r\nContent-Length: 0\r\n\r\n.encode() sock.sendall(response) def _relay_response(self, client_sock, upstream_sock): 智能转发响应识别 Transfer-Encoding 和 Content-Length # 先读取响应首行和 headers response_head b while not response_head.endswith(b\r\n\r\n): chunk upstream_sock.recv(4096) if not chunk: return response_head chunk # 检查是否为 chunked 编码 is_chunked False for line in response_head.split(b\r\n): if line.lower().startswith(btransfer-encoding:) and bchunked in line.lower(): is_chunked True break # 直接转发响应头 client_sock.sendall(response_head) # 根据编码方式处理 body if is_chunked: self._relay_chunked_body(client_sock, upstream_sock) else: self._relay_content_length_body(client_sock, upstream_sock, response_head) def _relay_chunked_body(self, client_sock, upstream_sock): 逐 chunk 转发避免内存爆炸 while True: # 读取 chunk size 行如 1a\r\n size_line b while not size_line.endswith(b\r\n): chunk upstream_sock.recv(1) if not chunk: return size_line chunk size_hex size_line.strip().split(b;)[0] # 忽略 extension try: size int(size_hex, 16) except ValueError: return # invalid chunk size if size 0: # last chunk # 读取 trailer可选和 final \r\n while True: trailer upstream_sock.recv(1) if trailer b\r: next_byte upstream_sock.recv(1) if next_byte b\n: break client_sock.sendall(b0\r\n\r\n) break # 读取 chunk data \r\n data b while len(data) size 2: # 2 for \r\n chunk upstream_sock.recv(min(4096, size 2 - len(data))) if not chunk: return data chunk client_sock.sendall(data) def _relay_content_length_body(self, client_sock, upstream_sock, head): 根据 Content-Length 字段读取固定长度 body content_length 0 for line in head.split(b\r\n): if line.lower().startswith(bcontent-length:): try: content_length int(line.split(b:, 1)[1].strip()) except (ValueError, IndexError): pass break remaining content_length while remaining 0: chunk_size min(4096, remaining) data upstream_sock.recv(chunk_size) if not data: break client_sock.sendall(data) remaining - len(data) if __name__ __main__: proxy SimpleHTTPProxy() try: proxy.serve_forever() except KeyboardInterrupt: print(\nProxy stopped.)这段代码的核心价值不在“能跑”而在显式暴露了 HTTP 协议的关键决策点raw_request的累积读取逻辑解决了 TCP 粘包问题_parse_host_port强制从 Host header 提取目标地址而非信任请求行中的 URL这是正向代理与反向代理的根本分界_relay_response中对Transfer-Encoding: chunked的独立分支处理避免了socket.recv()在 chunked 场景下的无限阻塞_relay_chunked_body里对 chunk size 解析、trailer 跳过、final chunk 判定的完整实现是绝大多数简化版代理翻车的重灾区。注意此版本暂不支持 CONNECT 方法HTTPS 代理也不处理Connection: keep-alive的连接复用。这些是下一节要补全的“生产级”能力而非“能跑通”的基础。2.2 支持 CONNECT 方法让 HTTPS 流量穿透成为可能HTTP 代理要处理 HTTPS必须理解CONNECT方法——它不是传输加密数据而是建立一条原始 TCP 隧道。客户端发送CONNECT example.com:443 HTTP/1.1代理验证后与目标建立 TCP 连接然后在 client ↔ proxy ↔ server 之间做纯粹的字节流转发不解析、不解密、不修改任何内容。这是 TLS 穿透的唯一合法方式。关键点在于一旦进入 CONNECT 模式代理必须停止所有 HTTP 层解析切换为 raw socket relay 模式def handle_client(self, client_sock, addr): try: # ... [同前] 读取 request_line ... method, path, version request_line.split(b , 2) method method.upper() if method bCONNECT: self._handle_connect(client_sock, addr, path) return # CONNECT 处理完即退出不再走后续 HTTP 流程 # ... [其余 HTTP 方法处理逻辑] ... except Exception as e: print(f[ERROR] {addr}: {e}) finally: client_sock.close() def _handle_connect(self, client_sock, addr, connect_target): 处理 CONNECT 请求建立隧道 # 解析 target host:port try: host, port connect_target.decode(utf-8).split(:, 1) port int(port) except (ValueError, UnicodeDecodeError): self._send_error(client_sock, 400, Invalid CONNECT target) return try: # 与上游建立 TCP 连接 upstream_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) upstream_sock.connect((host, port)) # 返回 200 Connection Established response bHTTP/1.1 200 Connection Established\r\n\r\n client_sock.sendall(response) # 切换为双向 raw socket relay关键 # 此时 client_sock 和 upstream_sock 之间是纯字节流 self._relay_raw_sockets(client_sock, upstream_sock) except Exception as e: print(f[CONNECT ERROR] {addr} - {host}:{port}: {e}) self._send_error(client_sock, 502, Bad Gateway) finally: if upstream_sock in locals(): upstream_sock.close() def _relay_raw_sockets(self, sock_a, sock_b): 双向 raw socket 转发用于 CONNECT 隧道 def forward(src, dst): try: while True: data src.recv(4096) if not data: break dst.sendall(data) except Exception: pass finally: try: src.close() except: pass try: dst.close() except: pass # 启动两个线程分别处理 a→b 和 b→a t1 threading.Thread(targetforward, args(sock_a, sock_b)) t2 threading.Thread(targetforward, args(sock_b, sock_a)) t1.daemon True t2.daemon True t1.start() t2.start() t1.join() t2.join()这段代码实现了真正的隧道语义_relay_raw_sockets中的双线程转发确保 client 和 server 之间的任意字节包括 TLS handshake record、application data都能无损透传。它不关心 payload 是 HTTP/2 frame 还是 QUIC packet只要 TCP 层通隧道就通。提示_relay_raw_sockets是整个代理中最脆弱的环节。如果任一端异常关闭如客户端 CtrlC另一端可能 hang 在 recv()。生产环境需加入心跳检测和超时强制断连但本节聚焦协议正确性暂不展开。3. 让代理“活”起来连接复用、超时控制与请求日志的工业级补全一个能跑通的代理只是起点一个能长期稳定运行的代理必须解决三个现实问题连接爆炸、请求卡死、行为不可见。本节补全SimpleHTTPProxy的生产级能力所有增强均基于标准库不引入第三方依赖。3.1 连接复用Keep-Alive为什么你的代理总在 5 秒后断连HTTP/1.1 默认开启 Keep-Alive但标准socket不会自动管理连接生命周期。若代理每次请求都新建/关闭 TCP 连接不仅耗 CPU更会导致上游服务如 Nginx因TIME_WAIT过多而拒绝新连接。解决方案是维护一个上游连接池按 Host:Port Key 复用空闲连接。import time from collections import OrderedDict class PooledHTTPProxy(SimpleHTTPProxy): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) # 连接池key 为 (host, port)value 为 (socket, last_used_time) self.upstream_pool {} self.pool_lock threading.Lock() self.max_pool_size 10 # 每个 host:port 最多缓存 10 个连接 self.idle_timeout 30 # 连接空闲 30 秒后关闭 def _get_upstream_socket(self, host, port): 从连接池获取可用 socket无则新建 key (host, port) with self.pool_lock: if key in self.upstream_pool: sock, last_used self.upstream_pool[key] # 检查连接是否还活跃简单 send probe try: sock.send(b) # 空 send 不触发数据只检测连接状态 self.upstream_pool[key] (sock, time.time()) return sock except OSError: # 连接已断清理 sock.close() del self.upstream_pool[key] # 新建连接 sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(10) # 连接超时 10s try: sock.connect((host, port)) except Exception as e: raise ConnectionError(fFailed to connect to {host}:{port}: {e}) with self.pool_lock: # LRU 策略满则踢出最久未用连接 if len(self.upstream_pool) self.max_pool_size: # 找到最久未用的 key oldest_key min(self.upstream_pool.keys(), keylambda k: self.upstream_pool[k][1]) old_sock, _ self.upstream_pool.pop(oldest_key) old_sock.close() self.upstream_pool[key] (sock, time.time()) return sock def _release_upstream_socket(self, host, port, sock): 将 socket 归还连接池 key (host, port) with self.pool_lock: # 检查是否已被其他线程关闭 try: sock.send(b) # probe self.upstream_pool[key] (sock, time.time()) except OSError: sock.close() def handle_client(self, client_sock, addr): upstream_sock None try: # ... [同前] 解析 request_line ... if method bCONNECT: self._handle_connect(client_sock, addr, path) return # 获取上游 socket复用或新建 target_host, target_port self._parse_host_port(host_header) upstream_sock self._get_upstream_socket(target_host, target_port) # 转发请求 upstream_sock.sendall(raw_request) # 转发响应 self._relay_response(client_sock, upstream_sock) except Exception as e: print(f[ERROR] {addr}: {e}) finally: client_sock.close() if upstream_sock: # 归还连接池非 CONNECT 场景 if method ! bCONNECT: self._release_upstream_socket(target_host, target_port, upstream_sock)这个连接池的关键设计点send(b)探活比select()更轻量且能捕获BrokenPipeErrorLRU 清理避免连接池无限增长max_pool_size可根据并发量调整idle_timeout未在此处实现实际应由后台线程定期扫描last_used_time并关闭过期连接本节为保持主线清晰暂略下一节补全。3.2 请求超时与连接保活别让一个坏请求拖垮整个服务没有超时控制的代理面对慢速攻击Slowloris或上游服务假死会迅速耗尽线程资源。我们必须为每个环节设置硬性超时环节超时类型推荐值作用socket.connect()连接建立超时5–10 秒避免 DNS 解析慢或目标宕机导致线程卡死socket.recv()读取超时30 秒防止上游返回一半就挂起socket.send()写入超时30 秒防止客户端网络拥塞导致 send 阻塞整体请求总超时60 秒作为兜底避免长尾请求占用资源def handle_client(self, client_sock, addr): client_sock.settimeout(60) # 整体请求超时 upstream_sock None try: # ... [同前] 读取 raw_request ... # 设置上游 socket 超时 upstream_sock self._get_upstream_socket(target_host, target_port) upstream_sock.settimeout(30) # 上游读写各 30s # 转发请求带超时 try: upstream_sock.sendall(raw_request) except socket.timeout: raise TimeoutError(Upstream send timeout) # 转发响应带超时 self._relay_response(client_sock, upstream_sock) except socket.timeout as e: self._send_error(client_sock, 504, Gateway Timeout) print(f[TIMEOUT] {addr}: {e}) except ConnectionError as e: self._send_error(client_sock, 502, Bad Gateway) print(f[CONN ERR] {addr}: {e}) except Exception as e: self._send_error(client_sock, 500, Internal Server Error) print(f[UNEXPECTED] {addr}: {e}) finally: client_sock.close() if upstream_sock: if method ! bCONNECT: self._release_upstream_socket(target_host, target_port, upstream_sock)注意socket.settimeout()是 per-socket 的不是全局设置。client_sock.settimeout(60)保证整个handle_client生命周期不超过 60 秒即使_relay_response内部多次recv()累计时间超限也会抛socket.timeout。3.3 请求日志没有日志的代理等于盲人开车生产环境必须记录每条请求的元信息时间、客户端 IP、目标 Host、HTTP 方法、状态码、响应大小、耗时。标准库logging足够胜任import logging from datetime import datetime # 配置日志建议写入文件而非 console logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(proxy.log, encodingutf-8), logging.StreamHandler() # 同时输出到终端便于调试 ] ) logger logging.getLogger(__name__) def handle_client(self, client_sock, addr): start_time time.time() client_ip, client_port addr status_code 0 response_size 0 target_host unknown try: # ... [同前] 解析 ... if method bCONNECT: # CONNECT 日志单独处理 logger.info(fCONNECT {host_header} from {client_ip}:{client_port}) self._handle_connect(client_sock, addr, path) return target_host, target_port self._parse_host_port(host_header) upstream_sock self._get_upstream_socket(target_host, target_port) upstream_sock.settimeout(30) upstream_sock.sendall(raw_request) # 记录请求开始 logger.info(fREQ {method.decode()} {path.decode()} - {target_host}:{target_port} from {client_ip}:{client_port}) # 转发响应此处需修改 _relay_response 以返回 status_code 和 size status_code, response_size self._relay_response_with_stats(client_sock, upstream_sock, raw_request) except Exception as e: status_code 500 logger.error(fERR {method.decode() if method in locals() else UNKNOWN} {path.decode() if path in locals() else } - {target_host}: {e}) finally: duration int((time.time() - start_time) * 1000) # 记录完成日志 logger.info(fRES {status_code} {response_size}B {duration}ms - {target_host} from {client_ip}:{client_port}) client_sock.close() if upstream_sock and method ! bCONNECT: self._release_upstream_socket(target_host, target_port, upstream_sock)_relay_response_with_stats需要改造_relay_response在转发过程中统计响应体字节数并从响应头中提取状态码def _relay_response_with_stats(self, client_sock, upstream_sock, raw_request): 返回 (status_code, response_size) response_head b while not response_head.endswith(b\r\n\r\n): chunk upstream_sock.recv(4096) if not chunk: return 0, 0 response_head chunk # 解析状态码 status_code 0 try: status_line response_head.split(b\r\n)[0] status_code int(status_line.split()[1]) except (IndexError, ValueError): pass # 计算响应体大小headers 不计入 response_size 0 is_chunked btransfer-encoding: chunked in response_head.lower() client_sock.sendall(response_head) if is_chunked: response_size self._relay_chunked_body_with_count(client_sock, upstream_sock) else: response_size self._relay_content_length_body_with_count(client_sock, upstream_sock, response_head) return status_code, response_size def _relay_chunked_body_with_count(self, client_sock, upstream_sock): count 0 while True: size_line b while not size_line.endswith(b\r\n): chunk upstream_sock.recv(1) if not chunk: return count size_line chunk size_hex size_line.strip().split(b;)[0] try: size int(size_hex, 16) except ValueError: return count if size 0: # read trailer and final \r\n while True: trailer upstream_sock.recv(1) if trailer b\r: next_byte upstream_sock.recv(1) if next_byte b\n: break client_sock.sendall(b0\r\n\r\n) break data b while len(data) size 2: chunk upstream_sock.recv(min(4096, size 2 - len(data))) if not chunk: return count data chunk client_sock.sendall(data) count len(data) - 2 # -2 for \r\n return count日志格式示例2024-06-15 14:22:33,123 - INFO - REQ GET /api/user - api.example.com:443 from 192.168.1.100:54321 2024-06-15 14:22:33,456 - INFO - RES 200 1248B 333ms - api.example.com from 192.168.1.100:54321这为后续排查“哪个客户端在高频刷接口”、“哪个上游服务响应变慢”提供了原始依据。4. 避坑指南那些让你调试到凌晨三点的 HTTP 代理经典翻车现场写一个能跑的代理容易写一个不出幺蛾子的代理难。以下是我在线上环境踩过的 5 个真实坑每个都附带现象、根因和血泪修复方案。它们不会出现在任何教程里但会真实消耗你 80% 的调试时间。4.1 现象curl 能通requests 就 timeoutWireshark 显示 client 发了 FIN但 proxy 没回 ACK原因requests默认启用Connection: keep-alive而你的代理在转发响应后没有正确处理Connection: closeheader导致 client 认为连接还活着继续发下一个请求但 proxy 已关闭 socket造成半关闭状态。解决在_relay_response结束后显式检查响应头中的Connection字段。如果值为close则立即client_sock.close()不等待 client 主动断连def _relay_response(self, client_sock, upstream_sock): # ... [同前] 读取 response_head ... # 检查 Connection header connection_close False for line in response_head.split(b\r\n): if line.lower().startswith(bconnection:) and bclose in line.lower(): connection_close True break # ... [转发 headers 和 body] ... # 关闭逻辑 if connection_close: client_sock.close() # 主动关闭避免半关闭 # 否则让 client 决定何时断连Keep-Alive 场景血泪经验不要相信Content-Length或Transfer-Encoding能替代Connectionheader 的语义。HTTP/1.1 规范明确要求Connection: close优先级高于其他。4.2 现象代理能访问 HTTP 站点但 HTTPS 站点返回502 Bad Gateway且日志显示Connection refused原因CONNECT请求的目标端口是443但你的_parse_host_port函数默认 fallback 到80导致 upstream socket 尝试连接example.com:80而非443。解决CONNECT方法必须强制使用443除非显式指定了其他端口def _handle_connect(self, client_sock, addr, connect_target): try: # CONNECT target 格式host:port 或 host隐含 443 target_str connect_target.decode(utf-8) if : in target_str: host, port target_str.split(:, 1) port int(port) else: host target_str port 443 # HTTPS 默认端口 # ... [后续 connect] ...4.3 现象代理运行几小时后OSError: Too many open files系统报错原因Linux 默认每个进程最多打开 1024 个文件描述符fd而每个 socket 连接占用 1 个 fd。连接池未做清理TIME_WAIT状态连接堆积最终耗尽 fd。解决三管齐下操作系统层sudo sysctl -w net.ipv4.ip_local_port_range1024 65535扩大端口范围Python 层在upstream_pool中加入后台清理线程定期关闭空闲超时连接def __init__(self, *args, **kwargs): # ... [同前] ... self.cleanup_thread threading.Thread(targetself._cleanup_pool) self.cleanup_thread.daemon True self.cleanup_thread.start() def _cleanup_pool(self): while True: time.sleep(10) # 每 10 秒扫描一次 now time.time() to_remove [] with self.pool_lock: for key, (sock, last_used) in self.upstream_pool.items(): if now - last_used self.idle_timeout: try: sock.close() except: pass to_remove.append(key) for key in to_remove: del self.upstream_pool[key]应用层在handle_client的finally块中确保upstream_sock.close()被调用即使异常发生。4.4 现象POST 请求体丢失、JSON 解析失败Wireshark 显示 body 只有前 100 字节原因raw_request累积读取时recv()返回的 chunk 可能包含不完整 HTTP body而你错误地认为\r\n\r\n出现后 body 就结束了。实际上Content-Length指定的 body 长度可能远大于初始 recv 的 chunk。解决raw_request只读取 headersbody 必须按Content-Length或Transfer-Encoding单独读取def handle_client(self, client_sock, addr): # Step 1: Read headers only raw_headers b while not raw_headers.endswith(b\r\n\r\n): chunk client_sock.recv(4096) if not chunk: return raw_headers chunk # Parse Content-Length or Transfer-Encoding content_length 0 is_chunked False for line in raw_headers.split(b\r\n): if line.lower().startswith(bcontent-length:): try: content_length int(line.split(b:, 1)[1].strip()) except: pass elif line.lower().startswith(btransfer-encoding:) and bchunked in line.lower(): is_chunked True # Step 2: Read body based on encoding raw_body b if is_chunked: raw_body self._read_chunked_body(client_sock) elif content_length 0: raw_body self._read_exact_bytes(client_sock, content_length) # Step 3: Assemble full request raw_request raw_headers raw_body # ... [后续逻辑] ..._read_exact_bytes和_read_chunked_body是独立函数确保 body 读取的原子性。4.5 现象代理部署到树莓派后CPU 占用 100%top显示大量python进程原因threading.Thread创建过多而树莓派 CPU 核心少、内存小线程调度开销巨大。one-thread-per-connection模型在高并发下失效。解决切换为thread pool 模型限制最大并发线程数from concurrent.futures import ThreadPoolExecutor class ThreadPoolHTTPProxy(PooledHTTPProxy): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.executor ThreadPoolExecutor(max_workers10) # 限制 10 个 worker def serve_forever(self): while True: try: client_sock, addr self.server_socket.accept() # 提交到线程池而非直接创建 Thread self.executor.submit(self.handle_client, client_sock, addr) except OSError as e: if Bad file descriptor in str(e): break raisemax_workers10是树莓派 4B 的安全值可根据nproc动态调整。5. 进阶实战用这个代理做三件真正有用的事——不只是转发写完一个健壮的代理下一步是让它产生业务价值。下面三个场景都是本文还有配套的精品资源点击获取
