Python多进程处理技术

Question

这个问题更多是关于事实调查和思考过程，而不是代码导向。

我有很多编译后的C++程序需要在不同的时间和参数下运行。我正在考虑使用Python multiprocessing从作业队列（rabbitmq）中读取作业，然后将该作业提供给要运行的C++程序（可能是子进程）。我选择multiprocessing模块，因为这将在双路Xeon服务器上运行，所以我想充分利用我的服务器的多处理器能力。

Python程序将是中央管理器，并简单地从队列中读取作业，使用适当的C++程序生成一个进程（或子进程）来运行该作业，获取结果（子进程的stdout和stderr），将其馈送给回调，然后把进程放回等待下一个作业运行的队列中。

首先，这听起来像一个有效的策略吗？

第二，是否有类似于此的示例？

谢谢您的帮助。

- FFredrixson

---

你是想同时运行多个C++程序，还是必须串行运行，下一次运行的参数取决于当前运行的输出？ - unutbu

通常我希望同时运行多个C++程序，彼此之间没有依赖关系。有时候，一个不同的C++程序在完成后需要按顺序运行后续程序。 - FFredrixson

Answer 1

Python程序将是中央管理器，只需从队列中读取作业，并使用适当的C++程序生成进程（或子进程？）来运行作业，获取结果（子进程stdout和stderr），将其馈送到回调函数中，并将进程放回等待下一个作业运行的进程队列中。你不需要使用multiprocessing模块。multiprocessing模块适用于将Python函数作为单独的进程运行。要运行C++程序并从stdout读取结果，您只需要使用subprocess模块。队列可以是列表，Python程序只需在列表非空时循环即可。但是，如果您想要：生成多个工作进程、它们从共同的队列中读取、使用队列参数并行生成C++程序、使用C++程序的输出将新项放入队列，则可以像这样使用multiprocessing：test.py

import multiprocessing as mp
import subprocess
import shlex

def worker(q):
    while True:
        # Get an argument from the queue
        x=q.get()

        # You might change this to run your C++ program
        proc=subprocess.Popen(
            shlex.split('test2.py {x}'.format(x=x)),stdout=subprocess.PIPE)
        out,err=proc.communicate()

        print('{name}: using argument {x} outputs {o}'.format(
            x=x,name=mp.current_process().name,o=out))

        q.task_done()

        # Put a new argument into the queue
        q.put(int(out))

def main():
    q=mp.JoinableQueue()

    # Put some initial values into the queue
    for t in range(1,3):
        q.put(t)

    # Create and start a pool of worker processes
    for i in range(3):
        p=mp.Process(target=worker, args=(q,))
        p.daemon=True
        p.start()
    q.join()
    print "Finished!"

if __name__=='__main__':
    main()

test2.py（一个简单的替代你的C++程序）：

import time
import sys

x=int(sys.argv[1])
time.sleep(0.5)
print(x+3)

运行test.py可能会得到以下结果：

Process-1: using argument 1 outputs 4
Process-3: using argument 3 outputs 6
Process-2: using argument 2 outputs 5
Process-3: using argument 6 outputs 9
Process-1: using argument 4 outputs 7
Process-2: using argument 5 outputs 8
Process-3: using argument 9 outputs 12
Process-1: using argument 7 outputs 10
Process-2: using argument 8 outputs 11
Process-1: using argument 10 outputs 13

请注意右侧列中的数字被反馈到队列中，并且（最终）作为 test2.py 的参数使用，出现在左侧列中。

- unutbu

Answer 2

首先，这个策略听起来合理吗？

是的。

其次，有没有类似于这个的例子？ Celery。

- S.Lott

Answer 3

听起来是个不错的策略，但你不需要使用 multiprocessing 模块，而是需要使用 subprocess 模块。 subprocess 用于从 Python 程序中运行子进程并与其交互（stdio、stdout、pipes 等），而 multiprocessing 更多地是通过并行处理来分发 Python 代码 以获得更好的性能。

根据响应策略，你也可以考虑使用 threading 来从线程中启动子进程。这将允许你在等待一个子进程的同时仍然对队列保持响应，以接受其他作业。

- Eli Bendersky