rocketMQ总结

消息系统作用

• 削峰填谷（主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题）
• 系统解耦（解决不同重要程度、不同能力级别系统之间依赖导致一死全死）
• 提升性能（当存在一对多调用时，可以发一条消息给消息系统，让消息系统通知相关系统）
• 蓄流压测（线上有些链路不好压测，可以通过堆积一定量消息再放开来压测）

一、RocketMQ组成

1、NameServer

协调者，基于内存完成，是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

 

2、Broker

实例分Master和Slave、暂存消息，Master读写，Slave只读，Master与Slave的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。

3、Topic

订阅标题

4、Tag

Topic里的标签

5、MessageQueue

Topic里的队列

6、Offset

标记消息在Message Queue里的位置，标记消费读取时自增长

7、Producer

Producer与Name Server集群中的其中一个节点(随机选择)建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

Producer每隔30s（由ClientConfig的pollNameServerInterval）从Name server获取所有topic队列的最新情况，这意味着如果Broker不可用，Producer最多30s能够感知，在此期间内发往Broker的所有消息都会失败。

Producer每隔30s（由ClientConfig中heartbeatBrokerInterval决定）向所有关联的broker发送心跳，Broker每隔10s中扫描所有存活的连接，如果Broker在2分钟内没有收到心跳数据，则关闭与Producer的连接。

8、Consumer

Consumer与Name Server集群中的其中一个节点(随机选择)建立长连接，定期从Name Server取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消息，订阅规则由Broker配置决定。

Consumer每隔30s从Name server获取topic的最新队列情况，这意味着Broker不可用时，Consumer最多最需要30s才能感知。

Consumer每隔30s（由ClientConfig中heartbeatBrokerInterval决定）向所有关联的broker发送心跳，Broker每隔10s扫描所有存活的连接，若某个连接2分钟内没有发送心跳数据，则关闭连接；并向该Consumer Group的所有Consumer发出通知，Group内的Consumer重新分配队列，然后继续消费。

当Consumer得到master宕机通知后，转向slave消费，slave不能保证master的消息100%都同步过来了，因此会有少量的消息丢失。但是一旦master恢复，未同步过去的消息会被最终消费掉。

二、消息模式

Clustering 

同一个 ConsumerGroup (GroupName相同)里的Consumer 只消费所订阅消息一部分内容。

Broadcasting
同一个 ConsumerGroup (GroupName相同)里的Consumer 只消费所订阅消息是全部内容。

实现区别:

Clustering模式即同组ConsumerGroup下的每个Consumer消费位置不同,由Broker端存储和控制Offset

Broadcasting模式下每个Consumer使用LocalFileOffsetStore本地存储Offset

三、生产消费

DefaultMQProducer

返回状态:

FLUSH_DISK_TIMEOUT 消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失

FLUSH_SLAVE_TIMEOUT 消息发送成功，但是服务器同步到Slave时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失

SLAVE_NOT_AVAILABLE 消息发送成功，但是此时slave不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失

Slave SEND_OK表示发送成功

延时消息:setDelayTimeLevel设置延迟时间

自定义消息发送规则:使用MessageQueueSelector类在覆写select方法中返回选中的MessageQueue

事务支持:两阶段提交协议 发送方向RocketMQ发送待确认消息-持久化后返回发送成功-将本地事务逻辑与发送确认消息包装在同一事务中并执行事务-RocketMQ收到确认消息后订阅方对消息可见并消费 发送方事务阶段异常则待确认消息作废 发送方提供给RocketMQ回查接口用于查询事务结果

同步发送: 需要等MQ返回相应

异步发送：无需MQ返回相应，需要实现SendCallback

消费者： (分push、pull模式)
interface MQPushConsumer:
DefaultMQPushConsumer：由系统控制读取操作，收到消息后自动调用传人的处理方法来处理

interface MQPullConsumer:
实现类 DefaultMQPullConsumer：读取操作中的大部分功能由使用者自主控制,使用者记录offset

Push与Pull比较:Push(推)由服务端主动推送消息至客户端,实际通过Pull保持连接并等待服务端获得从生产者发送来的消息,在等待期间若获得消息则通过连接发送至消费者,在等待超过限定时间后返回空结果至消费者 Pull(拉)由消费者发起连接到服务端获得消息,需预判消息发送频率,连接频率过长过短均有问题

四、协调者

 

1.功能

 

具体就是记录维护Topic - Broker的信息，及监控Broker的运行状态。为client提供路由能力（具体实现和zk有区别，NameServer是没有leader和follower区别的，不进行数据同步，通过Broker轮训修改信息）。

 

2.结构

3.Remoting模块

概览: 通信通过Remoting模块统一自定义消息格式RemotingCommand完成 

 

 

4.协议格式

 

心跳机制

单个Broker跟所有Namesrv保持心跳请求，心跳间隔为30秒，心跳请求中包括当前Broker所有的Topic信息。Namesrv会反查Broer的心跳信息， 如果某个Broker在2分钟之内都没有心跳，则认为该Broker下线，调整Topic跟Broker的对应关系。但此时Namesrv不会主动通知Producer、Consumer有Broker宕机。

Consumer跟Broker是长连接，会每隔30秒发心跳信息到Broker。Broker端每10秒检查一次当前存活的Consumer，若发现某个Consumer 2分钟内没有心跳， 就断开与该Consumer的连接，并且向该消费组的其他实例发送通知，触发该消费者集群的负载均衡(rebalance)。

生产者每30秒从Namesrv获取Topic跟Broker的映射关系，更新到本地内存中。再跟Topic涉及的所有Broker建立长连接，每隔30秒发一次心跳。 在Broker端也会每10秒扫描一次当前注册的Producer，如果发现某个Producer超过2分钟都没有发心跳，则断开连接。

五、消息队列的核心机制

---

1.消息存储结构

物理存储文件CommitLog 消息的逻辑队列ConsumerQueue类似数据库的索引文件 每个Topic下的每个MessageQueue有一个对应的ConsumeQueue文件

2.高可用机制

在创建Topic时将Topic的多个MessageQueue创建在多个Broker组中(相同Broker名称 不同BrokerId的机器组成一个Broker组)，当一个Broker组内的Master不可用时可向其他Broker组的Master发送消息

3.同步刷盘和异步刷盘

异步刷盘:写入到内存即返回写成功 当内存消息量累计到一定程度后，统一写入磁盘

同步刷盘:写入内存后通知刷盘线程刷盘，刷盘完成后刷盘线程唤醒等待的线程返回写成功的状态

4.同步复制和异步复制

同步复制:Master和Slave均写成功后返回成功状态

异步复制:Master写成功即返回成功状态

5.磁盘读取机制
顺序写，随机读，零拷贝

6.写入及复制机制
Master读和写，Slave只读，生产者写入Master，Master复制到Slave

六、可靠性优先的使用场景

顺序机制
1、完全顺序
需要把Topic的读写队列设置为1，Producer 和 Consumer 并发设置为1

2、部分顺序
1)生产者需要把消息发送到同一个Message Queue;
2)消费组需要不并发读一个Message Queue;

3、消息重复

消费者应处理消息产生的结果具有幂等性即无论接收到多少消息产生结果均一致或确保消息的id根据id去重

4、动态增减

动态增加NameServer可以通过HTTP服务方式 动态减少Broker时注意若只有一个Master要先关闭生产者而多个Master则若是同步发送则无影响 异步发送则丢失部分消息

5、消息优先级

对于不同的优先级消息可以创建多个Topic来消费或者同Topic下不同优先级的消息按照一定策略写入不同的MessageQueue后由消费者对MessageQueue顺序读取

 

七、吞吐量优先的使用场景

1.在Broker端过滤消息

方式:Tag和Key过滤由Broker在CommitLog读取消息时根据消息的Tag的HashCode进行过滤、SQL过滤、FilterServer过滤

 

 

2.提高Consumer处理能力

查看消费逻辑是否有优化空间 提高消费并行度通过提高Consumer数量但不能超过ReadQueue数量 批量方式消费信息一次读取多条信息 当消息积压过多通过Offset判断直接舍弃消息

3.Consumer的负载均衡

DefaultMQPushConsumer启动后触发doRebalance动作，加入新得到DefaultMQPushConsumer也触发doRebalance动作，Topic下的不同MessageQueue分配到不同Consumer下

DefaultMQPullConsumer使用registerMessageQueueListener或MQPullConsumerScheduleService

4.提高Producer的发送速度

Oneway方式将数据写入客户端的socket缓冲区即返回 增加Producer的并发量

八、Rocketmq如何支持分布式事务消息

https://www.jianshu.com/p/2838890f3284

 

九、为什么不用Zookeeper

RocketMQ：

RocketMQ中的master和slave可以认为是物理上的概念。一台机器要么是master，要么是slave，与topic无关，是在配置文件中指定的（也可以换种方式理解，认为master和slave是逻辑上的概念，broker是物理上的概念。但是一个broker要么是master，要么是slave）。master与slave通过配置brokerName来进行配对。相同的brokerName中，brokerId为0的表示master，其他表示slave。

示例图中有9台broker机器。

 

kafka：

kafka中的master和slave可以认为是逻辑上的概念。一个broker上可以部署master，也可以部署slave。

示例图中有3台broker机器。

 

我们可以看出其中较为关键的差异，kafka的master与slave是由选举产生的，需要zookeeper提供的选举机制。RocketMQ的master与slave是由配置确认的，一旦确认不会改变。

 

实际上，在早期的RocketMQ版本中，路由中心就是由zookeeper实现的，后来由于上述原因，改成了更高效的NameServer。

 

NameServer的启动流程

NameServer启动时，会加载配置文件，加载KV配置，进行一些初始化动作。

同时会开启两个定时任务，分别做Broker的清理和KV配置信息的打印。

 

路由注册

路由注册在broker启动时触发，broker启动时会和所有NameServer创建心跳连接，向NameServer发送Broker的相关信息。

NameServer在RouteInfoManager类中维护了Broker相关信息的缓存，进行更新动作。更新时用了读写锁，既保证了极高并发场景下的读效率，又避免了并发修改缓存。

 

 

路由删除

路由删除的触发点有两个：

1）NameServer启动时开启的定时任务，每隔10s扫描一次brokerLiveTable，检测上次心跳包与当前系统时间差，如果时间差大于120s，则移除Broker的相关信息。

2）Broker正常关闭，会向NameServer发送UNREGISTER_BROKER消息。

 

实现上和注册类似，加读写锁，维护相关的缓存信息，这里就不画图了。

 

路由发现

客户端定时向NameServer发起请求GET_ROUTEINFO_BY_TOPIC，获取对应的信息。流程较为简单，这边不画图了。

 

 

rocketMQ和kafka不同
1、偏向事务机制；
2、不支持Master选举，即不能Slave转Master