RocketMQ-broker存储机制-HA数据同步
HA机制解决读写分离模式下slave与master的数据同步问题，在master broker高负载的情况下，实现slave broker的数据订阅。HA的主要实现逻辑在HaServer类中，入口在putMessage的handleHA()方法初。
HA分为同步复制和异步复制，同步复制逻辑和同步刷盘机制差不多，都是同步等待通知的机制。put request之后，调用request的waitForFlush方法。
public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) { if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) { HAService service = this.defaultMessageStore.getHaService(); if (messageExt.isWaitStoreMsgOK()) { // Determine whether to wait if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) { GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes()); service.putRequest(request); service.getWaitNotifyObject().wakeupAll(); boolean flushOK = request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout()); if (!flushOK) { log.error("do sync transfer other node, wait return, but failed, topic: " + messageExt.getTopic() + " tags: " + messageExt.getTags() + " client address: " + messageExt.getBornHostNameString()); putMessageResult.setPutMessageStatus(PutMessageStatus.FLUSH_SLAVE_TIMEOUT); } } // Slave problem else { // Tell the producer, slave not available putMessageResult.setPutMessageStatus(PutMessageStatus.SLAVE_NOT_AVAILABLE); } } } }
具体的处理逻辑在GroupTransferService中，其跟同步刷盘一样，采用读写分离，通过比较slave传过来的已经完成的offset是否 >= 与当前数据在本地commitlog的offset,来确定是否完成数据同步
private void doWaitTransfer() { synchronized (this.requestsRead) { if (!this.requestsRead.isEmpty()) { for (CommitLog.GroupCommitRequest req : this.requestsRead) { boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset(); /** * 最多循环比较4次，如果发现已经同步到slave上的offset 超过了当前的需要同步数据在本地commitlog的offset的时候 * 表示已经成功同步 */ for (int i = 0; !transferOK && i < 5; i++) { this.notifyTransferObject.waitForRunning(1000); transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset(); } if (!transferOK) { log.warn("transfer messsage to slave timeout, " + req.getNextOffset()); } // 通知给客户端 req.wakeupCustomer(transferOK); } this.requestsRead.clear(); } } }
而真正的数据同步是采用异步方式具体实现在HaConnection中，服务端发送commitlog中的数据，并接收到slave同步进度的ACKoffset. 而slave在HaClient中，每5s中上报同步进度或者收到master的数据并写到commitlog中后发送同步进度。
public void start() throws Exception { // 注册服务端监听，绑定端口，注册selecter选择器和感兴趣的事件选择键 this.acceptSocketService.beginAccept(); // 启动服务端就绪选择，把接收到的客户端连接通道包装成HaConnection来处理 /** * HaConnection内部实现了ReadSocketService和WriteSocketService * ReadSocketService 用于master接收来自slave的同步进度 * WriteSocketService 用于发送commitlog的同步数据 */ this.acceptSocketService.start(); // 接受broker HA请求，并通知客户端是否完成数据同步的服务 this.groupTransferService.start(); // haClient是slave处理master的同步数据 以及 slave上报同步进度 this.haClient.start(); }
先来看一下master 的acceptSocketService.start()执行的逻辑
/** {@inheritDoc} */ @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) { // 接收到来自slave的连接 SocketChannel sc = ((ServerSocketChannel) k.channel()).accept(); if (sc != null) { HAService.log.info("HAService receive new connection, " + sc.socket().getRemoteSocketAddress()); try { // HAConnection管理着各个slave的读写 HAConnection conn = new HAConnection(HAService.this, sc); conn.start(); HAService.this.addConnection(conn); } catch (Exception e) { log.error("new HAConnection exception", e); sc.close(); } } } else { log.warn("Unexpected ops in select " + k.readyOps()); } } selected.clear(); } } catch (Exception e) { log.error(this.getServiceName() + " service has exception.", e); } } log.info(this.getServiceName() + " service end"); }
public void start() { // 处理slave的同步进度 this.readSocketService.start(); // 向slave发送同步数据 this.writeSocketService.start(); }
接下来看一下slave 是如何处理的
@Override public void run() { log.info(this.getServiceName() + " service started"); while (!this.isStopped()) { try { // 建立master連接 初始化socketChannel // 同时也注册了读事件 if (this.connectMaster()) { // 距离上一次slave请求master的时间是否超过5s if (this.isTimeToReportOffset()) { // 同步当前的offset到master boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset); if (!result) { this.closeMaster(); } } // 就绪选择 堵塞 this.selector.select(1000); // 收到master的数据 // 处理master的同步数据 并写到本地commitlog boolean ok = this.processReadEvent(); if (!ok) { this.closeMaster(); } // salve本地写完数据 则立即向master同步进度 if (!reportSlaveMaxOffsetPlus()) { continue; } long interval = HAService.this.getDefaultMessageStore().getSystemClock().now() - this.lastWriteTimestamp; // 距离上一次收到master同步的数据超过了20s 则断开连接 if (interval > HAService.this.getDefaultMessageStore().getMessageStoreConfig() .getHaHousekeepingInterval()) { log.warn("HAClient, housekeeping, found this connection[" + this.masterAddress + "] expired, " + interval); this.closeMaster(); log.warn("HAClient, master not response some time, so close connection"); } } else { this.waitForRunning(1000 * 5); } } catch (Exception e) { log.warn(this.getServiceName() + " service has exception. ", e); this.waitForRunning(1000 * 5); } } log.info(this.getServiceName() + " service end"); }