RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。

我们直接去GitHub上看Apache对他的描述可能会好点

发布/订阅消息传递模型

财务级交易消息

各种跨语言客户端，例如Java，C / C ++，Python，Go

可插拔的传输协议，例如TCP，SSL，AIO

内置的消息跟踪功能，还支持开放式跟踪

多功能的大数据和流生态系统集成

按时间或偏移量追溯消息

可靠的FIFO和严格的有序消息传递在同一队列中

高效的推拉消费模型

单个队列中的百万级消息累积容量

多种消息传递协议，例如JMS和OpenMessaging

灵活的分布式横向扩展部署架构

快如闪电的批量消息交换系统

各种消息过滤器机制，例如SQL和Tag

用于隔离测试和云隔离群集的Docker映像

功能丰富的管理仪表板，用于配置，指标和监视

认证与授权

他的优缺点是啥

RocketMQ优点：

单机吞吐量：十万级

可用性：非常高，分布式架构

消息可靠性：经过参数优化配置，消息可以做到0丢失

功能支持：MQ功能较为完善，还是分布式的，扩展性好

支持10亿级别的消息堆积，不会因为堆积导致性能下降

源码是java，我们可以自己阅读源码，定制自己公司的MQ，可以掌控

天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况

RoketMQ在稳定性上可能更值得信赖，这些业务场景在阿里双11已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择RocketMQ

RocketMQ缺点：

支持的客户端语言不多，目前是java及c++，其中c++不成熟

社区活跃度不是特别活跃那种

没有在 mq 核心中去实现JMS等接口，有些系统要迁移需要修改大量代码

消息去重

去重原则：使用业务端逻辑保持幂等性

幂等性：就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用，数据库的结果都是唯一的，不可变的。

只要保持幂等性，不管来多少条重复消息，最后处理的结果都一样，需要业务端来实现。

去重策略：保证每条消息都有唯一编号(比如唯一流水号)，且保证消息处理成功与去重表的日志同时出现。

建立一个消息表，拿到这个消息做数据库的insert操作。给这个消息做一个唯一主键（primary key）或者唯一约束，那么就算出现重复消费的情况，就会导致主键冲突，那么就不再处理这条消息。

消息重复

消息领域有一个对消息投递的QoS定义，分为：

最多一次（At most once）

至少一次（At least once）

仅一次（ Exactly once）

QoS：Quality of Service，服务质量

几乎所有的MQ产品都声称自己做到了At least once。

既然是至少一次，那避免不了消息重复，尤其是在分布式网络环境下。

比如：网络原因闪断，ACK返回失败等等故障，确认信息没有传送到消息队列，导致消息队列不知道自己已经消费过该消息了，再次将该消息分发给其他的消费者。

不同的消息队列发送的确认信息形式不同，例如RabbitMQ是发送一个ACK确认消息，RocketMQ是返回一个CONSUME_SUCCESS成功标志，Kafka实际上有个offset的概念。

RocketMQ没有内置消息去重的解决方案，最新版本是否支持还需确认。

消息的可用性

当我们选择好了集群模式之后，那么我们需要关心的就是怎么去存储和复制这个数据，RocketMQ对消息的刷盘提供了同步和异步的策略来满足我们的，当我们选择同步刷盘之后，如果刷盘超时会给返回FLUSH_DISK_TIMEOUT，如果是异步刷盘不会返回刷盘相关信息，选择同步刷盘可以尽最大程度满足我们的消息不会丢失。

除了存储有选择之后，我们的主从同步提供了同步和异步两种模式来进行复制，当然选择同步可以提升可用性，但是消息的发送RT时间会下降10%左右。

RocketMQ采用的是混合型的存储结构，即为Broker单个实例下所有的队列共用一个日志数据文件（即为CommitLog）来存储。

而Kafka采用的是独立型的存储结构，每个队列一个文件。

这里帅丙认为，RocketMQ采用混合型存储结构的缺点在于，会存在较多的随机读操作，因此读的效率偏低。同时消费消息需要依赖ConsumeQueue，构建该逻辑消费队列需要一定开销。

RocketMQ 刷盘实现

Broker在消息的存取时直接操作的是内存（内存映射文件），这可以提供系统的吞吐量，但是无法避免机器掉电时数据丢失，所以需要持久化到磁盘中。

刷盘的最终实现都是使用NIO中的 MappedByteBuffer.force() 将映射区的数据写入到磁盘，如果是同步刷盘的话，在Broker把消息写到CommitLog映射区后，就会等待写入完成。

异步而言，只是唤醒对应的线程，不保证执行的时机，流程如图所示。

顺序消息：

我简单的说一下我们使用的RocketMQ里面的一个简单实现吧。

Tip：为啥用RocketMQ举例呢，这玩意是阿里开源的，我问了下身边的朋友很多公司都有使用，所以读者大概率是这个的话我就用这个举例吧，具体的细节我后面会在RocketMQ和Kafka各自章节说到。

生产者消费者一般需要保证顺序消息的话，可能就是一个业务场景下的，比如订单的创建、支付、发货、收货。

那这些东西是不是一个订单号呢？一个订单的肯定是一个订单号的说，那简单了呀。

一个topic下有多个队列，为了保证发送有序，RocketMQ提供了MessageQueueSelector队列选择机制，他有三种实现:

我们可使用Hash取模法，让同一个订单发送到同一个队列中，再使用同步发送，只有同个订单的创建消息发送成功，再发送支付消息。这样，我们保证了发送有序。

RocketMQ的topic内的队列机制,可以保证存储满足FIFO（First Input First Output 简单说就是指先进先出）,剩下的只需要消费者顺序消费即可。

RocketMQ仅保证顺序发送，顺序消费由消费者业务保证!!!

这里很好理解，一个订单你发送的时候放到一个队列里面去，你同一个的订单号Hash一下是不是还是一样的结果，那肯定是一个消费者消费，那顺序是不是就保证了？

真正的顺序消费不同的中间件都有自己的不同实现我这里就举个例子，大家思路理解下。

分布式事务：

Half Message(半消息)

是指暂不能被Consumer消费的消息。Producer 已经把消息成功发送到了 Broker 端，但此消息被标记为暂不能投递状态，处于该种状态下的消息称为半消息。需要 Producer

对消息的二次确认后，Consumer才能去消费它。

消息回查

由于网络闪段，生产者应用重启等原因。导致Producer端一直没有对Half Message(半消息)进行二次确认。这是Brock服务器会定时扫描长期处于半消息的消息，会

主动询问Producer端 该消息的最终状态(Commit或者Rollback),该消息即为消息回查。

A服务先发送个Half Message给Brock端，消息中携带 B服务 即将要+100元的信息。

当A服务知道Half Message发送成功后，那么开始第3步执行本地事务。

执行本地事务(会有三种情况1、执行成功。2、执行失败。3、网络等原因导致没有响应)

如果本地事务成功，那么Product像Brock服务器发送Commit,这样B服务就可以消费该message。

如果本地事务失败，那么Product像Brock服务器发送Rollback,那么就会直接删除上面这条半消息。

如果因为网络等原因迟迟没有返回失败还是成功，那么会执行RocketMQ的回调接口,来进行事务的回查。

消息过滤

Broker端消息过滤

在Broker中，按照Consumer的要求做过滤，优点是减少了对于Consumer无用消息的网络传输。缺点是增加了Broker的负担，实现相对复杂。

Consumer端消息过滤

这种过滤方式可由应用完全自定义实现，但是缺点是很多无用的消息要传输到Consumer端。

Broker的Buffer问题

Broker的Buffer通常指的是Broker中一个队列的内存Buffer大小，这类Buffer通常大小有限。

另外，RocketMQ没有内存Buffer概念，RocketMQ的队列都是持久化磁盘，数据定期清除。

RocketMQ同其他MQ有非常显著的区别，RocketMQ的内存Buffer抽象成一个无限长度的队列，不管有多少数据进来都能装得下，这个无限是有前提的，Broker会定期删除过期的数据。

例如Broker只保存3天的消息，那么这个Buffer虽然长度无限，但是3天前的数据会被从队尾删除。

回溯消费

回溯消费是指Consumer已经消费成功的消息，由于业务上的需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度。

例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。

RocketMQ支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。

消息堆积

消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力，消息堆积分以下两种情况：

消息堆积在内存Buffer，一旦超过内存Buffer，可以根据一定的丢弃策略来丢弃消息，如CORBA Notification规范中描述。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于内存Buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。

消息堆积到持久化存储系统中，例如DB，KV存储，文件记录形式。 当消息不能在内存Cache命中时，要不可避免的访问磁盘，会产生大量读IO，读IO的吞吐量直接决定了消息堆积后的访问能力。

评估消息堆积能力主要有以下四点：

消息能堆积多少条，多少字节？即消息的堆积容量。

消息堆积后，发消息的吞吐量大小，是否会受堆积影响？

消息堆积后，正常消费的Consumer是否会受影响？

消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大？

定时消息

定时消息是指消息发到Broker后，不能立刻被Consumer消费，要到特定的时间点或者等待特定的时间后才能被消费。

如果要支持任意的时间精度，在Broker层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。

RocketMQ支持定时消息，但是不支持任意时间精度，支持特定的level，例如定时5s，10s，1m等。

总结

写这种单纯介绍中间件的枯燥乏味，大家看起来估计也累，以后我这种类型的少写，大家老是让我写点深度的，我说真的很多东西我源码一贴，看都没人看。

本号专注Java源码分析。喜欢底层源码的朋友可以来交流探讨。交流群：818491202 验证：88 整理了一份Java高级架构资料、Spring源码分析、Dubbo、Redis、Netty、zookeeper、Spring cloud、分布式等资