RocketMQ 4.9.1 源码分析：HA 模块 Master 读写处理揭秘

2023-09-23 15:53:05

RocketMQ HA：探索 Master 读写处理的奥秘

在分布式消息中间件中，高可用性 (HA) 至关重要，而 RocketMQ 的 HA 模块正是为实现这一目标而生的。本文将带你深入探究 RocketMQ 4.9.1 版本中 Master 读写处理的细节，揭开它如何保障数据完整性和一致性的秘密。

Master 读写处理

Master 在收到客户端请求时，会根据请求类型采取不同的处理方式：

1. 读请求处理

Master 会先检查本地存储中是否有对应消息，如果有，则直接返回。否则，它会向 Slave 发送读取请求，让 Slave 从其本地存储中读取数据并返回给客户端。

2. 写请求处理

Master 将数据写入本地存储后，会将数据同步到 Slave。这个过程包含以下步骤：

Master 创建一个复制请求，包含数据和 Slave 地址。
Master 将复制请求发送给 Slave。
Slave 收到复制请求后，将其写入本地存储。

数据同步机制

RocketMQ HA 模块采用两种数据同步机制：

1. 同步复制

Master 写入数据后立即同步到 Slave，保证数据一致性，但会增加延迟。

2. 异步复制

Master 不立即同步数据，而是由 Slave 定期从 Master 同步，从而降低延迟，但可能导致数据不一致。

故障转移处理

当 Master 故障时，Slave 会自动选举出一个新的 Master，并同步数据。这个过程如下：

Slave 检测到 Master 故障后，发送选举请求。
Slave 投票选举出一个新的 Master。
新 Master 向其他 Slave 发送同步请求。
Slave 从新 Master 同步数据。

代码示例

// Master 读写处理
public void handleWriteMessage(final Context context, final Message msg) {
    // 写入本地存储
    long phyOffset = writeBodyToDisk(context, msg);
    // 构建复制请求
    ReplicateRequest replicateRequest = buildReplicateRequest(msg, phyOffset);
    // 发送复制请求
    doAsyncReplica(replicateRequest);
}

// 同步复制
public void doSyncReplica(final ReplicateRequest replicateRequest) {
    // 获取 Slave 地址
    List<String> brokerAddrs = masterReplicate.getAddresses();
    // 向每个 Slave 发送复制请求
    for (String brokerAddr : brokerAddrs) {
        putReplicateRequest(brokerAddr, replicateRequest);
    }
}