作者：风声
部门：业务技术/社交电商
消息队列是分布式系统中重要的中间件，在实现系统高性能，高可用，可伸缩性和最终一致性架构框架中扮演着重要角色。是大型分布式系统不可缺少的核心中间件之一。
目前市面上比较常见的消息队列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ，NSQ等。本文将对三款优秀消息中间件（nsq,kafka,rocketMq）的实现架构进行简单介绍。
一、分布式场景下MQ能解决哪些具体问题？
1.1 系统解耦
在分布式特别是微服务场景下，系统之间的依赖关系会随着业务功能的不断拓展变得十分复杂。系统间调用链路会越来越长。最终导致整个系统内应用耦合会日趋严重。后期对系统进行服务拆分和优化会变的十分困难。
为了避免上述问题，可以采用消息队列的方式来处理部分系统调用。上游服务将数据发送到MQ服务，下游服务根据自身的业务需求和消费能力对MQ消息进行消费。
1.2 异步处理
系统间采用同步调用方式会受到系统吞吐量、并发量、响应时间等各种瓶颈限制。此时可以采用消息队列的方式将非必需或者实时性要求不高的请求异步化。在业务异步处理过程中还可以进行消息分裂，充分利用服务资源提高业务处理效率。
1.3 削峰填谷
在使用消息队列的场景下，当遇到流量洪峰时消息队列可以将大量消息蓄积在消息队列中，在洪峰结束之后消费者可以继续消费存量消息。如此可以大大提高系统的吞吐量，也有助于提高系统的稳定性和用户体验。在流量会出现剧烈波动场景特别适合使用，特别是电商促销场景。
1.4 数据分发
在分布式场景下，一条事件消息可能会被多个下游系统关注，大部分消息中间件均支持一对多消费或者消息广播的模式。消息中间件可以根据用户定义的规则进行消息路由分发，下游系统只需要关注自己感兴趣的消息即可。这一特性可以加以利用提高系统的扩展性。
二、分布式场景下消息中间件NSQ（有赞分支）、kafka、rocketMq，实现架构简述
消息中间件提供上述解决方案的同时也面临着一些列的技术挑战。如分布式、高并发场景下如何保证消息中间件稳定性、如何提高消息吞吐量、如何解决多副本协作一致性问题，如何提高消息存取的IO性能等。下面将就上述部分问题简单介绍下三种消息中间的技术落地方案。
2.1 多节点协作模型
分布式场景下实现服务的高性能、高可用、高扩展性必然要集群化部署，多节点协作完成服务处理。这就会引入服务感知、上下线、消息同步等问题。下面介绍下三种中间件服务节点之间是如何感知协作的。
2.1.1 nsq（有赞）
原生NSQ通过nsqlookup来做服务发现，为了保障注册服务的高可用性，nsqlookup可以部署多个服务。nsqlookup之间相互独立的互不干扰。有赞分支下对NSQ的服务感知模型做了升级，引入了ETCD服务来做集群管理协调工作和元数据存储。nsqd和nsqlookup服务启动后会向ETCD发起服务注册以便ETCD进行集群管理。nsqd会向nsqlookup上报节点负载信息，改造后nsqlookup服务会自动根据各nsqd节点的负载情况进行数据平衡，因此生产者和消费者都需要从nsqlookup服务查询topic对应的nsqd节点信息。有赞对于nsq消息存储模型也做了改进，提供了一致性保障，下文将介绍NSQ存储模型改进。架构图参见：图-1
图-1（[图片引用自NSQ有赞分支doc]）
2.1.2 kafka
Kafka使用ZooKeeper来做管理、协调工作，利用ZK的有序节点、临时节点和监听机制等特性完成负载均衡、集群管理、选举等功能。例如每个broker节点启动时都会到ZK上进行注册，在/brokers/ids目录下创建自己的节点。消费者组协作消费一个topic时需要将消费者和分区关系注册到ZK中以确保每个topic分区仅被同组的一个消费者消费。同时ZK还负责存储topic、分区、消费者等元信息。（下文中会介绍到消费者组的消费索引相关信息，并未存储在ZK中）架构图参见：图-2
图-2
2.1.3 rocketMq
rocketMq使用轻量级的NameServer服务进行服务的协调和治理工作，NameServer多节点部署时相互独立互不干扰。每一个rocketMq服务节点（broker节点）启动时都会遍历配置的NameServer列表并建立长链接，broker节点每30秒向NameServer发送一次心跳信息、NameServer每10秒会检查一次连接的broker是否存活。消费者和生产者会随机选择一个NameServer建立长连接，通过定期轮训更新的方式获取最新的服务信息。架构图参见：图-3
图-3
2.2 消息存储模型&数据同步模型&存取高性能
2.2.1 nsq(有赞分支)
- 消息存储模型优化
原生NSQ在消息送到达NSQD（NSQ核心服务）服务后会先存储在内存中，当内存中消息累积到一定量后才会落到数据盘中。原生NSQD节点之间无法实现分布式协作，并且单点故障时会出现消息丢失。有赞NSQ优化了消息存储模型，topic落盘方式改造为实时落盘并且增加了数据副本机制，保障了消息的可靠性。由于引入了数据副本机制需要在leader节点进行消息读写，为了提高读写的可扩展性，对topic引入分区的概念，每个topic可以指定多个分区，每个分区使用独立的leader节点，这样保证每个topic可以有多个可以读写的分区节点，提高了topic读写的性能。
- 消费模型优化
原生NSQ消费时channel会把消息从topic复制过来，然后push给channel下的消费者组。有赞NSQ对topic改造之后，channel本身不会存储消息数据了，只需要记录每个channel已经同步的数据偏移量和每个channel的消费偏移量即可。这样所有的channel引用的是同一份topic磁盘数据，每个channel维护自己独立的偏移信息即可。这样设计优化掉了数据复制的操作提高了消费性能。（nsq消费模型参见：图-4）
- 消息一致性&可靠性保障
有赞NSQ将每个topic的数据节点副本元信息写入etcd，然后通过etcd选举出每个topic的leader节点。选举的leader节点负责自己topic的数据副本同步，其他follower节点从leader节点同步topic数据。当某个leader节点失效时, 会触发etcd的watch事件, 从而触发nsqlookupd重新选择其他存活节点作为topic的新leader, 完成leader的快速切换后继续向外提供服务。
上文提到每个topic选举出来的leader节点负责同步数据到所有副本。为了支持副本节点的动态变化，有赞NSQ采用了ISR(In synced replica)的设计。用push模式来保证数据的同步复制，避免数据同步不一致。因此，数据写入首先由leader节点发起，并且同步到所有ISR副本节点成功后，才返回给客户端。如果同步ISR节点失败，则尝试动态调整ISR并重试直到成功为止。重试的过程中会检查leader的有效性，以及是否重复提交等条件。
图-4
2.2.2 kafka
- Topic&Partition 的存储
topic是一个存储消息的逻辑概念，一个topic代表一个消息集合，从硬件存储来说各个topic的消息是分开存储的。topic可以划分多个分区，分区或落在不同的borker节点上。当一条消息发送到broker时，会根据分区规则选择获取一个partition编号来存储消息。消息在被添加到分区时，都会被分配一个offset（偏移量），它是消息在此分区中的唯一编号，kafka通过offset保证消息在分区内的顺序，offset的顺序不跨分区，只保证在同一个分区内的消息是有序的。
Partition 是以文件的形式存储在文件系统中，比如创建一个名为topic1的topic，其中有3个partition，那么在kafka的数据目录（/tmp/kafka-log）中就有3个目录，topic1-0~2，命名规则topic_name-partition_id，创建3个分区的topic文件目录。每个目录下会有三个文件，一个log文件和两个索引文件（稀疏的偏移量索引和时间索引）。其中log文件又会按照规则去不停的切分为segment(默认1G一个)，其他技术细节比如索引文件的技术细节、文件清理细节等感兴趣的可以自行查阅官网，此处不在详述。（消息存储文件参见：图-5）
图-5
- 消息的消费原理
在使用时topic会配置多个partition，这样做首先可以减少单个分片上的消息数量并且可以并发写入不同分片提升写入速度，其次多个consumer 去同时并发消费同一个topic的不同分区，可以提升一个消费者组消费同一个topic的速度。 消费者组内的消费者一起协调来消费订阅topic的所有分区。一个分区仅可由一个consumer消费,消费者不足时，一个消费者可以消费多个分区。如果消费者数量大于分区数，则多出来的消费者会空跑。kafka支持多种消费者组和分区消费的对应逻辑，如循环分配、粘连分配等感兴趣可以官网上查询。
kafka提供了一个特殊的topic用于存储消费者组的消费偏移量。命名规则：_consumer_offsets* ，kafka会把offset信息写入到这个topic 中 。__consumer_offsets保存了每个consumer group某一时刻提交的 offset信息。通过offset信息结合消息目录下的索引文件信息，可以快速定位到消息所在的物理位置，具体实现细节此处不在赘述。kafka消费模型参见：图-6
- 消息一致性&可靠性保障
kafka利用zk临时节点特性从所有broker中选举出一个controller节点，controller节点会负责一些管理工作，如监听broker变化、监听topic变化、监听分区变化，管理分区信息等。
kafka对topic分区采用多副本机制来保障消息存储的可靠性，leader分区负责读写，follower仅负责从leader拉取数据做同步保障。因此分区的数量必须小于等于broker的数量且kafka会尽量保障每个broker所负责的分区数量达到一个均衡。
上面讲到follower会从leader同步信息，当leader 异常时，kafka会从与leader保持同步度高的副本(ISR)中选举一个新的leader。并在进行消息同步处理之后继续向外提供服务。具体细节比较复杂此处无法展开描述，可以自行去官网了解。
图-6
2.2.3 rocketMq
- 存储逻辑设计
rocketMq在消息存储上设计思路与kafka和NSQ的思路不同，实际存储中既没有分区的概念，也没有按照topic进行存储。而是将所有topic的消息全部写入同一个文件中（commit log），这样保证了IO写入的绝对顺序性，最大限度利用IO系统顺序读写带来的优势提升写入速度。
由于消息混合存储在一起，需要将每个消费者组消费topic最后的偏移量记录下来。这个文件就是consumer queue（索引文件）。所以消息在写入commit log 文件的同时还需将偏移量信息写入consumer queue文件。在索引文件中会记录消息的物理位置、偏移量offset,消息size等，消费者消费时根据上述信息就可以从commit log文件中快速找到消息信息。rocketMq消费模型参见：图-7
- 存储文件简介
Commit log 消息存储文件，rocket Mq会对commit log文件进行分割（默认大小1GB），新文件以消息最后一条消息的偏移量命名。
Consumer queue 会根据消费者情况有多个，每个文件记录30万数据，写满则会进行切割。
Index file是索引文件。由于rocketMq 支持对于设定的特定属性进行检索，所以必然会有一个hash索引来支撑这个功能。如果需要使用消息检索功能，则尽可能保证索引的字段具有高离散度，来保证检索的效率。
- 消息一致性&可靠性保障
连接到相同NameServer下配置集群名称相同的broker会自动组成集群。根据配置文件参数会自动组成主从节点，主从之间进行数据同步（一般建议设置成同步复制 异步刷盘模式）。
从服务器每5秒会通过TCP连接去主服务器拉取最大偏移量之后还未同步的消息。在2019年4.5.0版本之中集成了Dledger技术（基于raft管理Commit log,此功能默认关闭），在不需要外部协助的情况下可以自主进行故障转移。
rocketMq文件清理策略、过期策略、文件存取零拷贝等以及上述功能的技术细节不在展开描述，可以去网上查阅相关资料。
图-7
三、特性总结&选型分析
3.1 特性总结
3.2 选型分析
本文主要对于三种MQ的总体架构做了简单说明，并未涵盖所有功能，如事务性消息，死信队列，延迟消息等。实际场景需要使用Mq时可以根据自己场景来判断。提供如下几点参考建议：
- 日志&大数据相关场景建议选用Kafka，这两方面kafka久经考验比较成熟稳定。
- topic数量过多选择RocketMq，kafka和NSQ在topic达到几百个之后性能会显著下降。RocketMq可以支持几千个。原因上面有提到，前两者是根据topic和分区来存储文件，topic过多时物理存储的件数量增多，读写会越来越接近于随机IO。而RocketMq存储文件只有commit log是混合存储模型，从设计上避免了这种多topic带来的问题。
- 需要定制服务建议选择NSQ。NSQ对于扩展性最友好，比较轻量级可以方便的进行二次开发。
- 使用复杂度方面，kafka需要引入ZK来做管理，并且没有配套的管理后台，需要使用第三方的管理平台。NSQ和RocketMq都有自己的管理后台，并且部署比较方便。
