在 multiprocessing.Process 实例之间传递方法返回值

Question

如何从 multiprocessing.Process 的另一个实例中的方法获取返回值？

我有两个文件：

文件 hwmgr.py：

import multiprocessing as mp
from setproctitle import setproctitle
import smbus
import myLoggingModule as log

class HWManager(mp.Process):
    def __init__(self):
        mp.Process.__init__(self, cmd_q, res_q)
        self.i2c_lock = mp.Lock()
        self.commandQueue = cmd_q
        self.responseQueue = res_q
    def run(self):
        setproctitle('hwmgr')
        while True:
            cmd, args = self.commandQueue.get()
            if cmd is None: self.terminate()
            method = self.__getattribute__(cmd)
            result = method(**args)
            if result is not None:
                self.responseQueue.put(result)

    def get_voltage(self):
        with self.i2c_lock:
            # ...do i2c stuff to get a voltage with smbus module
        return voltage

文件 main.py：

import multiprocessing as mp
import hwmgr

cmd_q = mp.Queue()
res_q = mp.Queue()

hwm = hwmgr.HWManager(cmd_q, res_q)
hwm.start()

cmd_q.put(('get_voltage', {}))
battery = res_q.get()

print battery

虽然此解决方案有效，但 HWManager 进程的复杂性在未来可能会增加，并且使用相同机制的 main.py（代码已简化）衍生出其他进程。错误的进程显然有可能从它的res_q.get() 命令中得到错误的返回数据。

什么是更稳健的方法？
（我试图避免为每个其他进程返回一个 mp.Queue - 因为这需要每次都重新设计 HWManager 类以容纳额外的队列）

OK - WIP 代码如下：

hwmgr.py:

import multiprocessing as mp
from multiprocessing.connection import Listener
from setproctitle import setproctitle
import smbus

class HWManager(mp.Process):
    def __init__(self):
        mp.Process.__init__(self)
        self.i2c_lock = mp.Lock()

    def run(self):
        setproctitle('hwmgr')
        self.listener = Listener('/tmp/hw_sock', 'AF_UNIX')

        with self.i2c_lock:
            pass  # Set up I2C bus to take ADC readings

        while True:
            conn = self.listener.accept()
            cmd, args = conn.recv()

            if cmd is None: self.terminate()
            method = self.__getattribute__(cmd)
            result = method(**args)
            conn.send(result)

    def get_voltage(self):
        with self.i2c_lock:
            voltage = 12.0  # Actually, do i2c stuff to get a voltage with smbus module

        return voltage

文件client.py

import multiprocessing as mp
from multiprocessing.connection import Client
from setproctitle import setproctitle
from time import sleep

class HWClient(mp.Process):

def __init__(self):
    mp.Process.__init__(self)
    self.client = Client('/tmp/hw_sock', 'AF_UNIX')

def run(self):
    setproctitle('client')
    while True:
        self.client.send(('get_voltage', {}))
        battery = self.client.recv()
        print battery
        sleep(5)

main.py:

import hwmgr
import client

cl = client.HWClient()  # Put these lines here = one error (conn refused)
cl.start()
hwm = hwmgr.HWManager()
hwm.start()
# cl = client.HWClient()  # ...Or here, gives the other (in use)
# cl.start()

Answer 1

这听起来像是需要标准的客户端-服务器架构。您可以使用 UNIX 域套接字（或 Windows 上的命名管道）。多处理模块使得在进程之间传递 python 对象变得超级容易。服务器代码示例结构：

from multiprocessing.connection import Listener

listener = Listener('somefile', 'AF_UNIX')

queue = Queue()
def worker():
    while True:
        conn, cmd = queue.get()
        result = execute_cmd(cmd)
        conn.send(result)
        queue.task_done()


for i in range(num_worker_threads):
     t = Thread(target=worker)
     t.daemon = True
     t.start()


while True:
    conn = listener.accept()
    cmd = conn.recv()
    queue.put((conn, cmd)) # Do processing of the queue in another thread/process and write result to conn

客户端看起来像：

from multiprocessing.connection import Client
client = Client('somefile', 'AF_UNIX')

client.send(cmd)
result = client.recv()

上面的代码为工作线程使用线程，但您也可以使用多处理模块轻松为工作线程创建进程。详情请见docs。