python - 当我调用 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 模块。

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

基于任务的多处理

当您有一系列作业(例如文件名列表)并且您希望并行处理它们时 - 您可以按如下方式操作:

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() 等待(有关 this SO answer 上的 process.join() 和 process.daemon 的更多信息)你可以这样做:

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!