如何使用多处理在Python中运行多个异步进程？

Posted 2021-04-13

我需要使用多处理来运行多个后台异步函数。我有Popen的解决方案，但看起来有点不自然。例：

from time import sleep
from multiprocessing import Process, Value
import subprocess

def worker_email(keyword):
    subprocess.Popen(["python", "mongoworker.py", str(keyword)])
    return True

keywords_list = ['apple', 'banana', 'orange', 'strawberry']

if __name__ == '__main__':
    for keyword in keywords_list:
        # Do work
        p = Process(target=worker_email, args=(keyword,))
        p.start()
        p.join()

如果我尽量不使用Popen，比如：

def worker_email(keyword):
    print('Before:' + keyword)
    sleep(10)
    print('After:' + keyword)
    return True

函数一个接一个地运行，没有异步。那么，如何在不使用Popen的情况下同时运行所有功能？

UPD：我正在使用multiprocessing.Value返回Process的结果，如：

def worker_email(keyword, func_result):
    sleep(10)
    print('Yo:' + keyword)
    func_result.value = 1
    return True

func_result = Value('i', 0)
p = Process(target=worker_email, args=(doc['check_id'],func_result))
p.start()
# Change status
if func_result.value == 1:
    stream.update_one({'_id': doc['_id']}, {"$set": {"status": True}}, upsert=False)

但是没有.join（）就行不通。任何想法如何使其工作或类似的方式？ :)

答案

如果你只是删除行p.join()它应该工作。如果你想在进一步执行之前等待进程完成，你只需要p.join。在程序结束时，python等待所有进程在关闭之前完成，因此您不必担心这一点。

另一答案

解决了通过将结果检查和状态更新转换为工作函数来获取处理结果的问题。就像是：

# Update task status if work is done
def update_status(task_id, func_result):
    # Connect to DB
    client = MongoClient('mongodb://localhost:27017/')
    db = client.admetric
    stream = db.stream

    # Update task status if OK
    if func_result:
        stream.update_one({'_id': task_id}, {"$set": {"status": True}}, upsert=False)

    # Close DB connection
    client.close()

# Do work
def yo_func(keyword):
    sleep(10)
    print('Yo:' + keyword)
    return True

# Worker function
def worker_email(keyword, task_id):
    update_status(task_id, yo_func(keyword))