Python 中的 ProcessPoolExecutor 和 Lock

问题描述

我正在尝试将 concurrent.futures.ProcessPoolExecutor 与 Locks 一起使用，但出现运行时错误.(如果相关，我正在使用 Windows)

I am trying to use concurrent.futures.ProcessPoolExecutor with Locks, but I'm getting a run time error. (I'm working on Windows if that's relevant)

这是我的代码:

import multiprocessing
from concurrent.futures import ProcessPoolExecutor

import time


def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')


def main():
    lock = multiprocessing.Lock()
    pool = ProcessPoolExecutor()
    futures = [pool.submit(f, num, lock) for num in range(3)]
    for future in futures:
        future.result()


if __name__ == '__main__':
    main()

这是我得到的错误:

    Traceback (most recent call last):
  File "F:WinPython-64bit-3.4.3.2python-3.4.3.amd64Libmultiprocessingqueues.py", line 242, in _feed
    obj = ForkingPickler.dumps(obj)
  File "F:WinPython-64bit-3.4.3.2python-3.4.3.amd64Libmultiprocessingeduction.py", line 50, in dumps
    cls(buf, protocol).dump(obj)
  File "F:WinPython-64bit-3.4.3.2python-3.4.3.amd64Libmultiprocessingsynchronize.py", line 102, in __getstate__
    context.assert_spawning(self)
  File "F:WinPython-64bit-3.4.3.2python-3.4.3.amd64Libmultiprocessingcontext.py", line 347, in assert_spawning
    ' through inheritance' % type(obj).__name__
RuntimeError: Lock objects should only be shared between processes through inheritance

奇怪的是，如果我用 multiprocessing.Process 编写相同的代码，一切正常:

What's weird is that if I write the same code with multiprocessing.Process it all works fine:

import multiprocessing

import time


def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')


def main():
    lock = multiprocessing.Lock()
    processes = [multiprocessing.Process(target=f, args=(i, lock)) for i in range(3)]
    for process in processes:
        process.start()
    for process in processes:
        process.join()



if __name__ == '__main__':
    main()

这行得通，我得到:

1 hello
1 world
0 hello
0 world
2 hello
2 world

推荐答案

你需要使用 Manager 并使用 Manager.Lock() 代替:

You need to use a Manager and use a Manager.Lock() instead:

import multiprocessing
from concurrent.futures import ProcessPoolExecutor

import time

def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')

def main():
    pool = ProcessPoolExecutor()
    m = multiprocessing.Manager()
    lock = m.Lock()
    futures = [pool.submit(f, num, lock) for num in range(3)]
    for future in futures:
        future.result()


if __name__ == '__main__':
    main()

结果:

% python locks.py
0 hello
0 world
1 hello
1 world
2 hello
2 world