为 Python 脚本创建共享消息流的最佳方法是什么？

Question

我想做的事：我需要一个简单的消息流，这样一些脚本可以在那里发送结果，而另一个脚本可以获取结果并异步执行一些工作。

主要问题：我想看看发生了什么，所以如果出现问题 - 我可以快速修复它。我尝试使用 Celery+RabbitMQ（可以看到带 args 的工人，使用 Flower，但调度太复杂）和 multiprocessing.Queue（简单，但看不到带 args 的工人）。

---

我做了什么：我尝试构建类似的东西，使用 MongoDB 封顶集合并使用多个进程运行 Popen，以做出反应。一些脚本将 smth 写入集合，下面的脚本会监视它，如果满足某些条件 - 运行另一个脚本。

主要问题： subprocess.Popen() 从 multiprocessing.Process() 内部的使用看起来不自然（仍然有效），所以我试图找到更好或更稳定的解决方案： )

---

监听脚本：

from pymongo import MongoClient, CursorType
from time import sleep
from datetime import datetime

from multiprocessing import Process
import subprocess

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

def worker_checker(keyword):
     subprocess.Popen(["python", "worker_checker.py", str(keyword)])

if __name__ == '__main__':

    #DB connect
    client = MongoClient('mongodb://localhost:27017/')
    db = client.admetric
    coll = db.my_collection
    cursor = coll.find(cursor_type = CursorType.TAILABLE_AWAIT)

    #Script start UTC time
    utc_run = datetime.utcnow()

    while cursor.alive:
        try:
            doc = cursor.next()
            #Print doc name/args to see in command line, while Listener runs
            print(doc)
            #Filter docs without 'created' data
            if 'created' in doc.keys():
                #Ignore docs older than script
                if doc['created'] > utc_run:
                    #Filter docs without 'type' data
                    if 'type' in doc.keys():
                        #Check type
                        if doc['type'] == 'send_email':
                            #Create process and run external script
                            p = Process(target=worker_email, args=(doc['message'],))
                            p.start()
                            p.join()
                        #Check type
                        elif doc['type'] == 'check_data':
                            #Create process and run external script
                            p = Process(target=worker_checker, args=(doc['message'],))
                            p.start()
                            p.join()
        except StopIteration:
            sleep(1)

Answer 1

只要您可以控制worker_email 和worker_checker 逻辑，就不需要在单独的解释器中执行。

只需在两个模块中公开一个入口点并通过multiprocessing.Process 运行它们。

worker_email.py

def email_job(message):
    # start processing the message here

worker_checker.py

def check_job(message):
    # start checking the message here

listener_script.py

# you are not going to pollute the listener namespace
# as the only names you import are the entry points of the scripts
# therefore, encapsulation is preserved
from worker_email import email_job
from worker_checker import check_job

email_process = Process(target=email_job, args=[message])
check_process = Process(target=check_job, args=[message])

如果您无法从工作模块公开入口点，则只需运行 subprocess.Popen。将它们包装在 Process 中没有任何好处。