Python多进程Manager报错：'ForkAwareLocal'无'connection'属性求助

问题原因 #

• 主进程创建multiprocessing.Manager()后直接结束，导致Manager进程随主进程终止，子进程无法连接到共享队列，触发FileNotFoundError。
• 消费者使用get_nowait()时，队列为空会抛出Empty异常，未做捕获处理会导致进程崩溃。

修正后的代码 #

import time
import random
import multiprocessing
from multiprocessing import Empty


# 生产者类：模拟数据采集流程
class Producer:
    def __init__(self, queue):
        self.queue = queue

    def run(self):
        print("Producer process started...")
        while True:
            # 模拟数据采集耗时
            input_time = random.randrange(1, 4)
            time.sleep(input_time)

            # 模拟生成采集数据
            input_data = random.randrange(5, 10)
            try:
                self.queue.put_nowait(input_data)
                print(f" {input_data} 已采集，耗时 {input_time} 秒")
            except Exception as e:
                print(f"生产者放入数据失败: {e}")
                time.sleep(0.5)


# 消费者类：模拟数据处理流程
class Consumer:
    def __init__(self, queue):
        self.queue = queue

    def run(self):
        process_data = 0
        print("Consumer process started...")
        while True:
            try:
                data = self.queue.get_nowait()
                # 模拟数据处理耗时
                procss_time = random.randrange(6, 9)
                time.sleep(procss_time)

                # 模拟数据处理逻辑
                process_data += data
                print(f" 输入数据: {data}, 当前结果: {process_data}，处理耗时 {procss_time} 秒")
            except Empty:
                # 队列为空时短暂等待，避免空轮询占用资源
                time.sleep(0.5)
                continue
            except Exception as e:
                print(f"消费者获取数据失败: {e}")
                break


if __name__ == "__main__":
    # 使用原生共享队列替代Manager.Queue，避免依赖额外进程
    queue = multiprocessing.Queue()

    producer = Producer(queue)
    consumer = Consumer(queue)

    # 创建并启动进程
    producer_process = multiprocessing.Process(target=producer.run)
    consumer_process = multiprocessing.Process(target=consumer.run)

    producer_process.start()
    consumer_process.start()

    # 主进程等待子进程运行，防止提前退出释放资源
    producer_process.join()
    consumer_process.join()

关键修改说明 #

• 替换manager.Queue()为multiprocessing.Queue()：原生队列更轻量，无需依赖Manager进程，彻底解决连接问题。
• 添加join()方法：主进程等待子进程执行，避免提前退出导致队列资源被回收。
• 异常捕获处理：生产者捕获put_nowait()可能的异常，消费者捕获Empty异常并做等待处理，防止进程崩溃。
• 移除无用导入：删除未使用的Pool模块导入，精简代码。

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