如何在Python的ProcessPoolExecutor进程间安全访问变量？

首先要明确：你的原代码里，books列表在多进程环境下根本无法实现共享——因为ProcessPoolExecutor的子进程会复制主进程的内存空间，每个子进程修改的都是自己的books副本，主进程和其他子进程完全看不到这些修改。所以你需要用专门的进程间共享机制，而且确实有不用手动加锁的简便方案。

方案1：使用multiprocessing.Manager提供的共享数据结构 #

multiprocessing.Manager会创建一个独立的管理进程，负责维护共享数据，所有子进程通过代理访问这些数据，内部已经封装了同步逻辑（不用自己加锁）。最适合你的场景是用Manager.list()替代普通列表：

修改后的代码：

import concurrent.futures
import time
import random
import multiprocessing

def do_book_task(books, year):
    books.append(year)
    print(f'Doing task with books from {year}.')
    time.sleep(random.random() + 0.5)
    books.remove(year)
    return f'Result for {year}'

def main():
    years = ['1996', '1997', '1998', '1999', '2000', '2001']
    # 使用Manager创建进程安全的共享列表
    with multiprocessing.Manager() as manager:
        books = manager.list()
        with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
            future_year_map = {}
            futures = []
            for year in years:
                future = executor.submit(do_book_task, books, year)
                future_year_map[future] = year
                futures.append(future)
                print(f'Submitted {year} to process queue')
            
            for future in concurrent.futures.as_completed(futures):
                try:
                    year = future_year_map[future]
                    result = future.result()
                    print(f'Done {year}')
                    print(result)
                except Exception as e:
                    print(f'Error with year {year}: {e}')

if __name__ == '__main__':
    main()

注意：原代码用years[futures.index(future)]存在隐患，若有重复任务会拿到错误年份，改用future_year_map字典映射更可靠。

方案2：主进程维护状态（更高效，无需进程间共享） #

如果你的需求只是跟踪当前正在运行的任务年份，完全不需要让子进程修改共享变量，由主进程自己维护状态即可：

• 提交任务时，把年份加入主进程的列表
• 通过future的回调函数，在任务完成时从列表中移除

代码示例：

import concurrent.futures
import time
import random

def do_book_task(year):
    print(f'Doing task with books from {year}.')
    time.sleep(random.random() + 0.5)
    return f'Result for {year}'

def main():
    years = ['1996', '1997', '1998', '1999', '2000', '2001']
    running_years = []
    future_year_map = {}

    def on_task_done(future):
        year = future_year_map[future]
        running_years.remove(year)
        print(f'Done {year}')
        print(future.result())

    with concurrent.futures.ProcessPoolExecutor(max_workers=4) as executor:
        for year in years:
            future = executor.submit(do_book_task, year)
            future_year_map[future] = year
            running_years.append(year)
            future.add_done_callback(on_task_done)
            print(f'Submitted {year} to process queue')

if __name__ == '__main__':
    main()

这种方式完全避免了进程间通信的开销，所有状态操作都在主进程里完成，逻辑更简单安全。

补充：IO-bound任务的另一种选择 #

如果你的任务是纯IO-bound（比如文件读写、网络请求），可以考虑用ThreadPoolExecutor替代ProcessPoolExecutor：

• 线程共享主进程内存，普通列表的append是原子操作，但remove不是（需先查找元素），不过如果每个任务只操作自己的年份，也可以用线程锁（比进程锁简单），或者直接用主进程维护状态的方式，和方案2逻辑一致。

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