Talk about RocketMQ master-slave replication

RocketMQ master-slave replication is one of RocketMQ's high availability mechanisms. Data can be replicated from the master node to one or more slave nodes.

In this article, we will talk about RocketMQ's master-slave replication. I hope that after reading this article, you will be able to understand the essence of master-slave replication.

picture

1. Synchronous and asynchronous

In RocketMQ's cluster mode, Broker is divided into Master and Slave. A Master can correspond to multiple Slaves, but a Slave can only correspond to one Master.

Each Broker establishes a persistent connection with all nodes in the Name Server cluster and periodically registers Topic information to all Name Servers.

picture

The Master node is responsible for receiving write requests from clients and persisting messages to disk, while the Slave node is responsible for copying message data from the Master node and keeping it synchronized with the Master node.

1. Synchronous replication

picture

Each Master is configured with a Slave, and there are multiple Master-Slave pairs. HA uses a synchronous dual-write method, that is, only when both the master and the slave are written successfully, will success be returned to the application.

The advantages and disadvantages of this model are as follows:

• Advantages: There is no single point of failure for data and services. When the Master fails, messages are not delayed. Both service availability and data availability are very high.
• Disadvantages: The performance is slightly lower than that of the asynchronous replication mode (about 10% lower), the RT for sending a single message is slightly higher, and in the current version, the backup machine cannot automatically switch to the master after the master node goes down.

2. Asynchronous replication

picture

Each Master is configured with a Slave, and there are multiple Master-Slave pairs. HA uses asynchronous replication, and there is a short message delay (millisecond level) between the master and the slave. The advantages and disadvantages of this mode are as follows:

• Advantages: Even if the disk is damaged, very few messages are lost, and the real-time nature of the messages will not be affected. After the Master goes down, consumers can still consume from the Slave. This process is transparent to the application and does not require manual intervention. The performance is almost the same as the multi-Master mode.
• Disadvantages: If the Master crashes or the disk is damaged, a small amount of messages will be lost.

The replication process is divided into two parts: metadata replication and message data replication.

• Master-slave server synchronization topics, consumer progress, delayed consumption progress, consumer configuration data
• Master-slave server synchronization message data

2. Metadata Replication

The Slave Broker scheduled task synchronizes metadata every 10 seconds, including topics, consumption progress, delayed consumption progress, and consumer configuration.

picture

When synchronizing a topic, the Slave Broker sends an RPC request to the Master Broker. After the data is returned, it is first added to the local cache and then persisted locally.

picture

3. Message Data Replication

The figure below is a flowchart of Master and Slave message data synchronization.

picture

1. After the Master is started, it listens to the specified port;

After the Master is started, it creates the AcceptSocketService service to create a TCP link from the client to the server.

picture

RocketMQ abstracts the connection object HAConnection. HAConnection starts two threads, one for read service and one for write service:

• Read service: Processes requests sent by Slave
• Write service: used to transfer data to Slave

picture

2. After the Slave is started, it tries to connect to the Master and establish a TCP connection;

HAClient is the core class of the client Slave, responsible for establishing connections and data interaction with the Master.

picture

After starting, the client first tries to connect to the Master, queries the maximum physical offset in the current message storage, and stores it in the variable currentReportedOffset.

3. Slave reports the offset of the pulled message to the Master;

picture

The data format for reporting progress is a Long type Offset, 8 bytes, which is very concise.

picture

After sending to the Socket buffer, modify the last write time lastWriteTimestamp.

4. The Master parses the request offset and retrieves all messages after the offset from the message file;

When the Slave reports data to the Master, the SelectionKey.OP_READ event is triggered, and the Master passes the request to the ReadSocketService service for processing:

picture

When the Slave Broker passes the maxPhyOffset of its own commitlog, the Master will immediately interrupt selector.select(1000) and execute the processReadEvent method.

picture

The core logic of the processReadEvent method is to set the current progress offset of the Slave and then notify the replication thread of the current replication progress.

The write service WriteSocketService retrieves all messages after the offset from the message file (limited by the size of the transmission batch data) and sends the message data to the Slave.

picture

5. The slave receives the data and appends the message data to the message file commitlog.

picture

First, the dispatchReadRequest method is called in the HAClient class to parse the message data;

picture

Then append the message data to the local message storage.

picture

4. Implementation of synchronization

From the data replication flow chart, we find that data replication itself is an asynchronous execution, but how is synchronization achieved?

After receiving the request to write a message, the Master Broker calls the aysncPutMessage method of Commitlog to write the message.

picture

In this code, after commitLog executes appendMessage, it needs to execute two tasks: disk flushing and synchronous replication.

However, these two tasks are not executed synchronously, but asynchronously, using the asynchronous artifact CompletableFuture.

When the HAConnection read service receives progress feedback from the Slave and finds that the message data has been successfully copied, it wakes up the future.

picture

Finally, Broker assembles the response command and returns it to the client.

V. Conclusion

The implementation idea of ​​RocketMQ master-slave replication is very simple. The slave starts a thread, continuously pulls data in the Commit Log from the Master, and then asynchronously builds the Consume Queue data structure.

The key points are as follows:

1. Master-slave replication includes metadata replication and message data replication;

2. Metadata replication

The Slave Broker scheduled task sends an RPC request to the Master Broker every 10 seconds to synchronize the metadata to the cache and then persist it to the disk;

3. Message data replication

1. Master starts listening on the specified port
2. Slave starts HaClient service and creates TCP link with Master
3. Slave reports storage progress to Master
4. The Master receives the progress, retrieves all messages after the offset in the message file, and transmits them to the Slave
5. After receiving the data, the Slave appends the message data to the local message storage.

4. Synchronization Implementation

After commitLog executes appendMessage, it needs to execute two tasks: disk flushing and synchronous replication. The asynchronous artifact CompletableFuture is used here.

When the HAConnection read service receives progress feedback from the Slave and finds that the message data has been successfully copied, it wakes up the future. Finally, the Broker assembles the response command and returns the response command to the client.