多线程TCP Socket服务器遇[Errno 32] Broken pipe错误求助

我来帮你分析下问题所在，以及对应的修复方案：

问题根源 #

你的问题主要来自两个核心点：

1. 服务器处理单次请求后立即关闭连接：ThreadedTCPRequestHandler的handle方法只执行了一次recv就结束了，而socketserver.BaseRequestHandler会在handle方法执行完毕后自动关闭当前连接。但graphitesend.init()创建的客户端默认是长连接模式——它会复用同一个TCP连接发送多次数据。当服务器第一次处理完请求关闭连接后，后续客户端再用这个已失效的连接发送数据，就会触发Broken pipe错误。
2. 服务器被过早停止：你在启动客户端线程后立刻调用server.stop()，此时服务器可能还没处理完所有请求就被强制关闭，这也会导致部分连接异常中断。

修复方案 #

针对这两个问题，我们可以做以下调整：

1. 让服务器支持长连接（持续接收数据） #

修改handle方法，让它在循环中持续接收数据，直到客户端主动关闭连接：

class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
    def handle(self):
        # 循环接收数据，直到客户端关闭连接（recv返回空）
        while True:
            data = self.request.recv(1024)
            if not data:
                break
            print(f"Received from {self.client_address}: {data.decode('utf-8').strip()}")

2. 调整服务器停止时机，等待所有请求处理完成 #

使用threading.join()等待所有客户端线程执行完毕后，再停止服务器，同时给服务器一点时间收尾处理：

if __name__ == "__main__":
    HOST, PORT = "localhost", 2003
    server = ThreadedTCPServer(HOST, PORT)
    server.start()
    
    g = graphitesend.init(graphite_server=HOST, graphite_port=PORT, prefix=None)
    client_threads = []
    # 创建并保存客户端线程对象
    for _ in range(7):
        t = threading.Thread(target=client, args=(g,))
        t.start()
        client_threads.append(t)
    
    # 等待所有客户端线程执行完成
    for t in client_threads:
        t.join()
    
    # 给服务器一点时间处理最后一批数据
    import time
    time.sleep(0.2)
    
    server.stop()

额外说明 #

• graphitesend的长连接是默认行为，如果你想每次请求都新建连接，可以在init时加上autoreconnect=True，但长连接更高效，所以优先让服务器适配长连接是更合理的选择。
• 你设置的allow_reuse_address = True是正确的，能避免端口占用问题。

完整修复后的代码 #

import threading
import socketserver
import graphitesend
import time

class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
    def handle(self):
        # 持续接收数据直到连接关闭
        while True:
            data = self.request.recv(1024)
            if not data:
                break
            print(f"Received: {data.decode('utf-8').strip()}")

class ThreadedTCPServer(socketserver.ThreadingTCPServer):
    allow_reuse_address = True
    def __init__(self, host, port):
        socketserver.ThreadingTCPServer.__init__(self, (host, port), ThreadedTCPRequestHandler)
    def stop(self):
        self.server_close()
        self.shutdown()
    def start(self):
        threading.Thread(target=self._on_started).start()
    def _on_started(self):
        self.serve_forever()

def client(g):
    g.send("test", 1)

if __name__ == "__main__":
    HOST, PORT = "localhost", 2003
    server = ThreadedTCPServer(HOST, PORT)
    server.start()
    
    # 初始化graphitesend客户端，prefix设为None避免自动添加前缀
    g = graphitesend.init(graphite_server=HOST, graphite_port=PORT, prefix=None)
    client_threads = []
    
    # 创建7个客户端线程
    for _ in range(7):
        thread = threading.Thread(target=client, args=(g,))
        thread.start()
        client_threads.append(thread)
    
    # 等待所有客户端线程完成
    for thread in client_threads:
        thread.join()
    
    # 给服务器一点时间处理剩余数据
    time.sleep(0.2)
    
    server.stop()

这样修改后，服务器就能正常处理同一个客户端连接的多次请求，不会再出现Broken pipe错误了。

内容的提问来源于stack exchange，提问作者NicoAdrian