Netty构建分布式消息队列（AvatarMQ）设计指南之架构篇

　　目前业界流行的分布式消息队列系统（或者可以叫做消息中间件）种类繁多，比如，基于Erlang的RabbitMQ、基于Java的ActiveMQ/Apache Kafka、基于C/C++的ZeroMQ等等，都能进行大批量的消息路由转发。它们的共同特点是，都有一个消息中转路由节点，按照消息队列里面的专业术语，这个角色应该是broker。整个消息系统通过这个broker节点，进行从消息生产者Producer到消费者Consumer的消息路由。当然了，生产者和消费者可以是多对多的关系。消息路由的时候，可以根据关键字（专业的术语叫topic），进行关键字精确匹配、模糊匹配、广播方式的消息路由。

　　简单来说，一个极简的分布式消息队列系统主要的构成模块有：

　　Broker：简单来说就是消息队列服务器实体。

　　Producer：消息的生产者，主要用来发送消息给消费者。

　　Consumer：消息的消费者，主要用来接收生产者的消息。

　　Routing Key：路由关键字（Topic），主要用来控制生产者和消费者之间的发送与接收消息的对应关系。

　　Channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

　　到此为止，我们明白了一个分布式消息队列系统的主要构成模块，现在本人就通过Netty，这个优秀的Java NIO网络通讯框架，构建一个支持上述应用场景的分布式消息队列系统，本人把其命名为AvatarMQ。后续我会基于这个开源项目，连载出基于Netty构建分布式消息队列系统系列相关的文章，阐明主要的设计思路、组织结构、模块划分依据、类图结构等等。为了说明方便，后续本文中，如果没有特殊说明，有涉及基于Netty构建的分布式消息队列系统，就是指代AvatarMQ。由于整个开源项目涉及的代码量比较多，所以希望大家在本人编写系列博客文章的基础上，耐心地理解、分析其中的代码模块，相信一定不会让您失望，收获一定多多！

　　AvatarMQ基于Netty，所以首先，你要能清楚的理解Netty是什么？它能做什么？此外AvatarMQ还大量使用了Java多get='_blank'>线程的相关类库。所以希望在此之前，大家能回忆复习一下，这样理解起来会更加得心应手、事半功倍。

　　AvatarMQ是基于Netty构建的分布式消息队列系统，支持多个生产者和多个消费者之间的消息路由、传递。主要特性如下：

• AvatarMQ基于Java语言进行编写，网络通讯依赖Netty。
• 生产者和消费者的关系可以是一对多、多对一、多对多的关系。
• 若干个消费者可以组成消费者集群，生产者可以向这个消费者集群投递消息。
• 消费者集群对于有共同关注点的消费者支持消息的负载均衡策略。
• 支持动态新增、删除生产者、消费者。
• 目前仅仅支持关键字的精确匹配路由，后续会逐渐完善。
• 消息队列服务器Broker基于Netty的主从事件线程池模型开发设计。
• 网络消息序列化采用Kryo进行消息的网络序列化传输。
• Broker的消息派发、负载均衡、应答处理（ACK）基于异步多线程模型进行开发设计。
• Broker消息的投递，目前支持严格的消息顺序。其中Broker还支持消息的缓冲派发，即Broker会缓存一定数量的消息之后，再批量分配给对此消息感兴趣的消费者。

　　整个AvatarMQ依赖的jar包请参考：https://github.com/tang-jie/AvatarMQ/blob/master/nbproject/project.properties。

　　另外，值得注意的是：AvatarMQ的Netty是基于4.0版本的（下载地址：http://dl.bintray.com/netty/downloads/netty-4.0.37.Final.tar.bz2），消息序列化Kryo基于kryo-3.0.3（下载地址：https://github.com/EsotericSoftware/kryo/releases/tag/kryo-parent-3.0.3），请大家自行去官网下载使用。

　　现在，我们先来看下整合AvatarMQ项目的软件架构图：

　　从上述图例中，我们可以很清楚的看到：生产者和消费者之间是通过Broker进行消息的路由和转发，同时Broker还负责应答生产者和接收消费者的处理应答。

　　在了解了，整个AvatarMQ的组织架构之后，我们再来实际运行一下AvatarMQ！

　　首先，先启动一下Broker服务器（对应代码：https://github.com/tang-jie/AvatarMQ/blob/master/src/com/newlandframework/avatarmq/spring/AvatarMQServerStartup.java）

　　如果一切正常，终端控制台会打印如下输出：

　　接着，我们就来实际验证一下AvatarMQ的消息推送功能。

　　1、生产者发送1条消息给关注这条消息的消费者。我们先启动消费者，再启动生产者。其中消费者1的测试代码（AvatarMQConsumer1.java）为（对应github下载路径：https://github.com/tang-jie/AvatarMQ/blob/master/src/com/newlandframework/avatarmq/test/AvatarMQConsumer1.java）：

package com.newlandframework.avatarmq.test;import com.newlandframework.avatarmq.consumer.AvatarMQConsumer;import com.newlandframework.avatarmq.consumer.ProducerMessageHook;import com.newlandframework.avatarmq.msg.ConsumerAckMessage;import com.newlandframework.avatarmq.msg.Message;/** * @filename:AvatarMQConsumer1.java * @description:AvatarMQConsumer1功能模块 * @author tangjie<https://github.com/tang-jie> * @blog http://www.cnblogs.com/jietang/ * @since 2016-8-11 */public class AvatarMQConsumer1 {  private static ProducerMessageHook hook = new ProducerMessageHook() {    public ConsumerAckMessage hookMessage(Message message) {      System.out.printf("AvatarMQConsumer1 收到消息编号:%s,消息内容:%s\n", message.getMsgId(), new String(message.getBody()));      ConsumerAckMessage result = new ConsumerAckMessage();      result.setStatus(ConsumerAckMessage.SUCCESS);      return result;    }  };  public static void main(String[] args) {    AvatarMQConsumer consumer = new AvatarMQConsumer("127.0.0.1:18888", "AvatarMQ-Topic-1", hook);    consumer.init();    consumer.setClusterId("AvatarMQCluster");    consumer.receiveMode();    consumer.start();  }}

海外公司注册、海外银行开户、跨境平台代入驻、VAT、EPR等知识和在线办理：https://www.xlkjsw.com

原标题：Netty构建分布式消息队列（AvatarMQ）设计指南之架构篇

关键词：架构