ZeroMQ 结构分析
0x01 ZeroMQ 项目总体结构
目前业界流行的分布式消息队列系统(或者可以叫做消息中间件)种类繁多,比如,基于Erlang的RabbitMQ、基于Java的ActiveMQ/Apache Kafka等等,都能进行大批量的消息路由转发。它们的共同特点是,都有一个消息中转路由节点,按照消息队列里面的专业术语,这个角色应该是broker。整个消息系统通过这个broker节点,进行从消息生产者Producer到消费者Consumer的消息路由。当然了,生产者和消费者可以是多对多的关系。消息路由的时候,可以根据关键字(专业的术语叫topic),进行关键字精确匹配、模糊匹配、广播方式的消息路由。
我们所写的极简消息队列由以下几个部分组成:
- Broker:简单来说就是消息队列服务器实体
- Producer:消息的生产者,主要用来发送消息给消费者
- Consumer:消息的消费者,主要用来接收生产者的消息
- Routing Key:路由关键字(Topic),主要用来控制生产者和消费者之间的发送与接收消息的对应关系
所写的 ZeroMQ 项目具体的特性如下:
- ZeroMQ 基于 Java 语言进行编写,网络通讯依赖 Netty
- 生产者和消费者的关系可以是一对多、多对一、多对多的关系。若干个消费者可以组成消费者集群,生产者可以向这个消费者集群投递消息。
- 消费者集群对于有共同关注点的消费者支持消息的负载均衡策略。
- 目前仅仅支持关键字的精确匹配路由,后续会逐渐完善。
- 消息队列服务器 Broker 基于 Netty 的主从事件线程池模型开发设计。
- 网络消息序列化采用 Kryo 进行消息的网络序列化传输。
- Broker 的消息派发、负载均衡、应答处理(ACK)基于异步多线程模型进行开发设计。
- Broker 消息的投递,目前支持严格的消息顺序。其中 Broker 还支持消息的缓冲派发,即 Broker 会缓存一定数量的消息之后,再批量分配给对此消息感兴趣的消费者。
具体的结构图如下:
![ZeroMQ 结构](https://user-images.githubusercontent.com/15026221/202644070-d2eddbd3-5bd4-4e16-8fff-ca6c3f92ffb1.png)
在 ZeroMQ 消息队列中,传递的消息种类有以下几种:
- Subscribe:Subscribe 消息是 Consumer 发送给 Broker 端的,用来说明 Consumer 订阅的消息主题。
- Subscribe Ack:Broker 在接收到 Consumer 发送的 Subscribe 消息并且进行处理之后,会返回一个 Subscribe Ack 消息给 Consumer。
- Unsubscribe:Consumer 发送 Unsubscribe 消息给 Broker 端的,用来说明 Consumer 取消订阅的主题
- Unsubscribe Ack:Broker 在接收到 Consumer 发送的 Unsubscribe 消息并且进行处理之后,会返回一个 Unsubscribe Ack 消息给 Consumer。
- Message:Producer 发送 Message 消息给 Broker,然后 Broker 会把这个 Message 消息转发给 Consumer 端
- Producer Ack:Broker 在接收到 Producer 端的 Message 之后,会返回一个 Producer Ack 消息给 Producer
- Consumer Ack:Consumer 在接收到 Broker 转发的 Message 之后,就会返回一个 Consumer Ack 消息
接下来分别对 Consumer、Producer、Broker 这三个模块进行讲解。
0x02 Consumer
ZeroMQ 中使用 Consumer 订阅并且等待接收消息的例子如下:
public class ZeroMQConsumer1 {
// 由用户自定义,对 producer 发送过来的消息进行处理
private static MessageConsumeHook hook = new MessageConsumeHook() {
public ConsumerAckMessage consumeMessage(Message message) {
System.out.printf("ZeroMQConsumer1 收到消息编号:%s,消息内容:%s\n", message.getMsgId(), new String(message.getBody()));
// 返回 ConsumerAckMessage 消息给 broker 服务端
ConsumerAckMessage result = new ConsumerAckMessage();
// 设置消息消费结果为 SUCCESS
result.setStatus(ConsumerAckMessage.SUCCESS);
return result;
}
};
public static void main(String[] args) {
ZeroMQConsumer consumer = new ZeroMQConsumer("127.0.0.1:18888", "ZeroMQ-Topic-1", hook);
consumer.setClusterId("ZeroMQCluster1");
consumer.start();
}
}
上面的 MessageConsumeHook 是一个接口,具体由 Consumer 端实现,表明 Consumer 对 Broker 端发送过来的 Message 的处理方式。在上面,只是简单地打印发送过来的消息信息,然后返回 ConsumerAckMessage 响应给 Broker。在实现了 MessageConsumeHook 接口之后,就会创建一个 Consumer 对象。创建时,要指定 Broker 端的地址,以及要订阅的主题,最后还有 Consumer 实现的 MessageConsumerHook 接口对象。最后调用 start 方法启动 Consumer。start 方法主要完成的工作如下:
- 创建一个 ConsumerHandler 对象,并且将其设置到 pipeline 之中。ConsumerHandler 用来对 Broker 返回的消息进行处理。
- 连接到 Broker 服务器端
- 连接建立之后向 Broker 发送 Subscribe 消息,表明当前 Broker 订阅的主题,接着阻塞等待响应(这里使用同步发送的方式)
- 当 Broker 返回 SubscribeAckMessage 之后,会唤醒 Consumer 中阻塞的线程,根据 SubscribeAckMessage 中的 status 属性判断订阅主题是否成功打印相应的日志。
Consumer 端的 pipeline 网络结构如下所示:
![consumer pipeline](https://user-images.githubusercontent.com/15026221/202645992-c8506bb6-acdd-40e7-abcb-e197b8a46f42.png)
MessageObjectEncoder 和 MessageObjectDecoder 主要是对消息进行编解码和序列化,会在后面的章节进行讲解。ConsumerHandler 主要用来对 Broker 发送过来的消息进行处理。Broker 会返回 SubscribeAckMessage 消息以及 Message 消息,ConsumerHandler 中对这两种消息的处理都是封装成一个 Runnable 对象,然后放入到线程池中去处理,这样就不会阻塞 netty 的 io 线程:
- SubscribeAckMessage:获取到阻塞的 CallBackFuture 对象,会唤醒阻塞的 consumer 线程,并且如果订阅成功的话,打印日志信息;如果订阅失败的话,同样会打印失败的原因。
- Message:将 Broker 转发过来的 Message 交给 MessageConsumeHook 进行消费处理,并且向 Broker 返回一个响应 ConsumerAckMessage。
0x03 Producer
在 ZeroMQ 项目中使用 Producer 发送消息到 Broker 端的代码如下:
public class ZeroMQProducer1 {
public static void main(String[] args) throws InterruptedException {
ZeroMQProducer producer = new ZeroMQProducer("127.0.0.1:18888", "ZeroMQ-Topic-1");
producer.setClusterId("zeromq cluster");
producer.start();
System.out.println(StringUtils.center("zeromq producer1 消息发送开始", 50, "*"));
for (int i = 0; i < 10; i++) {
Message message = new Message();
String str = "hello zeromq from producer1[" + i + "]";
message.setBody(str.getBytes());
ProducerAckMessage result = producer.deliver(message);
if (result.getStatus() == (ProducerAckMessage.SUCCESS)) {
System.out.printf("zeromq producer1 发送消息编号:%s\n", result.getMsgId());
}
Thread.sleep(100);
}
producer.shutdown();
System.out.println(StringUtils.center("zeromq producer1 消息发送完毕", 50, "*"));
}
}
和 Consumer 一样,也是先创建一个 Producer 对象,并且在创建的时候指明 Broker 地址,要发送的消息主题,然后调用 start 方法启动 Producer 方法。Start 方法的流程如下:
- 创建一个 ProducerHandler 对象,并且将其设置到 pipeline 中,ProducerHandler 会对 Broker 返回的消息进行处理。
- 连接到 Broker 端
接着就是创建好 Message 消息对象,调用 deliver 方法将消息发送到 Broker 端,delive 方法的流程如下:
- 如果 producer 与 broker 之间没有建立连接或者连接已经关闭了,或者 producer 不处于运行状态,就会直接返回一个 ProducerAckMessage 对象,只不过将对象的 status 设置为 FAIL,表明发送失败。
- 将 Message 消息发送到 Broker 端,并且发送线程阻塞(使用同步发送方式),等到 Broker 端返回的响应信息 ProducerAckMessage,然后打印相应日志
Producer 端的 pipeline 网络结构如下:
![producer pipeline](https://user-images.githubusercontent.com/15026221/202646029-e2bd72cb-aac5-43e3-a544-565dbccd0977.png)
Broker 返回给 producer 的消息 ProducerAckMessage 会被 ProducerHandler 进行处理。
0x04 Broker
Broker 端的逻辑比较复杂,ZeroMQServer 启动的时候会创建以下三个任务提交到线程池中不断执行:
- SendMessageController:SendMessageController 不断循环从 MessageTaskQueue 中取出 MessageDispatchTask(MessageDispatchTask 是对 message, clusterId, topic 这三个变量的封装),这里取出 MessageDispatchTask 中的 message,然后将其转发给客户端 consumer,并且阻塞等待客户端返回 ConsumerAckMessage。注意,这里转发 Message 到 consumer 端是多个线程并发发送的,比如有 6 条消息,然后有 3 个线程,那么就由这 3 个线程并发地发送消息,每条线程发送 2 条消息。
- SendAckController:从 AckTaskQueue 中获取保存的 ProducerAckMessage,并且将其发送到 producer 端,表示 broker 已经收到消息。
- TransferAckController:在 BrokerProducerMessageStrategy 中会将 producer 发送过来的消息的 msgId 生成标识,保存到 AckMessageCache 中,另外如果 producer 发送过来的消息如果没有消费者进行订阅,就会跳过这一步,直接生成一个 ProducerAckMessage,保存到 AckTaskQueue 中。因此,这里 AckPushMessageController 的任务就是根据 AckMessageCache 保存的消息的标识,并发地创建 ProducerAckMessage,然后将其分发到 AckTaskQueue 中,等待 SendAckController 将其发送到 producer 端。比如有有 6 个标识,然后有 3 个线程,那么就由这 3 个线程来并发地创建 ProducerAckMessage,并且 ack 消息发送到 producer 端,每个线程处理 2 条标识。
ZeroMQServer 继承了 BrokerParallelServer,具体启动代码如下:
public class BrokerParallelServer implements RemotingServer {
// parallel 是核心线程数,也就是处理器的数目 * 2
protected int parallel = NettyClustersConfig.getWorkerThreads();
protected ExecutorService executor = Executors.newFixedThreadPool(parallel);
public BrokerParallelServer() {
}
public void start() {
for (int i = 0; i < parallel; i++) {
executor.submit(new SendMessageController());
executor.submit(new SendAckController());
executor.submit(new TransferAckController());
}
}
public void shutdown() {
executor.shutdown();
}
}
Broker 端的 pipeline 结构如下所示:
![broker pipeline](https://user-images.githubusercontent.com/15026221/202646049-d2c5e017-50e3-4a50-ba18-4cd07b48fda3.png)
Broker 会接受到 4 种类型的消息:Subscribe、Message、Unsubscribe、ConsumerAckMessage。然后会分别对其进行处理:
- Message:Producer 发送到 broker 的消息,由 ProducerStrategy 进行处理,处理流程如下:
- 获取消息的主题,并且得到订阅了该主题的消费者集群
- 如果没有消费者集群订阅该消息主题,那么直接创建一个 ProducerAckMessage,保存到 AckTaskQueue,然后返回。然后等待 SendAckController 从 AckTaskQueue 中取出 ProducerAckMessage,然后发送到 producer。
- 如果有消费者集群订阅该消息主题
- 为每一个订阅了该主题的消费者集群创建一个 MessageDispatchTask(一个消费者集群对应一个 task),然后将这些 task 保存到 MessageTaskQueue. 这个 MessageDispatchTask 会把消息分发给消费者集群中的一个消费者(注意,不是所有的消费者,只是其中一个)
- 根据这个消息的 msgId 创建一个标识,然后将其保存到 AckMessageCache 中,broker 会根据 AckMessageCache 中的标识并发地创建 ProducerAck,将其保存到 AckTaskQueue 中,最后将其分发到对应的 producer 端
- ConsumerAckMessage:consumer 接收到 broker 发送的 Message 消息之后,会返回 ConsumerAckMessage,由 ConsumerStrategy 进行处理。在 broker 使用 SendMessageLauncher 发送消息到 consumer 端时,会创建一个 CallBackFuture,然后阻塞。当 consumer 接收到发送过去的消息,并且进行了处理之后,就会返回一个 ConsumerAckMessage,然后在这里被接收到,然后唤醒前面阻塞在 CallBackFuture 中的线程。
- Subscribe:Consumer 发送给 Broker 的订阅消息,由 SubscribeStrategy 进行处理,会把 Consumer 的信息添加到 broker 中,具体如下:
- 如果 broker 没有此 consumer 对应的消费者集群,创建一个 ConsumeCluster:
- 把 <consumerId, channel> 连接信息保存到 channelMap 中,channel 是 consumer 端到 broker 端的连接
- 把 <topic, subscription> 保存到 subMap 中
- 把 channel 保存到 channelList 中
- 如果在消费者集群中找到了 consumerId 对应的消费者,删除这个消费者对应的连接,替换成新连接
- 如果在消费者集群中没有找到 consumerId 对应的消费者,则添加 consumerId 的相应信息到 ConsumerCluster 中
- Unsubscribe:Consumer 发送给 Broker 的取消订阅消息,由 UnsubscribeStrategy 进行处理。只是从 broker 端删除掉这个 consumer 的信息。
0x05 编解码
在 Consumer,Producer 以及 Broker 端的消息结构如下:
消息体分为两部分,第一部分是 4 字节的数据,表示后面消息数据的长度;第二部分就是发送的各种消息的字节数据。MessageObjectDecoder 和 MessageObjectEncoder 就是消息的编解码器,使用 kryo 作为序列化器:
- MessageObjectEncoder:首先使用 kryo 对消息对象进行序列化,并且获取到序列化后的消息长度(int 类型)。接着首先写入 int 类型的长度数据,再写入消息的字节数据。
- MessageObjectDecoder:首先判断获取到的字节数目是否小于 4,如果小于 4,直接返回。如果大于 4,读取 4 个字节的数据,表示消息数据的长度 len,然后判断可以获取到的字节数目是否小于 len,如果小于 len,那么同样直接返回,否则读取字节流,利用 kryo 反序列化成 Object 对象。
0x06 连接池
在 Consumer 和 producer 调用 start 方法启动的时候需要从连接池中获取和 broker 的连接,并且在 shutdown 之后需要归还连接到连接池中,这样就形成了连接的复用。连接池的对象为 GenericKeyedObjectPool<String, Connection>,其中 String 可以看成是 broker 的地址,Connection 可以看成是建立的到 broker 的连接。当 consumer 或者 producer 从 pool 中 borrow 连接时,如果 pool 中已经存在要连接的 broker 端的地址,那么就会直接从 pool 中获取到连接对象 Connection,如果没有 broker 端的地址的话,就会先创建一些 Connection 对象(默认创建 8 个),保存到连接池 pool 中,然后再去获取。