Python ZMQ响应者未收到消息

Question

我正在尝试一个简单的zmq 脚本，但不知何故响应者没有收到第一条消息。

响应器如下所示：

def main():
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.connect("tcp://localhost:{}".format(5560))
    print("connected ...")
    while True:
          #  Wait for next request from client
          message = socket.recv_pyobj()
          #print (message)
          print(message)
if __name__ == '__main__':
    main()

我正在从另一个进程发送带有以下代码的请求：

def main():
    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://localhost:{}".format(5560))
    print("sending object")
    socket.send_pyobj("ok")
    print("done")

if __name__ == '__main__':
    main()

有人知道为什么它没有到达吗？

Answer 1

问：有人知道为什么它没有到达吗？

哦，我当然知道了。

基于 ZeroMQ 的分布式计算系统的工作方式和工作方式有几个要点。

如果您不熟悉使用 ZeroMQ 或其他衍生工具 (nanomsg 等)，请务必不要错过 Pieter Hintjen 的必读书籍“Code Connected. Volume 1”。

有两个地方可能会错过（或主要无法送达）消息：

• 尚未准备好的进程，假定接收消息，是第一个
• 成功的.bind() 没有可用资源（端口），是第二个

localhost-only (vmci://-virtualised of internal port-abstracted network) 实验不会遇到严苛的网络传输条件相关问题

---

可治愈：

def main():
    context = zmq.Context()
    socket = context.socket( zmq.REQ )
    socket.setsockopt(       zmq.LINGER,    0 )              # ALWAYS PREVENT BLOCKING
    socket.setsockopt(       zmq.IMMEDIATE, 1 )              # BLOCK UNTIL CONN-READY
    #ocket.setsockpt(        zmq.ZMQ_HANDSHAKE_IVL, ... )    # IF TWEAKING NETWORK-WIDE
    # OR:
    # a stone-age wait for the other part get started in a one-computer demo:
    # sleep( 20 )
    # :o)
    socket.connect(         "tcp://localhost:{}".format( 5560 ) )
    print(                  "Will try to dispatch an object to Context() instance" )
    socket.send_pyobj(      "ok" )
    print(                  ".send() method has returned from a blocking-call mode" )
    ...
    #--------------------------------------------------------# ALWAYS
    socket.close()                                           # ALWAYS RELEASE RESOURCES
    context.term()                                           # ALWAYS RELEASE RESOURCES
    #                                                        # ALWAYS (not all versions
    #                                                        #         have "friendly"
    #                                                        #         defeaults to rely
    #                                                        #         on others,
    #                                                        #         so be explicit)
    #--------------------------------------------------------# ALWAYS

---

一方面，显然不一定是REP，但在这里它更适合，由于while，必须.bind()，其他（S）只是.connect()给一个已知的连接目标：

def main():
    context = zmq.Context()
    socket = context.socket( zmq.REP )
    socket.setsockopt(       zmq.LINGER,    0 )              # ALWAYS PREVENT BLOCKING
    socket.setsockopt(       zmq.IMMEDIATE, 1 )              # BLOCK UNTIL CONN-READY
    #ocket.setsockpt(        zmq.ZMQ_HANDSHAKE_IVL, ... )    # IF TWEAKING NETWORK-WIDE
    socket.bind(            "tcp://localhost:{}".format( 5560 ) )
    print(                  ".bind() ...")
    try:
        while True:             # Wait for next request from client:
              message = socket.recv_pyobj()
              print( message )
    except:
        print( "EXC'd" )
    finally:
        #----------------------------------------------------# ALWAYS
        socket.unbind( "tcp://localhost:{}".format( 5560 ) ) # ALWAYS RELEASE PORT
        socket.close()                                       # ALWAYS RELEASE RESOURCES
        context.term()                                       # ALWAYS RELEASE RESOURCES
        #                                                    # ALWAYS (not all versions
        #                                                    #         have "friendly"
        #                                                    #         defeaults to rely
        #                                                    #         on others,
        #                                                    #         so be explicit)
        #----------------------------------------------------# ALWAYS

---

最后但并非最不重要的一点是，这将开始工作，但由于可能遗漏了REQ/REP-behaviour 原型的原则，它将陷入无限等待。一个必须问（REQ.send()-s），另一个要回复的人必须听问题REP.recv()，但它也必须先回答... REP.send("something")我们可能会在两个人的两步探戈中前进，而 ASKER 也必须听到 REQ.recv() 的答案。

只有这样，ASKER 才能通过另一个 REQ.send() 发送另一个问题。

因此，您的发送 REQ-parts，但主要是接收 REP-part，在无限 while True:{...} 循环内必须进行修改，以便接收任何第二条和更多消息，即使在 @ 987654340@-s 一枪就死了，从不听REP的任何回答。

Answer 2

您必须在REP 代码 sn-p 中添加一个 IP 地址 .bind()，而不是 .connect()。在 REP/REQ pattern 中有请求和响应因子，因此您可以在响应者代码中提出反馈。因此您的代码将如下所示：

响应者：

import zmq

def main():
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://127.0.0.1:5560")
    while True:
        message = socket.recv_pyobj()
        print(message)
        socket.send_string('Your message received.')

if __name__ == '__main__':
    main()

---

请求代码sn-p：

import zmq

def main():
    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://localhost:5560")
    print("sending object")
    socket.send_pyobj("ok")
    message = socket.recv()
    print(message)

if __name__ == '__main__':
    main()

---

输出（请求）：

sending object
b'Your message received.'

出局（代表）：

ok

---

[注意]：

1234563 @ 在请求部分和 socket.recv() 而不是 socket.recv_pyobj() 在响应者中。

• 请注意，在.bind() 的情况下，您不应使用localhost 字符串^{Relevant Post}。