在ProcessPoolExecutor中调用socket.close()无法关闭连接的问题

需求思路 #

• 主线程运行Socket服务器监听并接受连接
• 使用connection_pool（max_workers=10）进程池处理客户端Socket连接
• 客户端Socket的数据由writing_pool（max_workers=1）进程池处理

问题现象 #

将客户端Socket传入ProcessPoolExecutor后，在connection_pool执行的处理函数中调用close()，客户端Socket连接并未实际关闭。

可能推测 #

怀疑客户端Socket对象传入进程池时，不是以引用方式传递，导致子进程内关闭的是副本而非原连接。

代码示例 #

核心代码 #

import socket
from concurrent.futures import ProcessPoolExecutor

connection_pool = ProcessPoolExecutor(max_workers=10)
writing_pool = ProcessPoolExecutor(max_workers=1)


def addData(data):
    try:
        print("ADD DATA: " + data)
    except Exception as e:
        print("[!] error", e)
        return 1
    return 0


def handleClient(client: socket):
    try:
        print("HANDLE CLIENT")
        data = client.recv(1024).decode()
        client.close()
        print("HANDLED CLIENT")
        writing_pool.submit(addData, data)
    except Exception as e:
        print("[!] error", e)
        return 1
    return 0

可用示例（主线程内关闭客户端连接） #

if __name__ == "__main__":
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        listener.bind(("127.0.0.1", 4040))
        listener.listen()
        while (True):
            client, addr = listener.accept()
            print("HANDLE CLIENT")
            data = client.recv(1024).decode()
            client.close()
            print("HANDLED CLIENT")
            addData(data)
    except Exception as e:
        print("[!] error", e)
        raise e
    finally:
        listener.close()

不可用示例（进程池内关闭客户端连接） #

if __name__ == "__main__":
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        listener.bind(("127.0.0.1", 4040))
        listener.listen()
        while (True):
            client, addr = listener.accept()
            print("CONNECTION")
            connection_pool.submit(handleClient, client)
    except Exception as e:
        print("[!] error", e)
        raise e
    finally:
        listener.close()

测试方式 #

用bash执行netcat测试：

for data in {1..9}; do echo "$data" | nc 127.0.0.1 4040 -q 1 ; done

环境信息 #

问题解答 #

问题根源在于**ProcessPoolExecutor基于多进程实现**，Python多进程间的对象传递依赖pickle序列化/反序列化。当你把客户端Socket传入子进程时，传递的是原Socket的副本而非原对象引用。子进程内调用close()仅关闭副本对应的文件描述符，主线程中的原Socket连接依然存在，导致客户端连接未真正断开。

解决方案

1. 改用ThreadPoolExecutor替代ProcessPoolExecutor：线程池内的线程共享同一进程的文件描述符，Socket对象可直接传递引用，子线程内关闭Socket会真正断开连接。修改后的代码示例：

import socket
from concurrent.futures import ThreadPoolExecutor

connection_pool = ThreadPoolExecutor(max_workers=10)
writing_pool = ThreadPoolExecutor(max_workers=1)

# addData和handleClient函数保持不变

if __name__ == "__main__":
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        listener.bind(("127.0.0.1", 4040))
        listener.listen()
        while True:
            client, addr = listener.accept()
            print("CONNECTION")
            connection_pool.submit(handleClient, client)
    except Exception as e:
        print("[!] error", e)
        raise e
    finally:
        listener.close()

2. 若必须使用多进程：可在主线程中接收数据后关闭Socket，再将数据传入进程池处理；或通过传递文件描述符的方式实现（需借助multiprocessing的共享工具），但该方式实现复杂，不如线程池直接。

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