天天看點
傳回

通過源碼分析RocketMQ主從複制原理

作者：京東雲開發者

作者：京東物流 宮丙來

一、主從複制概述

  • RocketMQ Broker的主從複制主要包括兩部分内容：CommitLog的消息複制和Broker中繼資料的複制。
  • CommitLog的消息複制是發生在消息寫入時，當消息寫完Broker Master時，會通過單獨的線程，将消息寫入到從伺服器，在寫入的時候支援同步寫入、異步寫入兩種方式。
  • Broker中繼資料的寫入，則是Broker從伺服器通過單獨的線程每隔10s從主Broker上擷取，然後更新從的配置，并持久化到相應的配置檔案中。
  • RocketMQ主從同步一個重要的特征：主從同步不具備主從切換功能，即當主節點當機後，從不會接管消息發送，但可以提供消息讀取。

二、CommitLog消息複制

2.1、整體概述

通過源碼分析RocketMQ主從複制原理

CommitLog主從複制的流程如下：

1.Producer發送消息到Broker Master，Broker進行消息存儲，并調用handleHA進行主從同步； 2.如果是同步複制的話，參考2.6章節的同步複制； 3.如果是異步複制的話，流程如下：

1. Broker Master啟動，并在指定端口監聽；
2. Broker Slave啟動，主動連接配接Broker Master，通過Java NIO建立TCP連接配接；
3.  Broker Slave以每隔5s的間隔時間向服務端拉取消息，如果是第一次拉取的話，先擷取本地CommitLog檔案中最大的偏移量，以該偏移量向服務端拉取消息
4.  Broker Master 解析請求，并傳回資料給Broker Slave；
5.Broker Slave收到一批消息後，将消息寫入本地CommitLog檔案中，然後向Master彙報拉取進度，并更新下一次待拉取偏移量；

我們先看下異步複制的整體流程，最後再看下同步複制的流程，異步複制的入口為HAService.start();

public void start() throws Exception {
 //broker master啟動，接收slave請求，并處理
    this.acceptSocketService.beginAccept();
    this.acceptSocketService.start();
 //同步複制線程啟動
    this.groupTransferService.start();
 //broker slave啟動
    this.haClient.start();
}

下面分别對上面的每一步做詳細說明。

2.2、HAService Master啟動

public void beginAccept() throws Exception {
    this.serverSocketChannel = ServerSocketChannel.open();
    this.selector = RemotingUtil.openSelector();
    this.serverSocketChannel.socket().setReuseAddress(true);
    this.serverSocketChannel.socket().bind(this.socketAddressListen);
    this.serverSocketChannel.configureBlocking(false);
    this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
}

在beginAccept方法中主要建立了ServerSocketChannel、Selector、設定TCP reuseAddress、綁定監聽端口、設定為非阻塞模式，并注冊OP_ACCEPT(連接配接事件)。可以看到在這裡是通過Java原生的NIO來實作的，并沒有通過Netty架構來實作。

acceptSocketService.start()啟動方法代碼如下：

while (!this.isStopped()) {
    try {
   //擷取事件
        this.selector.select(1000);
        Set<SelectionKey> selected = this.selector.selectedKeys();
        if (selected != null) {
            for (SelectionKey k : selected) {
//處理OP_ACCEPT事件，并建立HAConnection
                if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) {
                    SocketChannel sc = ((ServerSocketChannel) k.channel()).accept();
                    if (sc != null) {
                       HAConnection conn = new HAConnection(HAService.this, sc);
                       //主要是啟動readSocketService，writeSocketService這兩個線程
 conn.start();
                       HAService.this.addConnection(conn);
                    }
                }
            }
            selected.clear();
        }
    } catch (Exception e) {
        log.error(this.getServiceName() + " service has exception.", e);
    }
}

選擇器每1s處理一次處理一次連接配接就緒事件。連接配接事件就緒後，調用ServerSocketChannel的accept()方法建立SocketChannel，與服務端資料傳輸的通道。然後為每一個連接配接建立一個HAConnection對象，該HAConnection将負責Master-Slave資料同步邏輯。HAConnection.start方法如下：

public void start() {
	this.readSocketService.start();
	this.writeSocketService.start();
}

2.3、HAClient啟動

while (!this.isStopped()) {
	try {
		//和broker master建立連接配接，通過java nio來實作
		if (this.connectMaster()) {
			//在心跳的同時，上報offset
			if (this.isTimeToReportOffset()) {
				//上報offset
				boolean result = this.reportSlaveMaxOffset(this.currentReportedOffset);
				if (!result) {
					this.closeMaster();
				}
			}
			this.selector.select(1000);
			//處理網絡讀請求，也就是處理從Master傳回的消息資料
			boolean ok = this.processReadEvent();
			if (!ok) {
				this.closeMaster();
			}
			if (!reportSlaveMaxOffsetPlus()) {
				continue;
			}
			long interval =
				HAService.this.getDefaultMessageStore().getSystemClock().now()
					- this.lastWriteTimestamp;
			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);
	}
}

2.3.1、HAService主從建立連接配接

如果socketChannel為空，則嘗試連接配接Master，如果Master位址為空，傳回false。

private boolean connectMaster() throws ClosedChannelException {
	if (null == socketChannel) {
		String addr = this.masterAddress.get();
		if (addr != null) {
			SocketAddress socketAddress = RemotingUtil.string2SocketAddress(addr);
			if (socketAddress != null) {
				this.socketChannel = RemotingUtil.connect(socketAddress);
				if (this.socketChannel != null) {
					//注冊讀事件，監聽broker master傳回的資料
					this.socketChannel.register(this.selector, SelectionKey.OP_READ);
				}
			}
		}
		//擷取目前的offset
		this.currentReportedOffset = HAService.this.defaultMessageStore.getMaxPhyOffset();
		this.lastWriteTimestamp = System.currentTimeMillis();
	}
	return this.socketChannel != null;
}
  1. Broker 主從連接配接

Broker Slave通過NIO來進行Broker Master連接配接，代碼如下：

SocketChannel sc = null;
sc = SocketChannel.open();
sc.configureBlocking(true);
sc.socket().setSoLinger(false, -1);
sc.socket().setTcpNoDelay(true);
sc.socket().setReceiveBufferSize(1024 * 64);
sc.socket().setSendBufferSize(1024 * 64);
sc.socket().connect(remote, timeoutMillis);
sc.configureBlocking(false);
  1. Slave擷取目前offset 
public long getMaxPhyOffset() {
	return this.commitLog.getMaxOffset();
}
public long getMaxOffset() {
	return this.mappedFileQueue.getMaxOffset();
}
public long getMaxOffset() {
	MappedFile mappedFile = getLastMappedFile();
	if (mappedFile != null) {
		return mappedFile.getFileFromOffset() + mappedFile.getReadPosition();
	}
	return 0;
}

可以看到最終還是通過讀取MappedFile的position來擷取從的offset。

2.3.2、上報offset時間判斷

private boolean isTimeToReportOffset() {
	//目前時間-上次寫的時間
	long interval =
		HAService.this.defaultMessageStore.getSystemClock().now() - this.lastWriteTimestamp;
	boolean needHeart = interval > HAService.this.defaultMessageStore.getMessageStoreConfig()
		.getHaSendHeartbeatInterval();


	return needHeart;
}

判斷邏輯為目前時間-上次寫的時間>haSendHeartbeatInterval時，則進行心跳和offset的上報。haSendHeartbeatInterval預設為5s，可配置。

2.3.3、上報offset

private boolean reportSlaveMaxOffset(final long maxOffset) {
	this.reportOffset.position(0);
	this.reportOffset.limit(8);
	this.reportOffset.putLong(maxOffset);
	this.reportOffset.position(0);
	this.reportOffset.limit(8);
	//最多發送三次，reportOffset是否有剩餘
	for (int i = 0; i < 3 && this.reportOffset.hasRemaining(); i++) {
		try {
			this.socketChannel.write(this.reportOffset);
		} catch (IOException e) {
			log.error(this.getServiceName()
				+ "reportSlaveMaxOffset this.socketChannel.write exception", e);
			return false;
		}
	}
	return !this.reportOffset.hasRemaining();
}

主要還是通過NIO發送請求。

2.4、Broker Master處理請求

在主從建立連接配接時建立了HAConnection對象，該對象主要包含了如下兩個重要的線程服務類：

//負責寫，将commitlog資料發送到從
private WriteSocketService writeSocketService;
//負責讀，讀取從上報的offset，并根據offset從Broker Master讀取commitlog
private ReadSocketService readSocketService;

2.4.1、ReadSocketService接收讀請求

readSocketService.run方法如下：

while (!this.isStopped()) {
	try {
		this.selector.select(1000);
		//處理讀事件
		boolean ok = this.processReadEvent();
		if (!ok) {
			HAConnection.log.error("processReadEvent error");
			break;
		}
		long interval = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now() - this.lastReadTimestamp;
		if (interval > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaHousekeepingInterval()) {
			log.warn("ha housekeeping, found this connection[" + HAConnection.this.clientAddr + "] expired, " + interval);
			break;
		}
	} catch (Exception e) {
		HAConnection.log.error(this.getServiceName() + " service has exception.", e);
		break;
	}
}

processReadEvent的邏輯如下：

int readSize = this.socketChannel.read(this.byteBufferRead);
if (readSize > 0) {
	readSizeZeroTimes = 0;
	this.lastReadTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
	if ((this.byteBufferRead.position() - this.processPostion) >= 8) {
		int pos = this.byteBufferRead.position() - (this.byteBufferRead.position() % 8);
		//擷取slave 請求的offset
		long readOffset = this.byteBufferRead.getLong(pos - 8);
		this.processPostion = pos;


		HAConnection.this.slaveAckOffset = readOffset;
		if (HAConnection.this.slaveRequestOffset < 0) {
			HAConnection.this.slaveRequestOffset = readOffset;
			log.info("slave[" + HAConnection.this.clientAddr + "] request offset " + readOffset);
		}
		//如果是同步複制的話，判斷請求的offset是否push2SlaveMaxOffset相同，相同的話則喚醒master GroupTransferService
		HAConnection.this.haService.notifyTransferSome(HAConnection.this.slaveAckOffset);
	}
}

可以看到processReadEvent邏輯很簡單，就是從ByteBuffer中解析出offset，然後設定HAConnection.this.slaveRequestOffset；

2.4.2、WriteSocketService進行寫處理

Broker Master通過HAConnection.WriteSocketService進行CommitLog的讀取，run方法主邏輯如下：

this.selector.select(1000);
//nextTransferFromWhere下次傳輸commitLog的起始位置
if (-1 == this.nextTransferFromWhere) {
	if (0 == HAConnection.this.slaveRequestOffset) {
		long masterOffset = HAConnection.this.haService.getDefaultMessageStore().getCommitLog().getMaxOffset();
		masterOffset =
			masterOffset
				- (masterOffset % HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig()
				.getMapedFileSizeCommitLog());


		if (masterOffset < 0) {
			masterOffset = 0;
		}


		this.nextTransferFromWhere = masterOffset;
	} else {
		this.nextTransferFromWhere = HAConnection.this.slaveRequestOffset;
	}


	log.info("master transfer data from " + this.nextTransferFromWhere + " to slave[" + HAConnection.this.clientAddr
		+ "], and slave request " + HAConnection.this.slaveRequestOffset);
}


//擷取commitLog資料
SelectMappedBufferResult selectResult = HAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
//擷取commitLog資料
SelectMappedBufferResult selectResult =
	HAConnection.this.haService.getDefaultMessageStore().getCommitLogData(this.nextTransferFromWhere);
if (selectResult != null) {
	int size = selectResult.getSize();
	if (size > HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize()) {
		size = HAConnection.this.haService.getDefaultMessageStore().getMessageStoreConfig().getHaTransferBatchSize();
	}


	long thisOffset = this.nextTransferFromWhere;
	this.nextTransferFromWhere += size;


	selectResult.getByteBuffer().limit(size);
	this.selectMappedBufferResult = selectResult;


	// Build Header
	this.byteBufferHeader.position(0);
	this.byteBufferHeader.limit(headerSize);
	this.byteBufferHeader.putLong(thisOffset);
	this.byteBufferHeader.putInt(size);
	this.byteBufferHeader.flip();
	//nio發送commitlog
	this.lastWriteOver = this.transferData();
} else {
    //如果沒有擷取到commitLog資料，等待100ms
	HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(1

這裡面主要包括擷取CommitLog資料、發送CommitLog資料這兩個步驟。

2.4.2.1、擷取CommitLog資料

public SelectMappedBufferResult getData(final long offset, final boolean returnFirstOnNotFound) {
	int mappedFileSize = this.defaultMessageStore.getMessageStoreConfig().getMapedFileSizeCommitLog();
	MappedFile mappedFile = this.mappedFileQueue.findMappedFileByOffset(offset, returnFirstOnNotFound);
	if (mappedFile != null) {
		int pos = (int) (offset % mappedFileSize);
		SelectMappedBufferResult result = mappedFile.selectMappedBuffer(pos);
		return result;
	}
	return null;
}
public SelectMappedBufferResult selectMappedBuffer(int pos) {
	int readPosition = getReadPosition();
	if (pos < readPosition && pos >= 0) {
		if (this.hold()) {
			ByteBuffer byteBuffer = this.mappedByteBuffer.slice();
			byteBuffer.position(pos);
			int size = readPosition - pos;
			ByteBuffer byteBufferNew = byteBuffer.slice();
			byteBufferNew.limit(size);
			return new SelectMappedBufferResult(this.fileFromOffset + pos, byteBufferNew, size, this);
		}
	}
	return null;
}

可以看到最終還是根據offset從MappedFile讀取資料。

2.4.2.2、發送CommitLog資料

資料主要包括header、body兩部分，資料發送的話還是通過NIO來實作，主要代碼如下：

// Build Header
this.byteBufferHeader.position(0);
this.byteBufferHeader.limit(headerSize);
this.byteBufferHeader.putLong(thisOffset);
this.byteBufferHeader.putInt(size);
this.byteBufferHeader.flip();


int writeSize = this.socketChannel.write(this.byteBufferHeader);
// Write Body
if (!this.byteBufferHeader.hasRemaining()) {
	while (this.selectMappedBufferResult.getByteBuffer().hasRemaining()) {
		int writeSize = this.socketChannel.write(this.selectMappedBufferResult.getByteBuffer());
		if (writeSize > 0) {
			writeSizeZeroTimes = 0;
			this.lastWriteTimestamp = HAConnection.this.haService.getDefaultMessageStore().getSystemClock().now();
		} else if (writeSize == 0) {
			if (++writeSizeZeroTimes >= 3) {
				break;
			}
		} else {
			throw new Exception("ha master write body error < 0");
		}
	}
}

CommitLog主從發送完成後，Broker Slave則會監聽讀事件、擷取CommitLog資料，并進行CommitLog的寫入。

2.5、HAClient processReadEvent

在主從建立連接配接後，從注冊了可讀事件，目的就是讀取從Broker Master傳回的CommitLog資料，對應的方法為HAClient.processReadEvent：

int readSize = this.socketChannel.read(this.byteBufferRead);
if (readSize > 0) {
	lastWriteTimestamp = HAService.this.defaultMessageStore.getSystemClock().now();
	readSizeZeroTimes = 0;
	boolean result = this.dispatchReadRequest();
	if (!result) {
		log.error("HAClient, dispatchReadRequest error");
		return false;
	}
}

dispatchReadRequest方法如下：

//讀取傳回的body data
byte[] bodyData = new byte[bodySize];
this.byteBufferRead.position(this.dispatchPostion + msgHeaderSize);
this.byteBufferRead.get(bodyData);


HAService.this.defaultMessageStore.appendToCommitLog(masterPhyOffset, bodyData);


this.byteBufferRead.position(readSocketPos);
this.dispatchPostion += msgHeaderSize + bodySize;


//上報從的offset
if (!reportSlaveMaxOffsetPlus()) {
	return false;

裡面的核心邏輯主要包括如下三個步驟：

  1. 從byteBufferRead中讀取CommitLog資料；
  1. 調用defaultMessageStore.appendToCommitLog方法，将資料寫入到MappedFile檔案，寫入方法如下： 
public boolean appendToCommitLog(long startOffset, byte[] data) {
	//将資料寫到commitlog，同普通消息的存儲
	boolean result = this.commitLog.appendData(startOffset, data);
	//喚醒reputMessageService，建構consumeQueue,index
	this.reputMessageService.wakeup();
	return result;
}
  1. 上報從新的offset，也是讀取MappedFile的offset，然後上報Broker Master；

2.6、同步複制

上面主要介紹了Broker的異步複制，下面再來看下Broker的同步複制的實作。同步複制的整體流程圖如下：

通過源碼分析RocketMQ主從複制原理

大概說明如下：

  1. producer發送消息到broker，broker進行消息的存儲，将消息寫入到commitLog；
  2. broker master寫消息線程喚醒WriteSocketService線程，查詢commitLog資料，然後發送到從。在WriteSocketService擷取commitLog時，如果沒有擷取到commitLog資料，會等待100ms。是以當commitLog新寫入資料的時候，會喚醒WriteSocketService，然後查詢commitLog資料，發送到從。
  3. broker master建立GroupCommitRequest，同步等待主從複制完成；
  4. 從接受新的commitLog資料，然後寫commitLog資料，并傳回新的slave offset到主；
  5. 主更新push2SlaveMaxOffset，并判斷push2SlaveMaxOffset是否大于等于主從複制請求的offset，如果大于等于的話，則認為主從複制完成，傳回commitLog.handleHA方法成功，進而傳回消息儲存成功。

對應的代碼入口為CommitLog.handleHA方法。

public void handleHA(AppendMessageResult result, PutMessageResult putMessageResult, MessageExt messageExt) {
	//如果是broker主，并且是同步複制的話
	if (BrokerRole.SYNC_MASTER == this.defaultMessageStore.getMessageStoreConfig().getBrokerRole()) {
		//擷取HAService
		HAService service = this.defaultMessageStore.getHaService();
		//擷取Message上的MessageConst.PROPERTY_WAIT_STORE_MSG_OK，預設是需要等待主從複制完成
		if (messageExt.isWaitStoreMsgOK()) {
			/**
			 * 判斷從是否可用，判斷的邏輯是：（主offset-push2SlaveMaxOffset<1024 * 1024 * 256），也就是如果主從的offset差的太多，
			 * 則認為從不可用， Tell the producer, slave not available
			 * 這裡的result = mappedFile.appendMessage(msg, this.appendMessageCallback);
			 */
			if (service.isSlaveOK(result.getWroteOffset() + result.getWroteBytes())) {
				//組裝GroupCommitRequest，nextOffset=result.getWroteOffset() + result.getWroteBytes()，這裡的nextOffset指的就是從要寫到的offset
				GroupCommitRequest request = new GroupCommitRequest(result.getWroteOffset() + result.getWroteBytes());
				                    /**
                * 調用的是this.groupTransferService.putRequest(request);将request放到requestsWrite list中。
                  * HAService持有GroupTransferService groupTransferService引用;
                */
				service.putRequest(request);
				 /**
                     * 喚醒的是WriteSocketService，查詢commitLog資料，然後發送到從。
                     * 在WriteSocketService擷取commitLog時，如果沒有擷取到commitLog資料，等待100ms
                     * HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);
                     * 是以當commitLog新寫入資料的時候，會喚醒WriteSocketService，然後查詢commitLog資料，發送到從。
                     */
				service.getWaitNotifyObject().wakeupAll();


				//等待同步複制完成，判斷邏輯是： HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
				boolean flushOK =
					request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());


				//如果同步複制失敗的話，設定putMessageResult中的狀态為同步從逾時
				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);
			}
		}
	}

2.6.1、GroupTransferService啟動

在HAService啟動的時候，啟動了GroupTransferService線程，代碼如下：

public void run() {
	while (!this.isStopped()) {
		this.waitForRunning(10);
		this.doWaitTransfer();
	}
}
private void doWaitTransfer() {
	synchronized (this.requestsRead) {
		if (!this.requestsRead.isEmpty()) {
			for (CommitLog.GroupCommitRequest req : this.requestsRead) {
				/**
				 * req.getNextOffset：result.getWroteOffset() + result.getWroteBytes()
				 * push2SlaveMaxOffset：
				 */
				boolean transferOK = HAService.this.push2SlaveMaxOffset.get() >= req.getNextOffset();
				//在這循環5次，最多等待5s，因為slave 心跳間隔預設5s
				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());
				}
				//主從複制完成，喚醒handleHA後續操作
                req.wakeupCustomer(transferOK);
			}
			this.requestsRead.clear();
		}
	}
}

wakeupCustomer：

public void wakeupCustomer(final boolean flushOK) {
    this.flushOK = flushOK;
    this.countDownLatch.countDown();
}

2.6.2、喚醒WriteSocketService

service.getWaitNotifyObject().wakeupAll();

喚醒的是WriteSocketService，查詢commitLog資料，然後發送到從。在WriteSocketService擷取commitLog時，如果沒有擷取到commitLog資料，等待100ms。HAConnection.this.haService.getWaitNotifyObject().allWaitForRunning(100);是以當commitLog新寫入資料的時候，會喚醒WriteSocketService，然後查詢commitLog資料，發送到從。

2.6.3、同步等待，直到複制完成

boolean flushOK =
	request.waitForFlush(this.defaultMessageStore.getMessageStoreConfig().getSyncFlushTimeout());


public boolean waitForFlush(long timeout) {
	try {
		//等待同步複制完成
		this.countDownLatch.await(timeout, TimeUnit.MILLISECONDS);
		return this.flushOK;
	} catch (InterruptedException e) {
		log.error("Interrupted", e);
		return false;
	}
}
}

三、中繼資料的複制

broker中繼資料的複制，主要包括topicConfig、consumerOffset、delayOffset、subscriptionGroup這幾部分，整體流程圖如下：

通過源碼分析RocketMQ主從複制原理

從broker通過單獨的線程，每隔10s進行一次中繼資料的複制 ，代碼入口為：BrokerController.start -> SlaveSynchronize.syncAll：

slaveSyncFuture = this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        try {
            //10s 進行一次主從同步
            BrokerController.this.slaveSynchronize.syncAll();
        }
        catch (Throwable e) {
            log.error("ScheduledTask SlaveSynchronize syncAll error.", e);
        }
    }
}, 1000 * 3, 1000 * 10, TimeUnit.MILLISECONDS);


public void syncAll() {
    this.syncTopicConfig();
    this.syncConsumerOffset();
    this.syncDelayOffset();
    this.syncSubscriptionGroupConfig();
}

3.1、syncTopicConfig

//從Master擷取TopicConfig資訊，最終調用的是AdminBrokerProcessor.getAllTopicConfig
TopicConfigSerializeWrapper topicWrapper =
    this.brokerController.getBrokerOuterAPI().getAllTopicConfig(masterAddrBak);
if (!this.brokerController.getTopicConfigManager().getDataVersion()
    .equals(topicWrapper.getDataVersion())) {
    this.brokerController.getTopicConfigManager().getDataVersion()
        .assignNewOne(topicWrapper.getDataVersion());
    this.brokerController.getTopicConfigManager().getTopicConfigTable().clear();
    this.brokerController.getTopicConfigManager().getTopicConfigTable()
        .putAll(topicWrapper.getTopicConfigTable());
 //将topicConfig進行持久化，對應的檔案為topics.json
    this.brokerController.getTopicConfigManager().persist();
    log.info("Update slave topic config from master, {}", masterAddrBak)

3.2、syncConsumerOffset

//從"主Broker"擷取ConsumerOffset
ConsumerOffsetSerializeWrapper offsetWrapper =
        this.brokerController.getBrokerOuterAPI().getAllConsumerOffset(masterAddrBak);
//設定從的offsetTable
this.brokerController.getConsumerOffsetManager().getOffsetTable()
                    .putAll(offsetWrapper.getOffsetTable());
//并持久化到從的consumerOffset.json檔案中
this.brokerController.getConsumerOffsetManager().persist();

3.3、syncDelayOffset

String delayOffset = this.brokerController.getBrokerOuterAPI().getAllDelayOffset(masterAddrBak);
String fileName = StorePathConfigHelper.getDelayOffsetStorePath(this.brokerController
.getMessageStoreConfig().getStorePathRootDir());
 MixAll.string2File(delayOffset, fileName);

3.4、syncSubscriptionGroupConfig

SubscriptionGroupWrapper subscriptionWrapper =this.brokerController.getBrokerOuterAPI().getAllSubscriptionGroupConfig(masterAddrBak);
SubscriptionGroupManager subscriptionGroupManager =this.brokerController.getSubscriptionGroupManager();
subscriptionGroupManager.getDataVersion().assignNewOne(subscriptionWrapper.getDataVersion());
subscriptionGroupManager.getSubscriptionGroupTable().clear();
subscriptionGroupManager.getSubscriptionGroupTable().putAll(subscriptionWrapper.getSubscriptionGroupTable());
subscriptionGroupManager.persist();

四、思考與收獲

通過上面的分享，我們基本上了解了RocketMQ的主從複制原理，其中有些思想我們可以後續借鑒下：

  1. 在功能設計的時候将中繼資料、程式資料分開管理；
  2. 主從複制的時候，基本思想都是從請求主，請求時帶上offset，然後主查詢資料傳回從，從再執行；mysql的主從複制、redis的主從複制基本也是這樣；
  3. 主從複制包括異步複制、同步複制兩種方式，可以通過配置來決定使用哪種同步方式，這個需要根據實際業務場景來決定；
  4. 主從複制線程盡量和消息寫線程或者主線程分開；

由于時間、精力有限，難免會有纰漏、考慮不到之處，如有問題歡迎溝通、交流。

主從複制原理 通過源碼 源碼分析 複制原理 主從複制 RocketMQ
上一篇: （新華述評）堅決鞏固住來之不易的戰“疫”重大成果——貫徹落實中共中央政治局常委會會議精神系列述評之二
下一篇: 「Java基礎」萬字詳解ArrayList和LinkedList

繼續閱讀