RocketMQ设计原理与实践

1.MQ的作用

目前来说，消息队列主要有以下作用：降低耦合、实现异步处理、平谷削峰，这是消息队列最主要的作用，另外适当是使用消息队列还可以提高系统容错率、消息广播，偶尔也可以当做压测工具使用（消息积压 瞬间释放）。但是MQ的引入也会增加系统的复杂度和不稳定性，即使现在很多的MQ都是高可用的，也依然存在系统崩溃的危险。

2.主流开源MQ对比

目前主流的开源的MQ主要是三个，RabbitMQ、RocketMQ和Kafka。他们各有优缺点，也各有各的使用场景。下面是一些比较重要方面的比较：

• 性能和维护方面的对比

比较点	RabbitMQ	RocketMQ	Kafka
单机吞吐量	万级	十万级	十万级
时效性	微秒级	毫秒级	毫秒级
可用性	主从架构，高	分布式架构，非常高	分布式架构，非常高
自动主从切换	支持， 最早加入集群的slave会成为master；因为新加入的slave不同步master之前的数据，所以可能会出现部分数据丢失	不支持， master失效以后不能向master发送信息，consumer大概30s（默认）可以感知此事件，此后从slave消费；如果master无法恢复，异步复制时可能出现部分信息丢失	支持， N个副本，允许N-1个失效；master失效以后自动从isr中选择一个主；
消息堆积能力	一般， 生产者、消费者正常时，性能表现稳定；消费者不消费时，性能不稳定	非常好， 单机亿级堆积不是问题， 所有消息存储在同一个commit log中	非常好， 消息存储在log中，每个分区由一个或多个segment  log文件
稳定性	消息堆积时，性能不稳定、明显下降	队列较多、消息堆积时性能稳定	队列/分区多时性能不稳定，明显下降。 消息堆积时性能稳定
扩展性	基于erlang编写，源码难读，所以扩展性不好，定制也吃力，出现问题排查需要依靠开源社区，对于公司而言可控度低	基于java编写，维护和扩展成本都较低，出问题了也可以源码追踪，甚至可以基于它实现自己的MQ，可控度很高	基于Scala编写，kafka由于其特性，用在大数据领域的实时计算、日志采集等场景。维护成本高，但因为其架构特别成熟而且社区非常活跃

• 在功能上的一些对比

功能点	RabbitMQ	RocketMQ	Kafka
顺序消费	支持顺序消费	支持顺序消费，如果broker宕机，则消费队列会暂停	支持顺序消费 但是一台Broker宕机后，就会产生消息乱序
定时消息	不支持	支持特定level的定时消息	不支持
事务消息	不支持	支持	不支持
消息查询	不支持	支持 根据MessageId查询支持根据MessageKey查询消息	不支持
消息失败重试	支持失败重试	支持失败重试	不支持失败重试
批量发送	不支持	不支持	支持 默认producer缓存、压缩，然后批量发送

• 总结 
  通过上面的对比，可以发现每个MQ都有自己的特点：RabbitMQ长于并发和延时，RocketMQ长于复杂场景，Kafka长于吞吐。这就使得他们各自的使用场景可能不同。然而各自的缺点也很明显，RabbitMQ的高可用和高吞吐略逊色，最主要的是erlang语言导致其扩展和维护问题比较难；RocketMQ的缺点就是客户端只有java很成熟，其他比较少；Kafka的缺点就是只有核心功能，其他功能支持较少。而且RocketMQ是阿里巴巴中间件团队review了Kafka的源码后，用java重写的适用于阿里交易场景的消息中间件。

3.为什么选择RocketMQ

作为一个交易系统，面临的场景比较复杂，有些场景是必须要消息重试和消息查询的，而且事务消息也是必不可少的，很多关系到钱的场景都必须保证一点错误不出，就需要系统保证可靠性。RocketMQ在各方面性能都比较厉害，且经历了阿里双11的百万QPS的挑战，是有扛住万亿级流量的实践的，这点也是选择RocketMQ的重要一点，虽然RocketMQ也有些缺点，作为java程序员想要基于它定制或优化扩展的话是不难的。

4.RocketMQ的简单介绍

4.1 RocketMQ架构

• NameServer集群：简单的注册中心，保存broker的各种信息，彼此之间无关，每个内容都一样，即记录着全部broker的信息，broker启动的时候会轮询在每个NameServer上注册信息，所以每个broker都会和每一个NameServer保持一个长连接来保持信息同步和故障感知（每个30秒发送心跳测试）。所以这里感觉网络的开销是有点大的，所以理论上可以无限扩展，实际上不可能太多。一般三主三从就可以满足大部分需求。
• Producer和Consumer集群：producer不必多说，消息生产者，从配置里取到NameServer地址后，他与NameServer保持了长连接，以便定时拉取broker消息，拉取后会做本地缓存，拿到broker消息后，发送消息会与broker建立长连接，此连接每隔30秒会发心跳给broker以保持连接不被关闭。
• Broker集群：broker是RocketMQ中比较重要的部分，他负责消息存储和传递，消息查询，HA保证等，Broker服务器有几个重要的子模块 
  1)、远程处理模块，即代理的条目，处理来自客户端的请求； 
  2)、客户端模块，管理客户端（生产者/消费者）并维护消费者的主题订阅； 
  3)、Store Service，提供简单的API来存储或查询物理磁盘中的消息； 
  4)、HA服务，提供主代理和从代理之间的数据同步功能； 
  5)、索引服务，按指定**构建消息索引，并提供快速消息查询。 

4.2 RocketMQ的存储设计

RocketMQ很多方面借鉴的Kafka，但是也有很多自己的创新点，例如他的存储设计，优化了存储设计使得RocketMQ在单机五万队列的情况下不影响性能，这点比Kafka强很多。 
在Broker里，收到producer发送来的消息，首先将消息存储到Commit Log里，存储完毕即返回Producer客户端成功，这里的commit log是RocketMQ用来存储消息的文件，大小为1G，存储满了之后存到下一个commit log。在一个Broker中，所有的发送消息请求都被有顺序的存储在commit log里，由于是顺序写，所以可以利用page cache技术进行存储而不会出现很多的缺页中断，使得存储速度很快，接近内存的存储速度，然后RocketMQ根据刷盘策略选择写入page cache成功即返回成功（异步刷盘）或者写入page cache再写入硬盘后（同步刷盘）才返回客户端成功；一般场景使用异步刷盘策略就可以了，性能好，但是有丢失数据的风险；同步刷盘则是特定的场景需要的非常严格的策略，保证数据100%安全，但是势必会降低性能。 

说完消息的存储，再者就是消息的读取了，RocketMQ消费消息的时候，虽然是与consume queue交互，但是说到底还是在和commit log交互，因为consume Queue上面记录的就是该消息在commit log上的偏移量，简单说就是消息实体的引用，还是要去commit log读取数据的，而所有的consume Queue都共享一个commit log，故读取数据在整体上也变成了有序了，还是充分利用page cache读取数据，基本上都是从内存上读取出来的，速度很快。 

从上面的设计可以看出，在consume Queue数量很大的时候也基本不会影响它的读写速度，因为RocketMQ的读写永远是对一个commit log的顺序读写。

4.3 RocketMQ的高可用设计

RocketMQ有一点不好就是Broker主节点宕机之后，不支持broker从节点切换为主节点，虽然不支持切换，但是它还是有别的的设计来保证高可用。 
首先RocketMQ上的每个Topic在每个Broker上有topic分片，然后每个分片下都有好几个队列，所以一般一个Broker宕机的话，客户端发送消息可能会失败，但是当NameServer发现了该宕机broker的时候，就会剔除该broker的信息，producer客户端再去拉取信息的时候就不会再往故障的broker发送消息了，等手动恢复broker之后，上面的消息并不会丢失（某些情况会丢极少的数据），但是这种需要NameServer发现宕机的情况已经很晚了，最晚需要30秒，那在客户端还未从NameServer获取到新的路由信息的时候客户端是怎么处理的呢？两种机制重试和broker故障延时机制。 
重试机制就是客户端默认的机制，一般次数为3次，会选取另一个队列继续发送，但是这另一个队列也可能会在那个宕机的broker上，那么发送还是失败，所以有不小的概率导致你的应用系统发不出消息。 
第二种方式就是broker故障延时机制，相较于重试机制，更有针对性。默认是不开启的，开启需要设置sendLatencyFaultEnable=true，这种设计中会有一个集合，集合里存储了所有不可用的broker，并且记录他们被隔离的end时间，判断是否可用就是是不是发送消息失败了，发送失败就需要隔离；下次有消息需要发送的时候就会根据end时间来判断此broker是否可用。简单的说就是会把疑似故障的broker放入隔离集合隔离一段时间，过了时间之后再放到轮询集合里。这样的设计大大的提高客户端的应对能力，也提高了可用性。

5.RocketMQ实战

5.1 RocketMQ使用的一些总结

• RocketMQ支持信息的重复消费，很多时候会发生重复消费的情况，所以使用的业务方需要注意幂等性设计。
• 消息的发送和消费最好打出详细的日志，以便链路追踪和数据核对。
• 线上建议关闭autoCreateTopicEnable配置。
• 一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用*设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。
• RocketMQ在读取老数据的时候会产生大量的磁盘IO，所以slaveReadEnable这个参数最好设置为true来让老数据的读取都在从节点上进行，从而不影响主节点的写入。

5.2 事务消息场景分析

事务消息是一个二阶段提交的过程，大体流程如下 

交易场景实战：支付完成-发出创建子订单的消息-更新本地订单状态。这是一个很普通的场景，如果就按照普通的流程顺序处理就会出现问题，MQ发送了，但是本地的数据更新失败了，就会导致后面的对账系统反查的数据不对，最终流水没有写入；而对于子订单的创建也会有问题。有的人会说可以把更新本地数据放到前面，本地数据更新成功了才去发送消息，OK，那么更新本地数据成功后，发送消息失败了，整体需要回滚，那么发送这个步骤只能和更新数据放到本地大事务里，这就会导致一个问题，那就是消息已经发出去了，你的事务还没提交呢，数据照样没更新，结果照旧失败。

综上使用事务消息，当本地事务提交完成即数据更新完成之后再通知MQ提交消息就可以保证成功了。

6.RocketMQ的一些思考

首先RocketMQ的可用性是很高，但是没有主从切换有时候还是会出现问题，例如在使用顺序消息的时候，RocketMQ的解决方法是将消息投递到同一个队列里，这样必然会是有序的，这种场景下，如果该队列所在的broker挂了，那后面发送顺序消息都将失败！即功能不可用，所以主从的自动切换还是很有必要的，能进一步提高其可用性，现在也有不少公司基于开源的RocketMQ定制自己的消息队列就会加上主从自动切换。