# RocketMQ源码解析：主从同步和读写分离实现

在这里插入图片描述

# 启动HAService

在这里插入图片描述图片来自RocketMQ的官微，对整个主从同步实现的过程概括的比较清晰，对着图说一下具体的流程

master和slave端都会启动HAService，slave端的HAClient会和broker端的HAService建立连接
master端的AcceptSocketService用来处理slave端的连接，并将连接封装成HAConnection，一个连接封装成一个HAConnection
每个HAConnection会启动2个线程WriteSocketService和ReadSocketService
WriteSocketService，读取salve已经同步的offset并保存。ReadSocketService，发送commitLog数据

# 源码解析

# 主从同步

在这里插入图片描述

之前的文章我们已经分析了往内存中存储消息，以及刷盘的过程，我们接着看最后一步，主从同步。

在RocketMQ中主从同步有两种方式

同步，感觉和mysql中的半同步差不多，只要消息写入一个slave即可
异步，用户不用等到消息发送到slave即可收到ack

在这里插入图片描述

从代码中可以看出来，输盘和主从同步是同时进行（并不是刷盘完毕才进行主从同步） 写个例子演示一下

public CompletableFuture<Integer> getResult() {
    CompletableFuture<Integer> result1 = new CompletableFuture();
    CompletableFuture<Integer> result2 = new CompletableFuture();
    new Thread(() -> {
        sleepRandom();
        result1.complete(1);
    }).start();

    new Thread(() -> {
        sleepRandom();
        result2.complete(2);
    }).start();

    return result1.thenCombine(result2, (num1, num2) -> {
        return num1 + num2;
    });
}

@Test
public void demo() throws ExecutionException, InterruptedException {
    long start = System.currentTimeMillis();
    CompletableFuture<Integer> future = getResult();
    // 线程阻塞在这里等待结果，直到等到结果3
    System.out.println(future.get());
    // 2145
    System.out.println(System.currentTimeMillis() - start);
}

例子中的2个线程就相当于一个处理刷盘，一个处理主从同步

处理同步的套路和刷盘的套路差不多，将同步请求放到阻塞队列中，然后GroupTransferService不断处理这些请求，请求处理完毕则唤醒对应的线程

CommitLog#submitReplicaRequest 在这里插入图片描述当slave和master相差太多的时候，会返回SLAVE_NOT_AVAILABLE

GroupTransferService#doWaitTransfer 在这里插入图片描述 doWaitTransfer会不断的将push2SlaveMaxOffset（slave同步的最大偏移量，多个slave同时更新这一个值）和req.getNextOffset（当前消息存储完的偏移量）进行比较，如果大于说明至少有一个salve同步完成了，唤醒阻塞的线程即可