我正在尝试使用 ZeroMQ 的 pub-sub 套接字。zmq::context_t
但是,在创建套接字 () 时,我并不清楚上下文 () 的作用zmq::socket_t
。
假设我要创建 5 个订阅者套接字(zmq::socket_t
using ZMQ_SUB
),我是否需要 5 个上下文,每个订阅者套接字一个?或者我可以为所有 5 个套接字使用一个上下文吗?
假设我要创建 5 个订阅者套接字 (
zmq::socket_t
using
ZMQ_SUB
),我是否需要 5 个上下文,每个订阅者套接字一个?或者我可以为所有 5 个套接字使用一个上下文吗?
Context
对于这个轻量级用例,您只需要一个实例。检查下面附加的说明 0MQ 上下文使用的文档部分,以及我在本文末尾为您制作的示例。
ZeroMQ 应用程序总是从创建上下文开始,然后使用它来创建套接字。在 C 中,这是
zmq_ctx_new()
调用。你应该在你的过程中创建和使用一个上下文。从技术上讲,上下文是单个进程中所有套接字的容器,并充当inproc
套接字的传输,这是在一个进程中连接线程的最快方式。如果在运行时一个进程有两个上下文,它们就像单独的 ZeroMQ 实例。
我在下面为您做了一个示例,以帮助您理解ZMQ context
和ZMQ PUB-SUB
模式。只要您有 5 个生产服务,就可以创建 5 个订阅者套接字。但是,如果您有一个源发布通知,我建议您使用套接字PUB-SUB
的模式和过滤属性。ZMQ SUB
您可以在我的代码中检查如何在下面进行设置,以便在 和 之间进行publisher #1
通信subscriber
。
Publisher #1 发送温度和湿度更新。
import zmq
from time import sleep
# Server socket
context = zmq.Context()
socket = context.socket( zmq.PUB )
socket.bind( "tcp://*:5556" )
while True:
socket.send_multipart( [ "TEMP", "25.40" ] )
socket.send_multipart( [ "HUMD", "48.90" ] )
sleep( 1 )
Publisher #2 发送压力更新..
import zmq
from time import sleep
# Server socket
context = zmq.Context()
socket2 = context.socket( zmq.PUB )
socket2.bind( "tcp://*:5557" )
while True:
socket2.send_multipart( [ "PRSS", "10000.00" ] )
sleep( 1 )
订阅者在两个不同的服务器上注册了温度、湿度和压力更新。
import zmq
from time import sleep
# Sockets to talk to servers
context = zmq.Context()
socket = context.socket( zmq.SUB )
socket.connect( "tcp://localhost:5556" )
socket2 = context.socket( zmq.SUB )
socket2.connect( "tcp://localhost:5557" )
# Set filters
socket.setsockopt_string( zmq.SUBSCRIBE, "TEMP".decode( 'ascii' ) )
socket.setsockopt_string( zmq.SUBSCRIBE, "HUMD".decode( 'ascii' ) )
socket2.setsockopt_string( zmq.SUBSCRIBE, "PRSS".decode( 'ascii' ) )
poller = zmq.Poller()
poller.register( socket, zmq.POLLIN )
poller.register( socket2, zmq.POLLIN )
while True:
socks = dict( poller.poll() )
if socket in socks and socks[socket] == zmq.POLLIN:
[ measurement, value ] = socket.recv_multipart()
print measurement
print value
if socket2 in socks and socks[socket2] == zmq.POLLIN:
[ measurement, value ] = socket2.recv_multipart()
print measurement
print value
sleep( 1 )