当我调用 process.join 时，如何运行脚本？

Question

我有几个进程打算在 while 循环中运行。我基本上有一些收集数据的进程，在它们停止之前，我希望它们将数据保存到 csv 或 json 文件中。我现在拥有的是使用 super 函数来覆盖 multiprocessing.Process 类中的 join 方法。

class Processor(multiprocessing.Process):
    def __init__(self, arguments):
        multiprocessing.Process.__init__(self)

    def run(self):
        self.main_function()

    def main_function(self):
        While True:
            #do things to incoming data

    def function_on_join(self):
        #do one last thing before the process ends

    def join(self, timeout=None):
        self.function_on_join()
        super(Processor, self).join(timeout=timeout)

有没有更好的方法/正确的方法/更 Pythonic 的方法来做到这一点？

Answer 1

我建议你看看concurrent.futures 模块。

如果您可以将您的工作描述为一组工作人员要完成的任务列表。

基于任务的多处理

当您有一系列jobs（例如文件名列表）并且您希望并行处理它们时 - 您可以按照以下方式进行：

from concurrent.futures import ProcessPoolExecutor    
import requests

def get_url(url):
    resp = requests.get(url)
    print(f'{url} - {resp.status_code}')
    return url

jobs = ['http://google.com', 'http://python.org', 'http://facebook.com']

# create process pool of 3 workers
with ProcessPoolExecutor(max_workers=1) as pool:
    # run in parallel each job and gather the returned values
    return_values = list(pool.map(get_url, jobs))

print(return_values)

输出：

http://google.com - 200
http://python.org - 200
http://facebook.com - 200
['http://google.com', 'http://python.org', 'http://facebook.com']

不是基于任务的多处理

当您只想像第一种情况那样运行多个不消耗作业的子进程时，您可能需要使用multiprocessing.Process。

您可以像 threading.Thread 一样以程序方式和 OOP 方式使用它。

程序时尚示例（恕我直言，更 Pythonic）：

import os
from multiprocessing import Process

def func():
    print(f'hello from: {os.getpid()}')

processes = [Process(target=func) for _ in range(4)]  # creates 4 processes

for process in processes:
    process.daemon = True  # close the subprocess if the main program closes
    process.start()  # start the process

输出：

hello from: 31821
hello from: 31822
hello from: 31823
hello from: 31824

等待进程完成

如果您想使用 Process.join() 等待（更多信息请参见 process.join() 和 process.daemon this SO answer），您可以这样做：

import os
import time
from multiprocessing import Process

def func():
    time.sleep(3)
    print(f'hello from: {os.getpid()}')

processes = [Process(target=func) for _ in range(4)]  # creates 4 processes

for process in processes:
    process.start()  # start the process

for process in processes:
    process.join()  # wait for the process to finish

print('all processes are done!')

这个输出：

hello from: 31980
hello from: 31983
hello from: 31981
hello from: 31982
all processes are done!