基本原理
RocketMQ的broker会被划分成master和slave两种角色，只有master能接受producer所发送的消息，消息的replication示意图如下：
image
- Slave在初始化时，会先向master上报slave的消息文件（CommitLog）的offset
- Master收到slave的offset后，开始执行消息的同步，批量将offset后面的消息发送给Slave
- Slave收到master发送过来的消息后，将消息写入CommitLog，并向Master发送新的offset
- Master收到新的offset后，更新临时存储的slave offset信息(AtomicLong)，用于对producer做响应，当offset >= producer发送消息后的offset，则表示该producer发送的消息已经成功同步
消息复制的基本框架
前面介绍了RocketMQ的消息复制的基本原理，我们现在看看RocketMQ是如何做的消息复制，在RocketMQ中，有ha相关的包：
image
在store子工程中，虽然只有三个类，但代码结构确不简单，WaitNotifyObject是用于做等待唤醒的（生产者消费者模式），HA相关的主体代码在HAService和HAConnection及它们的内部类中，总体结构如下：
image
- AcceptSocketService：master用于接受slave的连接的类，其中包含一个线程
- HAConnection：对SocketChannel的read/write的封装
- SocketReadService：用于读取SocketChannel中的数据，读取的数据是slave发送过来的已同步的offset，表示此offset之前的消息都已全部同步，包含一个线程
- SocketWriteService：用于将消息数据发送slave，其中包含一个线程
- GroupTransferService：这个类主要是用于同步复制模式下，判断当前消息是否已经同步成功，没有则等待
- DefaultMessageStore：消息的存储，其中包括一个CommitLog
源码阅读
- 先看master向slave发消息（HAConnection中的WriteSocketService的run方法）：
image.png
可以看到，WriteSocketService线程每次从nextTransferFromWhere开始读取一定的消息数据，并通过transferData方法发送给slave，此外,nextTransferFromWhere会增加，增加的数量是读取的消息数量。将消息写到socketchannel时，除了消息本身，还会写入一定的头信息，putLong和putInt写入了12字节，分别表示offset和size
- Master读取Slave发送过来的offset（HAConnection中的ReadSocketService的processReadEvent方法）：
image
byteBufferRead是一个大小为1MB的ByteBuffer，因为offset是一个long类型的值，而long类型的值占8个字节，因此rocketmq在处理byteBufferRead中的数据时，需要以8字节为一个单位估数据的读取。
master数据收到的数据可能并不是一个完整的long，有可能会出现半包或者粘包的情况，因此会有一个int pos = this.byteBufferRead.position() – (this.byteBufferRead.position % 8)的方式计算出当前完整包的位置
当下一次读取数据时，会先判断byteBufferRead是不是满了，满了则清理一次，所以如果读到了半包，则下一次读取数据后，半包会变成一个完整的包：
image
rocketmq的master每次在读取到8字节及8字节以上的数据后，会取最后的完整的8字节作为offset。
- Slave读取Master发送的消息以及offset的上报
先看看HAClient的run方法，其主体逻辑是读取数据，然后上报offset：
image
- HAClient的reportSlaveMaxOffset方法的实现非常简单，就是将offset以long的形式发送给master：
image
- Slave通过HAClient的processReadEvent方法处理读取到的数据的处理，从run方法中调用processReadEvent，然后会调用dispatchReadRequest方法，在dispatchReadRequest方法中，我们可以看到实际上就是读取数据，然后写入到commitLog中
image
生产者发送消息的结果
我们回到CommitLog的putMessage方法，其中有handleDiskFlush的调用：
image
当刷盘超时，会返回FLUSH_DISK_TIMEOUT，我们再看看handleHA方法：
image
handleHA方法中可能会抛出FLUSH_SLAVE_TIMEOUT和SLAVE_NOT_AVAILABLE。
我们注意到，发消息的顺序：写byteBuffer -> flushDisk -> flushSlave，这三者之间没有事务保证，可能会出现以下情况：
- 写byteBuffer成功，flushDisk超时，此时可能会有数据丢失的风险，我们可以看看源码：
image
刷盘使用的是GroupCommitService，它包含一个队列和一个线程，刷盘实际上也是异步的，只不过SYNC_FLUSH模式下，会同步等待刷盘的结果，对于返回了FLUSH_DISK_TIMEOUT的消息，后面也会被刷到磁盘上，但是如果当master宕机，而消息还来不及刷盘。那就会有消息丢失了
- 对于同步slave超时的情况，需要看刷盘是否成功，如果刷盘成功，则不会丢消息，否则会有消息丢失。先看看相关代码：
image
实际上也是由HAService中的线程完成的同步slave，而SYNC_MASTER模式只是同步等待结果：
- 如果slave宕机，则slave恢复后，会从上次写入成功的offset开始，继续进行复制，消息不会丢
- 如果master宕机，则没来得及刷盘的消息可能会丢失
对于收到了FLUSH_DISK_TIMEOUT的结果时，生产者需要评估少量消息丢失是否会有影响，如果有，则需要生产者做重试，消费者支持幂等，如果没影响，则忽略即可
网友评论