如何改造带锁的Python多线程程序以支持多进程运行？

核心问题在于：你在父进程创建了Runner和Handler实例，其中Handler依赖的queue.Queue包含不可序列化的threading.Lock对象，而ProcessPoolExecutor需要将任务参数序列化后传递给子进程，导致报错。

解决思路是让每个子进程自行初始化Runner和内部对象，而非在父进程创建后传递。这样既避免了序列化问题，也符合多进程的内存隔离特性。

步骤1：修改Handler的run方法，添加退出条件 #

原代码中Handler.run是无限循环，会导致子进程无法正常退出。添加哨兵值（如None）来触发退出：

from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
from queue import Queue

class Handler:
    def __init__(self):
        self.queue = Queue()  # 线程安全队列，仅在进程内的线程间使用，无需跨进程

    def put(self, item):
        self.queue.put(item)

    def run(self):
        while True:
            item = self.queue.get()
            if item is None:  # 哨兵值，收到后终止循环
                break
            # 处理item的逻辑
            print(item)
        self.queue.task_done()  # 标记队列任务完成（可选）

步骤2：修改Runner的start方法，发送退出信号 #

在向Handler发送完业务数据后，发送哨兵值让run方法退出：

class Runner:
    def __init__(self, name):
        self.name = name
        self.a = Handler()
        self.b = Handler()

    def start(self):
        # 发送业务数据
        self.a.put(f'{self.name}: hello a')
        self.b.put(f'{self.name}: hello b')
        # 发送退出信号
        self.a.put(None)
        self.b.put(None)
        # 启动内部线程处理队列
        with ThreadPoolExecutor() as exe:
            futures = [exe.submit(r.run) for r in [self.a, self.b]]
        for future in futures:
            future.result()

步骤3：重构多进程运行逻辑 #

将Runner的创建和启动逻辑封装为顶层函数，仅向子进程传递可序列化的参数（如名字字符串）：

# 串行模式实现
def run_in_serial():
    for name in ['A', 'B', 'C']:
        runner = Runner(name)
        runner.start()

# 原多线程模式（无需修改）
def run_in_multi_thread():
    rA = Runner('A')
    rB = Runner('B')
    rC = Runner('C')
    with ThreadPoolExecutor() as exe:
        futures = [exe.submit(r.start) for r in [rA, rB, rC]]
        for future in futures:
            future.result()

# 修复后的多进程模式
def _run_runner(name):
    # 子进程内自行创建Runner实例
    runner = Runner(name)
    runner.start()

def run_in_multi_process():
    with ProcessPoolExecutor() as exe:
        # 仅传递可序列化的名字参数，而非Runner实例
        futures = [exe.submit(_run_runner, name) for name in ['A', 'B', 'C']]
        for future in futures:
            future.result()

步骤4：顶层模式选择 #

添加命令行参数或交互选择，让用户指定运行模式：

if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print("Usage: python script.py [serial|thread|process]")
        sys.exit(1)
    
    mode = sys.argv[1].lower()
    if mode == 'serial':
        run_in_serial()
    elif mode == 'thread':
        run_in_multi_thread()
    elif mode == 'process':
        run_in_multi_process()
    else:
        print("Invalid mode. Choose 'serial', 'thread', or 'process'")

关键说明 #

1. 为什么不用multiprocessing.Queue：你的场景中Handler的队列仅用于进程内部的线程间通信，queue.Queue是线程安全的，完全满足需求。multiprocessing.Queue是用于跨进程通信的，反而会引入额外的复杂度和限制（如必须通过继承或Manager共享）。
2. 避免序列化问题的核心：子进程的所有依赖对象（如Handler、Queue）都在子进程内部初始化，父进程仅传递简单的可序列化参数，彻底规避了不可pickle对象的传递问题。

内容的提问来源于stack exchange，提问作者quantum.snowball