Python3基于共享数据并行运行n个进程的实现问题求助

问题根源 #

• 单进程运行异常：创建进程时每启动一个就立刻调用join()，该方法会阻塞主进程等待当前子进程退出，而你的run()是无限循环永远不会结束，导致后续进程永远无法启动。
• Pool卡死异常：multiprocessing.Pool专为可结束的短任务设计，配套的任务调度方法会等待任务执行完成返回结果，无限循环任务永远不会返回，自然会陷入阻塞；此外进程默认拥有独立内存空间，普通Python对象传入子进程时是拷贝传递，各进程修改的数据无法互相同步。

实现方案 #

场景1：子进程无需共享修改后的数据 #

如果各子进程仅读取初始数据，修改仅在子进程内部生效，直接按先全启动再统一join的逻辑实现即可：

# 完整可运行代码
import multiprocessing

class App:
    items = {"item_id": {"field": "初始数据"}}

def check(item):
    # 你的检查逻辑，修改仅对当前子进程生效
    return item

def run(app):
    while True:
        check(app.items['item_id'])

if __name__ == '__main__':
    PROCESS_COUNT = 5 # 自定义需要启动的进程数
    app = App()
    processes = []
    # 先全部启动
    for _ in range(PROCESS_COUNT):
        p = multiprocessing.Process(target=run, args=(app,))
        p.daemon = True # 主进程退出时自动销毁子进程，按需开启
        processes.append(p)
        p.start()
    # 再统一等待
    for p in processes:
        p.join()

场景2：所有子进程需要同步共享数据 #

如果需要各子进程修改的items数据全局同步，要使用多进程安全的共享数据结构，并发修改时加锁避免竞态问题：

import multiprocessing
from multiprocessing import Manager, Lock

class App:
    def __init__(self, shared_dict, lock):
        self.items = shared_dict
        self.lock = lock

def check(item, lock):
    # 修改共享数据时加锁，仅读取可省略
    with lock:
        item['field'] = '更新后的值'
    return item

def run(app):
    while True:
        check(app.items['item_id'], app.lock)

if __name__ == '__main__':
    PROCESS_COUNT = 5
    with Manager() as manager:
        # 初始化共享字典，写入初始数据
        shared_items = manager.dict()
        shared_items['item_id'] = {"field": "初始值"}
        # 全局并发锁
        process_lock = Lock()
        app = App(shared_items, process_lock)
        processes = []
        for _ in range(PROCESS_COUNT):
            p = multiprocessing.Process(target=run, args=(app,))
            p.daemon = True
            processes.append(p)
            p.start()
        for p in processes:
            p.join()

注意事项 #

• 若check仅涉及读操作，不需要加锁，可大幅提升运行效率
• 若需要子进程在主进程退出后继续运行，去掉p.daemon = True配置即可

内容的提问来源于stack exchange，提问作者Yaniv.Yscr