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 zmq

def get_publisher(address, port):
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    connect_addr = 'tcp://%s:%s' % (address, port)
    socket.connect(connect_addr)
    return socket

def get_subscriber(address, port, topics):
	# Subscriber can register one more topics once
    context = 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 socket

def get_broker(xsub_port, xpub_port):
    context = zmq.Context()

    xsub_socket = context.socket(zmq.XSUB)
    xsub_addr = 'tcp://*:%s' % xsub_port
    xsub_socket.bind(xsub_addr)
    # make xsub receive any message
    xsub_socket.send(b'\x01')

    xpub_addr = 'tcp://*:%s' % xpub_port
    xpub_socket = context.socket(zmq.XPUB)
    xpub_socket.bind(xpub_addr)
    # make xpub receive verbose messages
    xpub_socket.setsockopt(zmq.XPUB_VERBOSE, 1)

    # zmq.proxy(xsub_socket, xpub_socket)
    return xsub_socket, xpub_socket

Publisher类

# -*- coding: utf-8 -*-
# Publisher.py

import csv
import time
import argparse
import utl

class 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 = topic
        self.pub_socket = utl.get_publisher(broker_address, broker_port)
        self.data = data
        self.rate = rate

    def 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.py

import sys
import time
import utl
import zmq


class 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 publishers
            if 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 subscribers
            if 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 number
    broker = Broker(sys.argv[1], sys.argv[2])
    broker.handler()

Subscriber类

# -*- coding: utf-8 -*-
# Subscriber.py

import argparse
import utl

class Subscriber(object):
    def __init__(self, broker_address, broker_port, topics):
        self.topics = topics
        self.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()