ZeroMQ XPub/XSub模式

Motivation

相对于XPub/XSub模式，我们很容易想到Pub/Sub模式，即订阅发布模式。当我们使用ZeroMQ创建一个包含订阅发布模式的系统时，我们通常创建一个消息的发布者，即Publisher，和若干个消息订阅者(Subscriber)。消息的发布者绑定端口，订阅者通过发布者的IP和端口连接发布者，并且注册消息主题(Topic)，然后进行接收匹配主题的消息。整体结构如下图：

在订阅发布模式下，订阅者可以动态加入，随时连接消息的发布者，然后接收消息。但是，在这种结构中，如果有新的Publisher加入，那么所有订阅者都需要连接到这个Publisher上。如果系统中有成百上千的订阅者，每一个新的Publisher的加入都会给系统造成很大的操作成本，这显然限制了系统规模。要解决这个问题，也很简单，就像只有一个发布者情况，所有的订阅者都只与这一个消息发布者交互，不管是Publisher内部发生什么变化，Subscriber都可以动态感知这种变化。所以很容易我们可以想到创建一个中间件来解耦Publishers和Subscribers，所有Subscriber都只与这一个中间件交互，换句话说，这个中间件从很多个Publisher那里接收消息，然后转发给Subscibers。事实上，有了这个中间件，我们可以做很多Pub/Sub模式做不了的事情，比如说对传送过程中的消息进行管理，重构，或者对系统进行负载均衡等等。我们把这个中间件称为Broker，上面说的这种模式，我们称之为XPub/XSub模式。

XPub/XSub

在XPub/XSub模式中，对Publisher来说由Pub/Sub模式中的bind操作变成了connect操作，connect的对象为Broker中的XSub端口。对Subscriber而言，和Publisher的操作一样，只不过connect的是Broker的XPub端口。在Broker中我们绑定XSub和XPub这两个端口。Proxy的作用即为中转消息，在ZMQ的API中提供了zmq.proxy方法来中转消息，其实Proxy就是一个代码块，在这个代码块中可以做任何我们想做的操作。后面会介绍一个简单的例子。从XPub/XSub这个模式中，我们可以发现，不管是Publisher还是Subscriber，它们的加入和离开都可以被系统动态发现。

XPub/XSub例子

在这个例子中，我们让Publisher从CSV文件中读取数据，在Broker中维护一个buffer，如果有Subscriber加入，我们首先发送缓冲区的历史消息，然后转发新消息给Subscribers。

封装ZMQ API

# -*- coding: utf-8 -*-
# utl.py
import zmqdef get_publisher(address, port):context = zmq.Context()socket = context.socket(zmq.PUB)connect_addr = 'tcp://%s:%s' % (address, port)socket.connect(connect_addr)return socketdef get_subscriber(address, port, topics):# Subscriber can register one more topics oncecontext = zmq.Context()socket = context.socket(zmq.SUB)connect_addr = 'tcp://%s:%s' % (address, port)socket.connect(connect_addr)if isinstance(topics, str):socket.subscribe(topics)elif isinstance(topics, list):[socket.subscribe(topic) for topic in topics]return socketdef get_broker(xsub_port, xpub_port):context = zmq.Context()xsub_socket = context.socket(zmq.XSUB)xsub_addr = 'tcp://*:%s' % xsub_portxsub_socket.bind(xsub_addr)# make xsub receive any messagexsub_socket.send(b'\x01')xpub_addr = 'tcp://*:%s' % xpub_portxpub_socket = context.socket(zmq.XPUB)xpub_socket.bind(xpub_addr)# make xpub receive verbose messagesxpub_socket.setsockopt(zmq.XPUB_VERBOSE, 1)# zmq.proxy(xsub_socket, xpub_socket)return xsub_socket, xpub_socket

Publisher类

# -*- coding: utf-8 -*-
# Publisher.pyimport csv
import time
import argparse
import utlclass Publisher(object):def __init__(self, topic, broker_address, broker_port, data, rate):''':param topic: the topic associated with messages:param broker_address: broker public IP:param broker_port: XSub port number:param data: csv file path:param rate: publishing rate, unit is second'''self.topic = topicself.pub_socket = utl.get_publisher(broker_address, broker_port)self.data = dataself.rate = ratedef publish_data(self):with open(self.data, newline='') as csv_file:reader = csv.reader(csv_file, delimiter=',')for row in reader:row.insert(0, self.topic)record = ','.join(row)self.pub_socket.send_string(record)print('[Publisher] Published message: %s' % record)time.sleep(self.rate)if __name__ == '__main__':parser = argparse.ArgumentParser()parser.add_argument('-t', '--topic', type=str, help='Topic')parser.add_argument('-a', '--address', type=str, help='Broker public IP address')parser.add_argument('-p', '--port', type=str, help='Broker XSub port number')parser.add_argument('-f', '--file', type=str, help='Data file path')parser.add_argument('-r', '--rate', type=int, help='Publishing rate in second')args = parser.parse_args()pub = Publisher(args.topic, args.address, args.port, args.file, args.rate)pub.publish_data()

Broker类

# -*- coding: utf-8 -*-
# Broker.pyimport sys
import time
import utl
import zmqclass Broker(object):def __init__(self, xsub_port, xpub_port):self.xsub_socket, self.xpub_socket = utl.get_broker(xsub_port, xpub_port)self.poller = zmq.Poller()self.poller.register(socket=self.xpub_socket, flags=zmq.POLLIN)self.poller.register(socket=self.xsub_socket, flags=zmq.POLLIN)self.buffer = {}def update_buffer(self, msg):topic = msg.split(',')[0]if topic in self.buffer:self.buffer[topic].append(msg)else:self.buffer.update({topic: [msg]})def handler(self):while True:events = dict(self.poller.poll(1000))# events from publishersif self.xsub_socket in events:msg = self.xsub_socket.recv_string()self.xpub_socket.send_string(msg)print('[Broker] Forwarded message: %s' % msg)self.update_buffer(msg)# events from subscribersif self.xpub_socket in events:topic = ''.join(list(self.xpub_socket.recv_string())[1:])if topic in self.buffer:# send history messages[self.xpub_socket.send_string(item) for item in self.buffer[topic]]else:self.xsub_socket.send_string(topic)if __name__ == '__main__':# The 1st argument is XSub port number, the 2nd is XPub port numberbroker = Broker(sys.argv[1], sys.argv[2])broker.handler()

Subscriber类

# -*- coding: utf-8 -*-
# Subscriber.pyimport argparse
import utlclass Subscriber(object):def __init__(self, broker_address, broker_port, topics):self.topics = topicsself.socket = utl.get_subscriber(broker_address, broker_port, topics)def subscribe(self):while True:msg = self.socket.recv_string()print('[Subscriber] Received message: %s' % msg)if __name__ == '__main__':parser = argparse.ArgumentParser()parser.add_argument('-t', '--topics', type=str, help='Topics separated by comma')parser.add_argument('-a', '--address', type=str, help='Broker address')parser.add_argument('-p', '--port', type=str, help='Broker port number')args = parser.parse_args()topics = args.topics.split(',')sub = Subscriber(args.address, args.port, topics)sub.subscribe()