RocketMQ源码：主从同步实现原理(HA)

曾经终败给现在 2024-03-23 13:14 127阅读 0赞

主从同步的实现逻辑主要在`HAService`中，在`DefaultMessageStore`的构造函数中，对`HAService`进行了实例化，并在start方法中，启动了`HAService`：

public class DefaultMessageStore implements MessageStore {
        public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
            final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
            // ...
            if (!messageStoreConfig.isEnableDLegerCommitLog()) {
                // 初始化HAService
                this.haService = new HAService(this);
            } else {
                this.haService = null;
            }
            // ...
        }
    
        public void start() throws Exception {
            // ...
            if (!messageStoreConfig.isEnableDLegerCommitLog()) {
                // 启动HAService
                this.haService.start();
                this.handleScheduleMessageService(messageStoreConfig.getBrokerRole());
            }
            // ...
        }
    }
    复制代码

在`HAService`的构造函数中，创建了`AcceptSocketService`、`GroupTransferService`和`HAClient`，在start方法中主要做了如下几件事：

1.  调用`AcceptSocketService`的`beginAccept`方法，这一步**主要是进行端口绑定，在端口上监听从节点的连接请求**（可以看做是`运行在master节点的`）;
2.  调用`AcceptSocketService`的start方法启动服务，这一步**主要为了处理从节点的连接请求，与从节点建立连接**（可以看做是`运行在master节点`的）;
3.  调用`GroupTransferService`的start方法，**主要用于在主从同步的时候，等待数据传输完毕**（可以看做是`运行在master节点`的）;
4.  调用`HAClient`的start方法启动，**里面与master节点建立连接，向master汇报主从同步进度并存储master发送过来的同步数据**（可以看做是`运行在从节点`的）;

public class HAService {
         public HAService(final DefaultMessageStore defaultMessageStore) throws IOException {
            this.defaultMessageStore = defaultMessageStore;
            // 创建AcceptSocketService
            this.acceptSocketService =
                new AcceptSocketService(defaultMessageStore.getMessageStoreConfig().getHaListenPort());
            this.groupTransferService = new GroupTransferService();
            // 创建HAClient
            this.haClient = new HAClient();
        }
    
        public void start() throws Exception {
            // 开始监听从服务器的连接
            this.acceptSocketService.beginAccept();
            // 启动服务
            this.acceptSocketService.start();
            // 启动GroupTransferService
            this.groupTransferService.start();
            // 启动
            this.haClient.start();
        }
    }
    复制代码

![format_png][]

## 1. 监听从节点连接请求 ##

`AcceptSocketService`的`beginAccept`方法里面首先获取了`ServerSocketChannel`，然后进行端口绑定，并在selector上面注册了`OP_ACCEPT`事件的监听，监听从节点的连接请求：

public class HAService {
        class AcceptSocketService extends ServiceThread {
            /**
             * 监听从节点的连接
             *
             * @throws Exception If fails.
             */
            public void beginAccept() throws Exception {
                // 创建ServerSocketChannel
                this.serverSocketChannel = ServerSocketChannel.open();
                // 获取selector
                this.selector = RemotingUtil.openSelector();
                this.serverSocketChannel.socket().setReuseAddress(true);
                // 绑定端口
                this.serverSocketChannel.socket().bind(this.socketAddressListen);
                // 设置非阻塞
                this.serverSocketChannel.configureBlocking(false);
                // 注册OP_ACCEPT连接事件的监听
                this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
            }
        }
    }
    复制代码

## 2. 处理从节点连接请求 ##

`AcceptSocketService`的run方法中，对监听到的连接请求进行了处理，处理逻辑大致如下：

1.  从`selector`中获取到监听到的事件；
2.  如果是`OP_ACCEPT`连接事件，创建与从节点的连接对象`HAConnection`，与从节点建立连接，然后调用`HAConnection`的start方法进行启动，并创建的`HAConnection`对象加入到连接集合中，**HAConnection中封装了Master节点和从节点的数据同步逻辑**；

public class HAService {
        class AcceptSocketService extends ServiceThread {
            @Override
            public void run() {
                log.info(this.getServiceName() + " service started");
                // 如果服务未停止
                while (!this.isStopped()) {
                    try {
                        this.selector.select(1000);
                        // 获取监听到的事件
                        Set<SelectionKey> selected = this.selector.selectedKeys();
                        // 处理事件
                        if (selected != null) {
                            for (SelectionKey k : selected) {
                                // 如果是连接事件
                                if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                                    SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();
                                    if (sc != null) {
                                        HAService.log.info("HAService receive new connection, "
                                            + sc.socket().getRemoteSocketAddress());
                                        try {
                                            // 创建HAConnection，建立连接
                                            HAConnection conn = new HAConnection(HAService.this, sc);
                                            // 启动
                                            conn.start();
                                            // todo 添加连接
                                            HAService.this.addConnection(conn);
                                        }
                                        ...
                           
            }
        }
    }
    复制代码

## 3. 等待主从复制传输结束 ##

`GroupTransferService`的`run方法`主要是为了在进行`主从数据同步的时候，等待从节点数据同步完毕`。

在运行时首先进会调用`waitForRunning`进行等待，因为此时可能还有没有开始主从同步，所以先进行等待，之后如果有同步请求，会唤醒该线程，然后调用`doWaitTransfer`方法等待数据同步完成：

public class HAService {
        class GroupTransferService extends ServiceThread {
    
             public void run() {
                log.info(this.getServiceName() + " service started");
                // 如果服务未停止
                while (!this.isStopped()) {
                    try {
                        // 等待运行
                        this.waitForRunning(10);
                        // todo 如果被唤醒，调用doWaitTransfer等待主从同步完成
                        this.doWaitTransfer();
                    } catch (Exception e) {
                        log.warn(this.getServiceName() + " service has exception. ", e);
                    }
                }
    
                log.info(this.getServiceName() + " service end");
            }
        }
    }
    复制代码

在看`doWaitTransfer`方法之前，首先看下是如何判断有数据需要同步的。

Master节点中，当消息被写入到`CommitLog`以后，会调用`submitReplicaRequest`方法处主从同步，首先判断当前Broker的角色是否是`SYNC_MASTER`，如果是则会构建消息提交请求`GroupCommitRequest`，然后调用`HAService`的`putRequest`添加到请求集合中，并唤醒`GroupTransferService`中在等待的线程：

public class CommitLog {
        public CompletableFuture<PutMessageStatus> submitReplicaRequest(AppendMessageResult result, MessageExt messageExt) {
            if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
                HAService service = this.defaultMessageStore.getHaService();
                if (messageExt.isWaitStoreMsgOK()) {
                    if (service.isSlaveOK(result.getWroteBytes() + result.get

[format_png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/23/d80bd7ea1ac847e293c0fa29f1836b18.png