pyzmq 用自己的套接字创建一个进程

Question

我有一些代码正在监视其他一些更改的文件，我想做的是启动使用 zeromq 和不同套接字的代码，我现在这样做的方式似乎导致断言在 libzmq 的某个地方失败，因为我可能正在重用同一个套接字。当我从监视器类创建一个新进程时，如何确保上下文不会被重用？这就是我认为正在发生的事情，如果你能告诉我还有其他一些愚蠢，请指教。 这是一些代码：

import zmq
from zmq.eventloop import ioloop
from  zmq.eventloop.zmqstream  import ZMQStream
class Monitor(object):
    def __init(self)
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.DEALER)
        self.socket.connect("tcp//127.0.0.1:5055")
        self.stream =  ZMQStream(self._socket)
        self.stream.on_recv(self.somefunc)

    def initialize(self,id)
        self._id = id

    def somefunc(self, something)
        """work here and send back results if any """
        import json
        jdecoded = json.loads(something)
        if self_id == jdecoded['_id']
           """ good im the right monitor for you """
           work = jdecoded['message']
           results  = algorithm (work)
           self.socket.send(json.dumps(results))
        else:
           """let some other process deal with it, not mine """
           pass

 class Prefect(object):
    def __init(self, id)
        self.context = zmq.Context()
        self.socket = self.context.socket(zmq.DEALER)
        self.socket.bind("tcp//127.0.0.1:5055")
        self.stream =  ZMQStream(self._socket)
        self.stream.on_recv(self.check_if)
        self._id = id
        self.monitors = []
    def check_if(self,message):
        """find out from message's id whether we have
            started a proces for it previously"""
        import json
        jdecoded = json.loads(message)
        this_id = jdecoded['_id']
        if this_id in self.monitors:
            pass
        else:
            """start new process for it should have its won socket """
            new = Monitor()
            import Process
            newp = Process(target=new.initialize,args=(this_id) )
            newp.start()
            self.monitors.append(this_id) ## ensure its remembered

发生了什么是我想要所有的监控进程和一个单一的进程监听同一个端口，所以当prefect看到一个它没有看到的请求时，它会为它启动一个进程，所有存在的进程可能应该也听，但忽略不适合他们的信息。 就目前而言，如果我这样做，我可能会遇到一些崩溃，这可能与某些东西对同一个 zmq 套接字的并发访问有关（我尝试过 threading.thread，仍然崩溃）我在某处读到不同线程对 zmq 套接字的并发访问是不可能的.我将如何确保新进程获得自己的 zmq 套接字？

编辑： 我的应用程序中的主要交易是请求通过 zmq 套接字进入，并且正在侦听的进程通过以下方式对消息作出反应：

1. If its directed at that process judged by the _id field, do some reading on a file and reply since one of the monitors match the messages _id, if none match, then:
2  If the messages _id files is not recognized, all monitors ignore it but the Prefect creates a process to handle that _id and all future messages to that id.
3. I want all the messages to be seen by the monitor processes as well as the prefect process, seems that seems easiest, 
4. All the messages are very small, avarage ~4096 bytes.   
5. The monitor does some non-blocking read and for each ioloop it sends what it has found out

more-edit=>现在 prefect 进程绑定，它将接收消息并回显它们，以便监视器可以看到它们。这就是我的想法，作为架构，但它不是最终的。 .

所有消息都是通过浏览器从远程用户到达的，这让服务器知道客户端想要什么，服务器通过 zmq 将消息发送到后端（我没有显示这个，但并不难）所以在生产中他们可能无法绑定/连接到本地主机。 我选择了 DEALER，因为它允许任意方向的异步/无限消息（参见第 5 点。）并且 DEALER 可以与 DEALER 绑定，并且初始请求/回复可以从任何一方到达。另一个可以做到这一点的可能是 DEALER/ROUTER。

Answer 1

您是正确的，您不能在分叉边界上继续使用相同的套接字（多处理使用分叉）。一般来说，这意味着您不希望在子进程启动之前创建将在分叉进程中使用的套接字。 由于在您的情况下，套接字是 Monitor 对象上的一个属性，因此您根本不想在主进程中创建 Monitor。看起来像这样：

def start_monitor(this_id):
    monitor = Monitor()
    monitor.initialize(this_id)
    # run the eventloop, or this will return immediately and destroy the monitor

... inside Prefect.check_if():

    proc = Process(target=start_monitor, args=(this_id,))
    proc.start()
    self.monitors.append(this_id)

而不是您的示例，子进程所做的唯一事情就是分配一个 ID，然后终止该进程，最终没有任何效果。