python之如何使用多处理实现发布/订阅模式

有没有什么方法可以使用multiprocessing 数据结构来实现发布/订阅模式？换句话说，我想要一个类似于队列的东西，除了发布者可以同时向多个工作人员发送一个命令。

请您参考如下方法:

您可以使用 multiprocessing.Queue 的包装器创建自己的数据结构来实现简单的发布/订阅模式:

import os 
import multiprocessing 
from functools import wraps 
 
 
def ensure_parent(func): 
    @wraps(func) 
    def inner(self, *args, **kwargs): 
        if os.getpid() != self._creator_pid: 
            raise RuntimeError("{} can only be called in the " 
                               "parent.".format(func.__name__)) 
        return func(self, *args, **kwargs) 
    return inner 
 
class PublishQueue(object): 
    def __init__(self): 
        self._queues = [] 
        self._creator_pid = os.getpid() 
 
    def __getstate__(self): 
        self_dict = self.__dict__ 
        self_dict['_queues'] = [] 
        return self_dict 
 
    def __setstate__(self, state): 
        self.__dict__.update(state) 
 
    @ensure_parent 
    def register(self): 
        q = multiprocessing.Queue() 
        self._queues.append(q) 
        return q 
 
    @ensure_parent 
    def publish(self, val): 
        for q in self._queues: 
            q.put(val) 
 
def worker(q): 
    for item in iter(q.get, None): 
        print("got item {} in process {}".format(item, os.getpid())) 
 
if __name__ == "__main__": 
    q = PublishQueue() 
    processes = [] 
    for _ in range(3): 
        p = multiprocessing.Process(target=worker, args=(q.register(),)) 
        p.start() 
        processes.append(p) 
    q.publish('1') 
    q.publish(2) 
    q.publish(None)  # Shut down workers 
 
    for p in processes: 
        p.join() 

输出:

got item 1 in process 4383 
got item 2 in process 4383 
got item 1 in process 4381 
got item 2 in process 4381 
got item 1 in process 4382 
got item 2 in process 4382 

只要父进程是唯一进行发布的进程，这种模式就可以很好地工作，并且您为父进程中的每个工作进程注册一个订阅队列，然后使用其 将该订阅队列传递给工作进程multiprocessing.Process 构造函数。这些限制是由于 multiprocessing.Queue 是 unpicklable 造成的。如果您需要将订阅队列传递给已经在运行的工作线程，则需要调整实现以使用 multiprocessing.Manager.Queue 代替。