我正在尝试使用 ProcessPoolExecutor，但我收到错误“队列对象只能通过继承在进程之间共享”，但我没有使用队列(至少没有明确使用)。我找不到任何可以解释我做错了什么的东西。
这是一些演示问题的代码(不是我的实际代码):
from concurrent.futures import ProcessPoolExecutor, as_completed
class WhyDoesntThisWork:
def __init__(self):
self.executor = ProcessPoolExecutor(4)
def execute_something(self, starting_letter):
futures = [self.executor.submit(self.something, starting_letter, d) for d in range(4)]
letter = None
for future in as_completed(futures):
letter = future.result()
print(letter)
def something(self, letter, d):
# do something pointless for the example
for x in range(d):
letter = chr(ord(letter) + 1)
if __name__ == '__main__':
WhyDoesntThisWork(). execute_something('A')
El Ruso 指出将 something() 设为静态方法或类方法会使错误消失。不幸的是，我的实际代码需要使用 self 调用其他方法。
请您参考如下方法:
可以不使用静态方法解决。
使用进程时，每个进程运行在独立的内存空间中。这与使用线程时不同，当不同的线程在同一进程下运行时，使用相同的内存空间。因此，当您使用时不会发生错误 ThreadPoolExecutor 但出现在 ProcessPoolExecutor 中。
所以当类实例的函数被传递到单独的子进程中时，多处理机制会对该函数进行pickle，这样函数就可以作为一个独立的实例传递到子进程中。并且当子流程加入时，该类由返回的函数实例更新为未拾取的。
要让它工作，只需在类中添加__getstate__() 和__setstate__() 函数来指导类如何pickle 和unpickle 函数。在 pickling 中，可以排除不必要的字段，如 del self_dict['executor'] 所示。
import multiprocessing
import time
from concurrent.futures import ProcessPoolExecutor, as_completed
class GuessItWorksNow():
def __init__(self):
self.executor = ProcessPoolExecutor(4)
def __getstate__(self):
state = self.__dict__.copy()
del state['executor']
return state
def __setstate__(self, state):
self.__dict__.update(state)
def something(self, letter, d):
# do something pointless for the example
p = multiprocessing.current_process()
time.sleep(1)
for x in range(d):
letter = chr(ord(letter) + 1)
return (f'[{p.pid}] ({p.name}) ({letter})')
def execute_something(self, starting_letter):
futures = [self.executor.submit(self.something, starting_letter, d) for d in range(10)]
for future in as_completed(futures):
print(future.result())
if __name__ == '__main__':
obj = GuessItWorksNow()
obj.execute_something('A')
