决战圣地玛丽乔亚Day07----RocketMQ

回顾上次的内容：

NameServer、Broker、Producer、Consumer

生产者向Broker发送消息，放在消息队列，消费者通过消息推过来/自己拉取的方式消费消息。

NameServer和Broker保持心跳。他们之间建立长连接进行通信。

Producer想发消息，就要问NameServer要Broker的信息，所以Producer和NameServer也建立了长链接通信。获取消息后可以根据Topic来对目标Broker建立连接发送信息，再根据负载均衡选一台Broker。

Broker一般采用主从+集群架构方式，Broker集群中的每一个都需要和NameServer集群建立长链接进行信息同步，例如更新Topic信息等。

Consumer和Producer类似，也会和NameServer集群中的某一个建立联系

Consumer可以和Master订阅消息，也可以和Slave订阅消息，Broker可以配置。

生产者生产信息：

、

生产者拉取到NameServer的消息，发现TopA有4个MessageQueue，分布在两个Broker上，然后Producer会根据MessageQueueSelector消息队列的选择具体实现或者默认配置（发送到某个MessageQueue）然后分散给不同的broker。

宕机：

如上图中，如果第二个Master Broker宕机了，虽然Slave Broker会自动切换成Master，但在切换的过程中该Broker服务是不可用的。 对于这个问题，在Producer中有一个开关sendLatencyFaultEnable，这个开启后会有一个容错机制，比如某次访问一个Broker有500ms延迟还无法访问，那么接下来就会回避访问该Broker一段时间，比如3000ms内不再访问该Broker，避免消息打到故障的Broker上。

消息持久化：
消息持久化是RocketMQ最核心的环节，它决定了生产者消息写入的吞吐量，决定了消息会不会丢失，消费者获取消息的吞吐量。 Broker的消息持久化依赖于两个文件CommitLog和ConsumeQueue。 当Broker收到一条消息后，首先会把该消息顺序写入磁盘文件CommitLog

这个CommitLog是很多磁盘文件，每个文件最多1GB，当一个文件写满之后，就新建一个。 现在我们的消息已经持久化在了磁盘上，但是有一个问题，当消费者要消费一条消息时，它怎么知道从CommitLog中具体获取哪个消息呢？ 这时就用到另一个磁盘文件ConsumeQueue，在Broker中，每个MessageQueue都有一系列ConsumeQueue文件，如： $HOME/store/consumequeue/{topic}/{queueid}/{filename}

queueid就是对应MessageQueue，这个ConsumeQueue文件储存的就是一条消息在CommitLog中的偏移量，看到这里是不是有的懵逼，到底什么意思呢？ 其实就是当Broker收到一条消息后，会把消息在CommitLog中的物理位置，也就是一个文件偏移量，记录在对应的MessageQueue的ConsumeQueue文件中。（可以理解为List中的角标索引）

ConsumeQueue文件的记录结构：

 ConsumeQueue文件里一条数据是20个字节，最多存放 30W条数据，文件最大5.72MB，存放满了就新建一个。所以实际上Topic的每个 MessageQueue 都对应了 Broker 机器上的多个 ConsumeQueue，保存了 MessageQueue 的消息在 CommitLog 中的物理位置，也就是偏移量。

消费者消费信息：

首先我们要了解一个消费组的概念，即一群消费者组成的组，将Topic中的消息队列分散给消费组中的多个消费者，可以提高消息的消费的效率。

上图中，一个Topic有两个消费组来消费，Broker 会分别发送消息到这两个组中，根据订阅规则（集群模式、广播模式）来决定是组里的每台机器都消费还是只有一台来消费，消费者消息消息的过程是先从ComsumerQueue读取消息的offset，

然后根据offset到CommitLog读取相应的消息，集群模式消费者会把从CosumerQueue读取的offset保存的Broker，广播模式则存储在本地。

订阅规则 RocketMq有两种订阅规则：集群模式、广播模式。 集群模式就是一条消息一个消费组里只会有一台机器会去消费。

广播模式就是一条消息一个消费组里每一台机器都会去消费。集群模式的原理就是：RocketMq有一个 MessageQueue 分配算法，默认会把 MessageQueue 平均分配给每个Consumer，且一个 MessageQueue 只会分配给一个 Consumer，这是因为消费者每次读取消息的时候，

会把最后一个消息的offset保存到Broker存储，下次读取会根据上次保存的offset进行消息读取，如果多个消费读取同一个队列，就会导致偏移量保存被覆盖，进而导致消息被重复

度。 消费方式 RocketMq提供了2种消费方式：PUSH和PULL，但实际上，这两种方式本质上是一样的，都是消费者主动去 Broker 拉取消息。 Push模式其实也是消费者主动去 Broker 拉取消息，只不过它实时性很高，就像 Broker 在 Push 一

样。它是通过长轮询来实现的，当消费者发起请求到 Broker ,如果没有消息的话，就会把线程挂起（默认15秒），在此期间会有一个后台线程每隔一段时间就去检查一下是否有新的消息，如果有，就唤起线程。

订阅规则：

RocketMq有两种订阅规则：集群模式、广播模式。 集群模式就是一条消息一个消费组里只会有一台机器会去消费。 广播模式就是一条消息一个消费组里每一台机器都会去消费。

集群模式的原理就是：RocketMq有一个 MessageQueue 分配算法，默认会把 MessageQueue 平均分配给每个Consumer，且一个 MessageQueue 只会分配给一个 Consumer，这是因为消费者每次读取消息的时候，会把最后一个消息的offset保存到Broker存储，

下次读取会根据上次保存的offset进行消息读取，如果多个消费读取同一个队列，就会导致偏移量保存被覆盖，进而导致消息被重复度。

消费方式

RocketMq提供了2种消费方式：PUSH和PULL，但实际上，这两种方式本质上是一样的，都是消费者主动去 Broker 拉取消息。 Push模式其实也是消费者主动去 Broker 拉取消息，只不过它实时性很高，就像 Broker 在 Push 一样。

它是通过长轮询来实现的，当消费者发起请求到 Broker ,如果没有消息的话，就会把线程挂起（默认15秒），在此期间会有一个后台线程每隔一段时间就去检查一下是否有新的消息，如果有，就唤起线程。