RocketMQ

消息堆积能力

消息存储性能

同步刷盘不会丢失

异步刷盘会少量丢失

单节点开启异步复制时，只丢失少量

可引入双写机制

不发生消息堆积时，以长轮询模式实现准实时的消息推送

已经消费完的消息，可以根据业务要求重新消费

消息只能在指定时间之后才能被消费【16个等级，如5秒，15秒等】
收费版才支持任意时间精度

重发时不会发给同一broker，因为有可能该broker已经挂了【规避原则】

发送同一类消息的生产者称之为一个生产者组

如果 Producer 中途意外宕机，
Broker 会主动回调 Producer Group 内的任意一台机器来确认事务状态

Producer 实例线程安全

业务层面的唯一性标识，可以用来查找消息（有索引文件）

RocketMQ不能保证messegeId唯一，所以要设置唯一key

不同的消息类型(定时消息、事务消息)使用不同的主题

不同的业务类型消息(天猫订单、京东订单)，使用不同的主题

不同的消息量级使用不同的主题

一个topic下默认4个分片，RocketMQ里没有队列的概念

二级消息类型，用来进一步区分某个 Topic 下的消息分类，用于消费端过滤消息

淘宝交易订单的不同类型（女装订单、电器、化妆品订单）使用不同的tag

消息的全局唯一标识：机器 IP+消息偏移量【所以可能重复】

发送标识：成功、失败等

点击这里查看SendResult里状态标识

RocketMQ收到消息后，所有主题的消息都存储在commitlog中，然后采用异步转发到consumerQueue中
ConsumeQueue中，每个主题的数据会做数据分片(Queue)。Kafka中的存储是没有队列的概念的

一个Topic，默认有4个Queue，就不能保证顺序了。会返回给生产者一个queueId

验证消息、查找路由(第一次从NameServer)、消息发送

一批的消息不超过4M，且具有相同主题、标签等。组装成一个List

重发时，不会再发给同一broker【规避原则】，以提高重发成功率

DefaultMQProducerImpl

SendResult sendDefaultImpl  ( Message msg, final CommunicationMode communicationMode
, final SendCallback sendCallback, final long timeout    )

不管怎么重试，总的超时时间是不会超过方法入参里的timeout的

int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;

从上面源码可知，同步发送才有重试次数，单向发送和异步发送只有一次发送机会

getRetryTimesWhenSendFailed()，默认重试次数是 2 

由下面源码可知，如果获取MessageQueue和Topice超时超过总的超时限制（方法传进来的超时时间）则不会发送消息

long costTime = beginTimestampPrev - beginTimestampFirst;
if (timeout < costTime) {
       callTimeout = true;
       break; // 跳出循环不走后面的发送代码，也不会重试
 }

Consumer默认是16次

consumer重试有时间间隔，可在配置文件中配置

在广播模式下不会进行重试

异常重试：消费者发生异常在catch块里主动返回RECONSUME_LATER时会进行重试

超时重试：是指消费者没有返回RECONSUME_LATER，也没有返回consume_success时，会一直重试到最大重试次数

broker会一直重试到16次（默认）后丢进死信队列，已和架构组确认

版本问题

如果broker配置的重试次数是16，但我们消费者只想重试3次怎么办？

在消费代码的catch块里，获取重试次数reconsumeTimes，如果到了3次，直接返回成功

若出现网络问题导致消息根本没有发送到消费者，则会重新投递

使用相同group Id的订阅者属于同一个消费群组

一条消息只需要被群组内的任意一个消费者消费即可

每条消息，会推送给群组内的所有消费者，且保证被每个消费者消费一次

会将broker里的Queue都拉过来，也可以指定MessageQueue读取消息

读取的offset，需要自己保存处理；而推模式很多事都不用管，比如偏移量直接在broker里

拉取消息后，返回结果会有以下四种状态之一：
Found(获取到消息)，没有匹配到消息，没有新消息，非法偏移量

拉取消息数量配置注意

长轮询

流量控制

重新分布机制

消息确认（ACK）

消息进度存储

推模式原理

Offset是指某个 Topic下的一条消息在某个 Message Queue里的 位置

集群模式时，由 Broker 端存储和控制 Offset 的值，使用 RemoteBrokerOffsetStore 结构

广播模式时，每个 Consumer 都收到这个 Topic 的全部消息，各个 Consumer 间相互没有干扰，
 RocketMQ 使 用 LocalfileOffsetStore，把 Offset存到本地

rocketmq的指定offset消费，只能是刚启动时有用，后面都是接着上次的消费进度消费

生产者和消费者的并发设置成1，Queue也设置成1【性能不高】

mqadmin update Topic -t AllOrder -c DefaultCluster -r 1 -w 1 -n 127.0.0.1:9876

一个主题根据sharding key进行分区，在同一分区内保证生产和消费的FIFO
【通过不同的tag实现的】

1、应用模块遇到要发送事务消息的场景时，先发送prepare消息给MQ。
     默认的RMQ_SYS_TRANS_HALF_TOPIC主题

2、prepare消息发送成功后，应用模块执行数据库事务（本地事务）。
     在实现TransactionListener接口的类里处理

3、根据数据库事务执行的结果，再返回Commit或Rollback给MQ。

4、如果是Commit，MQ把消息下发给Consumer端【此时移到真实Topic里】
      如果是Rollback，直接删掉prepare消息。

5、如果第3步的执行返回结果是UNKNOW，或是超时的，
     则启动定时任务(自己开发，麻破火山口)回查事务状态（最多重试15次，
     超过了默认丢弃此消息），处理结果同第4步。也是在TransactionListener实现类里处理

6、MQ消费的成功机制由MQ自己保证。

主要是为了不确认是否真正成功时的一个查询判断

通过transactionCheckInerval设置回查时间间隔，默认1分钟

网络闪断或客户端宕机时，客户端应答失败，生产者会重复发消息，
消费者后面就会收到两条messageID一样的两条消息

投递给消费者时，网络闪断应答失败，也会重复投递

broker或客户端重启、扩容等会触发Rebalance时，也可能收到重复消息

以唯一的消息key标识，例如订单号

有了唯一订单号，消费者在消费时可通过订单号先去查询判断是否已经处理过

FilterServerConsumer类，4.3.0以后不支持了

tag过滤

SQL92过滤

文件系统  > KV存储 > 关系数据库

关系数据库 > KV存储 > 文件系统

所有主题的所有消息，都会顺序追加在commitlog(默认1G）文件中，顺序追加大大增加了写消息的速度

在broker里设置mapedFileSizeCommitLog可改变大小，写满一个commitlog文件又会新起一个

commitlog文件名为文件里第一条消息在整个commitlog文件组中的偏移量，
当知道某条消息的offset时，减去文件名就能算出在该文件中的绝对地址

commitlog的文件存储逻辑视图中，每个主题有4个字节存储消息的总长度

存储了consumeQueue里的队列、messageKey、Tag等所有信息
所以既保证了和consumequeue的一致性，也保证consumeQueue丢失也能找回来

消息写到commitlog文件中后，都是异步转发消息条目给ConsumeQueue和index文件的(不是转发的整个消息)

为什么要有consumeQueue?

存储设计

使用consumeQueue查找

分区

为什么要有index文件？

index存储结构

解决哈希冲突

像topics.json、subscribeOffset.json等配置文件，不要在线上打开此文件，不然会阻塞

abort文件是启动时创建，正常退出之前删除

启动时若出现abort文件，说明broker是异常退出的

正常退出时，会通过注册JVM的钩子函数删除abort文件

存储commitlog文件最后一次刷盘时间戳、consumeQueue最后一次刷盘时间、index索引文件最后一次刷盘时间

全量消息都存储在commitlog文件中，然后异步生成转发任务更新ConsumeQueue文件、index文件
由于是异步，在broker重启时，就有可能出现数据并没有真正更新到ConsumeQueue和index文件中

加载时，会先判断是否有abort文件，有则是异常退出，没有则是正常退出

commitlog存储了全量消息，checkpoint文件又存储了所有刷盘点，重启时直接根据上次的刷盘点进行重新转发就行

异步停止恢复时，会将最后一个有效文件中的所有消息全部重新转发到ConsumeQueue和index文件中，
虽然确保了不丢失消息，但带来了消息重复的问题。所以消费者一定要做幂等设计

通过broker里的该属性配置是异步还是同步刷盘

消息写入pageCache之后，立刻通知刷盘并等待，
刷完了后再唤醒该线程，最后返回写入成功状态

消息只是写了内存的页面缓存里就返回成功，会有一个单独的线程按照某个频率，统一执行刷盘操作

异步刷盘，可批量刷，提高性能
但是需要通过集群或数据备份保证宕机不丢数据

堆内存——>堆外内存——>映射内存——>磁盘

consumeQueue的刷盘方式固定为异步刷盘，反正丢了也能重新生成

index文件是每更新一次index文件就将上一次的改动写入磁盘

通过fileReservedTime配置文件保留时间，默认72小时，超过这个时间该文件没有再次更新则认为过期

删除文件时间间隔

删除的文件还在被引用时，
允许再等一段时间才删除

指定删除时间点

磁盘是否充足

物理使用率

物理磁盘使用率

性能最高，但是有一个master宕机期间，未被消费的消息不能消费

线上数据要求高的推荐，主与从之间采用同步方式复制数据，
但刷盘方式都设置为异步刷盘

主挂了，从还可以继续提供消息的消费。可能会丢失消息

消息实时性高，没有消息不用做其他的操作

接收方可能处理不了这么多消息

接收方可根据自己的情况来决定拉消息：处理不过来等会儿拉，或暂停拉取

消费者可以定义缓存多少数据后批量拉取回来

消息延迟：消费者不能频繁拉取，所以会有一定的延迟

太多无用拉请求：某段时间内一直没消息，则循环拉请求就会做无用功

默认都是推模式

拉模式，推也是拉封装的

RebalanceService 线程会根据 topic 的队列数量和当前消费组的消费者个数做负载均衡，
每个队列产生的 pullRequest 放入阻塞队列 pullRequestQueue 中

PullMessageService 线程不断的从阻塞队列 pullRequestQueue 中获取 pullRequest，
然后通过网络请求 broker，这样实现的准实时拉取消息

PullMessageProcessor 里面的 processRequest 方法是用来处理拉消息请求的，有消息就直接返回

如果没有消息，且拉请求允许，则会挂起当前请求，有数据后，ReputMessageService会唤醒

ReputMessageService 线程，这个线程用来不断地从 commitLog 中解析数据并分发请求，
构建出 ConsumeQueue 和 IndexFile 两种类型的数据，并且也会有唤醒请求的操作