与 0mq 通信

Question

我正在尝试创建一个系统来收集发送的数据 我有两个小脚本，一个receiver.py，它应该接收数据，一个发送者应该发送它，目前我尝试使用1-1连接，但最后我需要多个发送者和一个接收者来处理传入的数据。我尝试使用 0mq 发布者/订阅者模式来完成此操作。

#receiver.py

def receive():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.setsockopt(zmq.SUBSCRIBE, 'Child:') 
    socket.bind('tcp://localhost:5000')
    while True:
        print 'Parent received: %s' % socket.recv()

receive()

#sender.py

def send(data):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.connect('tcp://localhost:5000')
    socket.send('Sender: %i' % data)
    socket.close()
    print "sent"

send(10)

当我启动receiver.py 时，它只是等待数据，当我运行sender.py 时什么也没有收到。非常感谢您的建议，实际上我什至不确定发布者/订阅者是否是我的方案的最佳模式（多个传感器通过本地网络将数据发送到一台服务器进行实时处理）。

Answer 1

如果您无法使示例正常工作，则说明您的 ZMQ 库安装存在问题，或者您的库版本与您使用的绑定不兼容，这是一个好兆头。从那里开始检查，并始终尝试先逐字运行示例，以确保所有内容至少可以与参考代码一起使用。

但是，我确实发现您的代码至少有一个问题会导致您永远不会收到消息。

在您的订阅者中，您订阅了“孩子：”，但在您的发布者中，您永远不会发送与该内容匹配的消息。执行此操作的“正确”方法是发送多帧消息，但为了简化代码，您还可以发送以主题开头的字符串，如下所示：

socket.send('Child: Sender: %i' % data)

或者，您可以更改您的订阅者以适应您当前的消息模式：

socket.setsockopt(zmq.SUBSCRIBE, 'Sender:')

最后但同样重要的是，如果您想订阅发布者可能发送的所有内容，您可以订阅空字符串：

socket.setsockopt(zmq.SUBSCRIBE, '')

...最后一个可能适合您的情况。因此，生成的代码如下：

#receiver.py

def receive():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.setsockopt(zmq.SUBSCRIBE, '') 
    socket.bind('tcp://localhost:5000')
    while True:
        print 'Parent received: %s' % socket.recv()

receive()

#sender.py

def send(data):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.connect('tcp://localhost:5000')
    socket.send('Sender: %i' % data)
    socket.close()
    print "sent"

send(10)

解决子点、问题和cmets：

• 在这种情况下，更适合bind() 的订阅者和connect() 的发布者（正如您所做的那样），ZMQ 并不关心您仅根据套接字类型使用哪个。您的订阅者是您的“服务器”，您的发布者是您的“客户”，所以 SUB 应该是 bind()，PUB 应该是 connect()。
• PUB/SUB 是适合您的场景的一种很好的模式，只要通信是严格单向的，就像您的情况一样。

Answer 2

为了理解发生了什么，我重写了它：

import zmq
import threading
import time

def receive():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.bind("tcp://127.0.0.1:5000")
    socket.setsockopt(zmq.SUBSCRIBE, '')
    while True:
        print 'Parent received: %s' % socket.recv()

threading.Thread(target=receive).start()

def send(data):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.connect("tcp://localhost:5000")
    while data:
        socket.send('Sender: %i' % data)
        data -= 1
        time.sleep(1)

输出：

>>> send(10)
Parent received: Sender: 9
Parent received: Sender: 8
Parent received: Sender: 7
Parent received: Sender: 6
Parent received: Sender: 5
Parent received: Sender: 4
Parent received: Sender: 3
Parent received: Sender: 2
Parent received: Sender: 1

重点是：

1. 如Why doesn't zeromq work on localhost? 中所述，您必须使用127.0.0.1 绑定您的本地主机
2. 发布者不公开名称，因此我们必须从订阅者中删除 Child: 或将其更改为 Sender: 或将 send() Sender: 更改为 Child:（我选择了第一个）李>
3. 也许您需要某种事件来说明订阅者的连接在发送数据之前处于活动状态（zeromq 是一个异步框架），否则您会丢失第一条消息。作为证明，您可以更改 send() 就像下面的示例一样，您不会丢失任何消息。

.

def send(data):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.connect("tcp://localhost:5000")
    while data:
        time.sleep(1)
        socket.send('Sender: %i' % data)
        data -= 1
    socket.close()

输出：

>>> send(10)
Parent received: Sender: 10
Parent received: Sender: 9
Parent received: Sender: 8
Parent received: Sender: 7
Parent received: Sender: 6
Parent received: Sender: 5
Parent received: Sender: 4
Parent received: Sender: 3
Parent received: Sender: 2
Parent received: Sender: 1