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

Question

我需要使用多处理来运行多个背景异步函数。我有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：我正在使用多处理。值返回流程结果，例如：

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（）。有什么想法使它起作用或类似的方式？：）

Answer 1

如果仅删除该行p.join()，则应起作用。只有要等待该过程完成之前，才需要p.join。在程序结束时，Python等待所有过程在关闭之前完成，因此您无需担心。

Answer 2

通过将结果检查和状态更新转移到工作函数中来解决过程结果。类似：

# 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))