python - concurrent.futures.ProcessPoolExecutor() python 中的共享变量

Question

我想在 python 中使用模块 concurrent.futures 使用并行更新全局变量

原来，使用 ThreadPoolExecutor 可以更新我的全局变量，但 CPU 并没有充分利用它们的潜力(总是在 5-10%)，这太慢了

和 ProcessPoolExecutor 可以使用所有内核(100%)但我的全局变量无法更新，因为它们不共享相同的全局变量

如何在 concurrent.futures 模型中使用 ProcessPoolExecutor 共享我的全局变量。非常感谢您的帮助

Answer 1

进程似乎不像使用相同内存空间的线程。所以你需要一些特殊的方法来更新变量。 ProcessPoolExecutor使用 multiprocessing模块，有两种共享数据的方式，共享内存和服务器进程。第一种使用共享内存映射的方式，服务器进程使用 Manager像代理一样保存共享数据的对象。服务器进程更灵活，共享内存更高效。
使用服务器进程共享数据，如 ThreadPoolExecutor ，只需将参数传递给您的函数。

def running_proxy(mval):
    # consider lock if you need
    return mval.value

def start_executor():
    with multiprocessing.Manager() as manager:
        executor = ProcessPoolExecutor(max_workers=5)
        mval = manager.Value('b', 1)
        futures = [executor.submit(running_proxy, mval) for _ in range(5)]
        results = [x.result() for x in futures]
        executor.shutdown()

但是共享内存方式有一些不同，您需要将共享变量设置为全局。

def running_shared():
    # consider lock if you need
    return sval.value

def set_global(args):
    global sval
    sval = args

def start_executor():
    sval = multiprocessing.Value('b', 1)
    # for 3.7+
    executor = ProcessPoolExecutor(max_workers=5, initializer=set_global, initargs=(sval,))
    # for ~3.6
    # set_global(sval)
    # executor = ProcessPoolExecutor(max_workers=5)
    futures = [executor.submit(running_shared) for _ in range(5)]
    results = [x.result() for x in futures]
    executor.shutdown()