RocketMQ DLedger初识
前言
众所周知，作为一个出色的分布式消息中间件，RocketMQ 在全球范围内获得了广泛的应用，那么作为一个分布式消息中间件，最重要的是什么？
协议？持久化？消息分发实现？高可用？高可靠？
好的协议可以保证通讯的稳定，持久化可以保证数据的存储，消息分发实现可以结合多场景加速业务，高可用可以保证业务大量运行，高可靠可以保证业务的持续运行。
今天，我们想谈一谈 RocketMQ 的高可用机制
常见的消息中间件集群实现方式
- 主从复制方式（非对称）：这种方式下，一主多从的架构被广泛应用。消息中间件的写入操作只能在主节点上进行，而读取操作在所有节点上都可以执行。主节点负责数据的同步复制到所有从节点上，当主节点出现故障时，一个备份从节点会自动地竞选为新的主节点，以保证系统的持续运行。这种方式下，主节点和从节点之间的延时可能会导致数据不一致或者消息丢失。
- 对等网络方式（对称）：这种方式下，每个节点都可以承担读写操作的任务，数据可以通过节点之间的同步来保持一致性。每个节点的能力相同，不存在主从的概念。当一个节点出现故障时，其他节点会自动接管它的任务，因此系统没有单点故障。这种方式下，网络的复杂性和节点间的通信有可能造成性能瓶颈。
RocktMQ 高可用方式
在4.5版本之前，RocketMQ 不支持节点的自动晋升，那么如果主节点挂了，未消费的数据会从从节点上被继续消费，但是这一组节点就失去了作用，无法再被写入，若集群中主节点数量较少，可能会引起故障，于是在4.5版本，升级支持 DLedger 模式完成自动选主。
我们现在看一下 DLedger 如何实现自动选主。
首先，我们要明白 DLedger 模式的本质便是使用了 Raft 算法，接着我们来看一下 RocketMQ 的代码实现。在启动 broke r时，会将 CommitLog 转换为 DLedgerCommitLog 类型，并添加 DLedgerRoleChangeHandler 处理器，那么我们来看看 CommitLog 与 DLedgerCommitLog 的区别
commitLog 是 RocketMQ 最核心的数据存储。它是一个顺序写的文件，用于存储 Producer 发送的消息和 Consumer 消费的消息，也就是全部通过消息中间件传递的消息。每一个写入 commitLog 的消息都会被分配一个唯一的 offset （偏移量），用于标识该条消息在 commitLog 中的位置。 commitLog 中消息的存储格式包括消息长度、消息属性（如是否压缩、是否顺序消费、是否是事务消息等）、消息体等信息。
public class CommitLog {
public final static int MESSAGE_MAGIC_CODE = -626843481;
protected static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);
// 空文件结束标识
protected final static int BLANK_MAGIC_CODE = -875286124;
// 文件队列，用于存储在磁盘上的消息
protected final MappedFileQueue mappedFileQueue;
// 默认的消息存储对象
protected final DefaultMessageStore defaultMessageStore;
// 用于刷盘和提交的服务
private final FlushCommitLogService flushCommitLogService;
// 如果启用了TransientStorePool，我们必须在固定的时间内将消息刷新到FileChannel
private final FlushCommitLogService commitLogService;
// 消息发送的回调函数
private final AppendMessageCallback appendMessageCallback;
private final ThreadLocal<MessageExtBatchEncoder> batchEncoderThreadLocal;
// 用于存储每个 topic 的队列
protected HashMap<String/* topi c-queueid */, Long/* offset */> topicQueueTable = new HashMap<String, Long>(1024);
// 确认偏移量
protected volatile long confirmOffset = -1L;
// 加锁期间的起始时间
private volatile long beginTimeInLock = 0;
// 消息发送的锁，用于防止并发发送。
protected final PutMessageLock putMessageLock;
}
DLedgerCommitLog 是 RocketMQ 用作持久化存储的一种实现方式，它基于Apache DistributedLog (DLedger) 实现了高可靠、高性能的分布式日志存储，也正是它，使得 CommitLog 拥有了选举复制的能力。
public class DLedgerCommitLog extends CommitLog {
// DLedger实例
private final DLedgerServer dLedgerServer;
// DLedger的配置信息
private final DLedgerConfig dLedgerConfig;
// 用于存储mmap文件的存储类
private final DLedgerMmapFileStore dLedgerFileStore;
// mmap文件列表
private final MmapFileList dLedgerFileList;
// id标识代理角色，0表示主，其他表示从
private final int id;
private final MessageSerializer messageSerializer;
// 进入DLedger锁的开始时间
private volatile long beginTimeInDledgerLock = 0;
// 分隔旧的commitlog和DLedgerCommitlog的偏移量
private long dividedCommitlogOffset = -1;
// 是否正在恢复旧的commitlog
private boolean isInrecoveringOldCommitlog = false;
}
了解完 CommitLog，我们回到 broker 的启动，核心类 BrokerController 的 initialize 方法，首先会根据配置生成一个 DefaultMessageStore，这里会判断当前是否支持 DLedgerCommitLog，如果支持，则会创建一个 DLedgerRoleChangeHandler 对象并注册为 leader 选举的回调方法。接着与老版本一致会创建一个 BrokerStats 对象和一个 MessageStorePluginContext 对象。最后，会将 CommitLogDispatcherCalcBitMap 对象添加到 MessageStore 的 DispatcherList 中。
public class BrokerController {
public boolean initialize() throws CloneNotSupportedException {
……
if (result) {
try {
this.messageStore =
new DefaultMessageStore(this.messageStoreConfig, this.brokerStatsManager, this.messageArrivingListener,
this.brokerConfig);
// 如果支持DLegerCommitLog
if (messageStoreConfig.isEnableDLegerCommitLog()) {
// 创建DLedgerRoleChangeHandler
DLedgerRoleChangeHandler roleChangeHandler = new DLedgerRoleChangeHandler(this, (DefaultMessageStore) messageStore);
// 将CommitLog转换为DLegerCommitLog，并添加DLedgerRoleChangeHandler处理器
((DLedgerCommitLog)((DefaultMessageStore) messageStore).getCommitLog()).getdLedgerServer().getdLedgerLeaderElector().addRoleChangeHandler(roleChangeHandler);
}
this.brokerStats = new BrokerStats((DefaultMessageStore) this.messageStore);
// 加载插件
MessageStorePluginContext context = new MessageStorePluginContext(messageStoreConfig, brokerStatsManager, messageArrivingListener, brokerConfig);
this.messageStore = MessageStoreFactory.build(context, this.messageStore);
this.messageStore.getDispatcherList().addFirst(new CommitLogDispatcherCalcBitMap(this.brokerConfig, this.consumerFilterManager));
}
……
}
}
接着我们进入 DLedgerCommitLog 的代码，这里可以看到使用了 openmessaging 包下的 DLedgerServer 组件
public class DLedgerCommitLog extends CommitLog {
@Override
public void start() {
// 启动dLedgerServer
dLedgerServer.startup();
}
}
基于 dLedgerLeaderElector 做了 Leader 选举的操作
public class DLedgerServer extends AbstractDLedgerServer {
public synchronized void startup() {
if (!isStarted) {
this.dLedgerStore.startup();
this.fsmCaller.ifPresent(x -> {
// 启动状态机调用程序并加载现有快照以进行数据恢复
x.start();
x.getSnapshotManager().loadSnapshot();
});
if (RpcServiceMode.EXCLUSIVE.equals(this.rpcServiceMode)) {
this.dLedgerRpcService.startup();
}
this.dLedgerEntryPusher.startup();
// 进行leader选举
this.dLedgerLeaderElector.startup();
executorService.scheduleAtFixedRate(this::checkPreferredLeader, 1000, 1000, TimeUnit.MILLISECONDS);
isStarted = true;
}
}
}
启动状态机
public class DLedgerLeaderElector {
public void startup() {
// 启动状态机
stateMaintainer.start();
for (RoleChangeHandler roleChangeHandler : roleChangeHandlers) {
roleChangeHandler.startup();
}
}
}
状态机
public class StateMaintainer extends ShutdownAbleThread {
public StateMaintainer(String name, Logger logger) {
super(name, logger);
}
@Override
public void doWork() {
try {
// 是否支持Leader选举
if (DLedgerLeaderElector.this.dLedgerConfig.isEnableLeaderElector()) {
DLedgerLeaderElector.this.refreshIntervals(dLedgerConfig);
// 状态机核心方法
DLedgerLeaderElector.this.maintainState();
}
sleep(10);
} catch (Throwable t) {
DLedgerLeaderElector.LOGGER.error("Error in heartbeat", t);
}
}
}
状态机核心方法，Raft 中的三个角色，Leader、Follower、Candidate，这里对三个角色不做过多叙述，可以参见《In Search of an Understandable Consensus Algorithm》 一文
public class DLedgerLeaderElector {
private void maintainState() throws Exception {
// Leader角色
if (memberState.isLeader()) {
maintainAsLeader();
} else if (memberState.isFollower()) {
// Follower角色
maintainAsFollower();
} else {
// Candidate角色
maintainAsCandidate();
}
}
}
接下来我们来看 raft 的核心实现，首先看 Leader 角色的实现代码 当上次发送心跳时间大于心跳包间隔时，会重新发送心跳
public class DLedgerLeaderElector {
private void maintainAsLeader() throws Exception {
// 上次发送心跳时间是否大于心跳包间隔时间
if (DLedgerUtils.elapsed(lastSendHeartBeatTime) > heartBeatTimeIntervalMs) {
// 任期
long term;
// 主节点id
String leaderId;
synchronized (memberState) {
if (!memberState.isLeader()) {
//stop sending
return;
}
term = memberState.currTerm();
leaderId = memberState.getLeaderId();
lastSendHeartBeatTime = System.currentTimeMillis();
}
// 发送心跳
sendHeartbeats(term, leaderId);
}
}
}
当我们看心跳代码前，首先回顾下 raft 理论中对于心跳返回的描述
再让我们回到代码
public class DLedgerLeaderElector {
private void sendHeartbeats(long term, String leaderId) throws Exception {
……
for (String id : memberState.getPeerMap().keySet()) {
if (memberState.getSelfId().equals(id)) {
continue;
}
HeartBeatRequest heartBeatRequest = new HeartBeatRequest();
heartBeatRequest.setGroup(memberState.getGroup());
heartBeatRequest.setLocalId(memberState.getSelfId());
heartBeatRequest.setRemoteId(id);
// 主节点id
heartBeatRequest.setLeaderId(leaderId);
// 任期
heartBeatRequest.setTerm(term);
// 异步发送心跳
CompletableFuture<HeartBeatResponse> future = dLedgerRpcService.heartBeat(heartBeatRequest);
future.whenComplete((HeartBeatResponse x, Throwable ex) -> {
try {
if (ex != null) {
memberState.getPeersLiveTable().put(id, Boolean.FALSE);
throw ex;
}
// 获取心跳结果
switch (DLedgerResponseCode.valueOf(x.getCode())) {
// 成功
case SUCCESS:
succNum.incrementAndGet();
break;
// 主节点的Term小于从节点
case EXPIRED_TERM:
maxTerm.set(x.getTerm());
break;
// 从节点的主节点非当前节点
case INCONSISTENT_LEADER:
inconsistLeader.compareAndSet(false, true);
break;
// 从节点尚未准备完毕
case TERM_NOT_READY:
notReadyNum.incrementAndGet();
break;
default:
break;
}
……
} catch (Throwable t) {
LOGGER.error("heartbeat response failed", t);
} finally {
allNum.incrementAndGet();
if (allNum.get() == memberState.peerSize()) {
beatLatch.countDown();
}
}
});
}
long voteResultWaitTime = 10;
beatLatch.await(heartBeatTimeIntervalMs - voteResultWaitTime, TimeUnit.MILLISECONDS);
Thread.sleep(voteResultWaitTime);
// 当从节点返回的term大于自身时，直接退化为candidate
if (maxTerm.get() > term) {
LOGGER.warn("[{}] currentTerm{} is not the biggest={}, deal with it", memberState.getSelfId(), term, maxTerm.get());
changeRoleToCandidate(maxTerm.get());
return;
}
// 当半数以上正常返回心跳时，Leader状态正常，重置心跳时间
if (memberState.isQuorum(succNum.get())) {
lastSuccHeartBeatTime = System.currentTimeMillis();
} else {
LOGGER.info("[{}] Parse heartbeat responses in cost={} term={} allNum={} succNum={} notReadyNum={} inconsistLeader={} maxTerm={} peerSize={} lastSuccHeartBeatTime={}",
memberState.getSelfId(), DLedgerUtils.elapsed(startHeartbeatTimeMs), term, allNum.get(), succNum.get(), notReadyNum.get(), inconsistLeader.get(), maxTerm.get(), memberState.peerSize(), new Timestamp(lastSuccHeartBeatTime));
// 当正常心跳 + 未准备的心跳大于半数时，立即发送心跳
if (memberState.isQuorum(succNum.get() + notReadyNum.get())) {
lastSendHeartBeatTime = -1;
// 当从节点中有其他主节点时，直接退化为candidate
} else if (inconsistLeader.get()) {
changeRoleToCandidate(term);
// 如果上次心跳包时间大于3次心跳间隔时间，直接退化为candidate
} else if (DLedgerUtils.elapsed(lastSuccHeartBeatTime) > (long) maxHeartBeatLeak * heartBeatTimeIntervalMs) {
changeRoleToCandidate(term);
}
}
}
}
从简单上来说，Leader 只做了一件事，那就是发送心跳，根据心跳结果判断服务是否正常及自己的地位。接着让我们看 Follower 做了什么，Follower 在选举中的流程比较简单
public class DLedgerLeaderElector {
private void maintainAsFollower() {
// 如果上次心跳时间大于2次心跳间隔
if (DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > 2L * heartBeatTimeIntervalMs) {
synchronized (memberState) {
// 如果当前角色是Follower，并且心跳大于3次心跳间隔，升级到candidate
if (memberState.isFollower() && DLedgerUtils.elapsed(lastLeaderHeartBeatTime) > (long) maxHeartBeatLeak * heartBeatTimeIntervalMs) {
LOGGER.info("[{}][HeartBeatTimeOut] lastLeaderHeartBeatTime: {} heartBeatTimeIntervalMs: {} lastLeader={}", memberState.getSelfId(), new Timestamp(lastLeaderHeartBeatTime), heartBeatTimeIntervalMs, memberState.getLeaderId());
changeRoleToCandidate(memberState.currTerm());
}
}
}
}
}
最后我们来看 Candidate 角色，这块的代码比较多，让我们逐行来进行分析
public class DLedgerLeaderElector {
private void maintainAsCandidate() throws Exception {
// 如果当前时间小于下次发起投票时间或者不应该立即发起投票，返回
if (System.currentTimeMillis() < nextTimeToRequestVote && !needIncreaseTermImmediately) {
return;
}
// 任期
long term;
// Leader节点的投票任期
long ledgerEndTerm;
// 日志的最大索引
long ledgerEndIndex;
// 如果不是Candidate，直接返回
if (!memberState.isCandidate()) {
return;
}
synchronized (memberState) {
if (!memberState.isCandidate()) {
return;
}
// 如果上次投票结果是等待下一轮或者需要立即投票
if (lastParseResult == VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT || needIncreaseTermImmediately) {
// 记录当前任期
long prevTerm = memberState.currTerm();
// 记录下一任期
term = memberState.nextTerm();
LOGGER.info("{}_[INCREASE_TERM] from {} to {}", memberState.getSelfId(), prevTerm, term);
// 将上次投票结果重置为等待重新投票
lastParseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
} else {
term = memberState.currTerm();
}
// 设置日志的最大索引
ledgerEndIndex = memberState.getLedgerEndIndex();
// 设置节点的投票任期
ledgerEndTerm = memberState.getLedgerEndTerm();
}
// 需要立即投票
if (needIncreaseTermImmediately) {
// 设置下次投票超时时间，这中间为了减少同时发起的概率，会有随机时间
nextTimeToRequestVote = getNextTimeToRequestVote();
// 重置需要立即投票标识
needIncreaseTermImmediately = false;
return;
}
// 发起投票申请，包含当前任期，自身维护的最大任期，自身维护的最大日志索引，直接给自己投票
final List<CompletableFuture<VoteResponse>> quorumVoteResponses = voteForQuorumResponses(term, ledgerEndTerm, ledgerEndIndex);
……
for (CompletableFuture<VoteResponse> future : quorumVoteResponses) {
future.whenComplete((VoteResponse x, Throwable ex) -> {
try {
if (ex != null) {
throw ex;
}
LOGGER.info("[{}][GetVoteResponse] {}", memberState.getSelfId(), JSON.toJSONString(x));
if (x.getVoteResult() != VoteResponse.RESULT.UNKNOWN) {
validNum.incrementAndGet();
}
synchronized (knownMaxTermInGroup) {
switch (x.getVoteResult()) {
// 赞成，成功数加一
case ACCEPT:
acceptedNum.incrementAndGet();
break;
// 被已有Leader的节点拒绝
case REJECT_ALREADY_HAS_LEADER:
alreadyHasLeader.compareAndSet(false, true);
break;
// 任期小于其他选举人
case REJECT_TERM_SMALL_THAN_LEDGER:
case REJECT_EXPIRED_VOTE_TERM:
if (x.getTerm() > knownMaxTermInGroup.get()) {
// 维护最大任期
knownMaxTermInGroup.set(x.getTerm());
}
break;
// 任期小于对方
case REJECT_EXPIRED_LEDGER_TERM:
// 日志小于对方
case REJECT_SMALL_LEDGER_END_INDEX:
biggerLedgerNum.incrementAndGet();
break;
// 对方尚未准备完成
case REJECT_TERM_NOT_READY:
notReadyTermNum.incrementAndGet();
break;
// 已投票
case REJECT_ALREADY_VOTED:
// 拒绝接受领导
case REJECT_TAKING_LEADERSHIP:
default:
break;
}
}
// 如果已经有leader或已接受的投票数量满足 quorum 或者已接受和未准备好的数量之和满足 quorum，释放阻塞状态
if (alreadyHasLeader.get()
|| memberState.isQuorum(acceptedNum.get())
|| memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {
voteLatch.countDown();
}
} catch (Throwable t) {
LOGGER.error("vote response failed", t);
} finally {
allNum.incrementAndGet();
// 所有异步请求结束时，释放阻塞状态
if (allNum.get() == memberState.peerSize()) {
voteLatch.countDown();
}
}
});
}
try {
// 生成一个随机数的阻塞时间
voteLatch.await(2000 + random.nextInt(maxVoteIntervalMs), TimeUnit.MILLISECONDS);
} catch (Throwable ignore) {
}
lastVoteCost = DLedgerUtils.elapsed(startVoteTimeMs);
VoteResponse.ParseResult parseResult;
if (knownMaxTermInGroup.get() > term) {
// 已知的最大任期比当前任期要大，则返回 WAIT_TO_VOTE_NEXT，并转变为Candidate
parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
nextTimeToRequestVote = getNextTimeToRequestVote();
changeRoleToCandidate(knownMaxTermInGroup.get());
} else if (alreadyHasLeader.get()) {
// 已经存在Leader，则返回 WAIT_TO_VOTE_NEXT
parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
nextTimeToRequestVote = getNextTimeToRequestVote() + (long) heartBeatTimeIntervalMs * maxHeartBeatLeak;
} else if (!memberState.isQuorum(validNum.get())) {
// 有效响应的数量无法满足 quorum，则返回 WAIT_TO_REVOTE
parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
nextTimeToRequestVote = getNextTimeToRequestVote();
} else if (!memberState.isQuorum(validNum.get() - biggerLedgerNum.get())) {
// 有效响应的数量减去日志条目大于自身的数量无法满足 quorum，则返回 WAIT_TO_REVOTE
parseResult = VoteResponse.ParseResult.WAIT_TO_REVOTE;
nextTimeToRequestVote = getNextTimeToRequestVote() + maxVoteIntervalMs;
} else if (memberState.isQuorum(acceptedNum.get())) {
// 接受的投票数量满足 quorum，则本次投票通过
parseResult = VoteResponse.ParseResult.PASSED;
} else if (memberState.isQuorum(acceptedNum.get() + notReadyTermNum.get())) {
// 已接受和未准备好的数量之和满足 quorum，则立即进行投票
parseResult = VoteResponse.ParseResult.REVOTE_IMMEDIATELY;
} else {
parseResult = VoteResponse.ParseResult.WAIT_TO_VOTE_NEXT;
nextTimeToRequestVote = getNextTimeToRequestVote();
}
lastParseResult = parseResult;
LOGGER.info("[{}] [PARSE_VOTE_RESULT] cost={} term={} memberNum={} allNum={} acceptedNum={} notReadyTermNum={} biggerLedgerNum={} alreadyHasLeader={} maxTerm={} result={}",
memberState.getSelfId(), lastVoteCost, term, memberState.peerSize(), allNum, acceptedNum, notReadyTermNum, biggerLedgerNum, alreadyHasLeader, knownMaxTermInGroup.get(), parseResult);
if (parseResult == VoteResponse.ParseResult.PASSED) {
LOGGER.info("[{}] [VOTE_RESULT] has been elected to be the leader in term {}", memberState.getSelfId(), term);
// 如果是通过，则转变为Leader对象
changeRoleToLeader(term);
}
}
}
由于篇幅的问题，我们并没有一一将 Dledger 的核心代码在这里展现，在此仅展示了选主等基本流程，对于写入、复制、日志存储、消息传递等都不多加诉说。Dledger 算法的核心原理是 Raft 协议，当一个节点发起投票请求时，其他节点会收到请求并发送响应，响应的结果将根据投票数量判断是否达成共识。如果共识达成，则新的 leader 将被选举出来，同时新的日志将被追加到磁盘上。如果共识未达成，则需要等待一定时间后重新发起投票请求。
DLedger 模式的弊端
那么这写法有问题吗？看起来似乎非常完美，解决了选主的问题，但是同时给用户造成了很大的困扰，首先，Broker 的副本必须是三个及以上，副本的 ACK 必须遵循多数派协议，这一点造成了成本与性能损耗的上升，其次，这使得 RocketMQ 存在两套 HA 复制流程，且 Raft 模式下的复制无法利用 RocketMQ 原生的存储能力。
RocketMQ 5.0
于是 RocketMQ 在 5.0 版本出了一个全新的模式，Controller 模式。一个基于 Raft 的一致性模块（DLedger Controller），并当作一个可选的选主组件，支持独立部署，也可以嵌入在 Nameserver 中，Broker 通过与 Controller 的交互完成 Master 的选举。
- 内嵌在 Nameserver 中
- 独立部署
DLedger Controller模式的优劣
然而，即使是 5.0 中做了优化的 DLedger Controller，仍然存在一些问题，下面让我们看看这个模式带来的优缺点
- 内嵌在 NameServer 中
- 资源竞争：如果 DLedger Controller 占用了过多的资源，可能会影响 NameServer 的其他功能。
- 稳定性问题：DLedger Controller 可能对 NameServer 的稳定性产生影响，例如当 DLedger Controller 出现故障或卡顿时，可能会影响整个 NameServer 运行的稳定性。
- 难以分离：将 DLedger Controller 内嵌在 NameServer 中，可能会降低其可分离性，增加其耦合度。
- 简化了部署：将 DLedger Controller 内嵌在 NameServer 中，可以减少需要部署的进程数量，降低了部署和维护的复杂度。
- 减少网络通信：内嵌在 NameServer 中的 DLedger Controller 可以直接与 NameServer 进行通信，减少了网络通信开销，提高了性能和可靠性。
- 整合了多个功能：内嵌在 NameServer 中的 DLedger Controller 可以整合 NameServer 的多个功能，如目录服务、路由服务、定时任务等，提高了系统集成的效率和灵活性。
- 优点
- 缺点
- 独立部署
- 复杂度高：DLedger Controller 的实现需要复杂的算法和数据结构，对开发人员的要求较高。
- 成本高：为了保证高可用性和稳定性，需要配置较多的物理服务器和网络设备，成本较高。
- 维护难度大：DLedger Controller 需要不断地进行监控和维护，一旦出现问题，可能需要进行复杂的排错和修复，增加了维护难度。
- 高可用：DLedger Controller 可提供高可用的服务，即使其中一台服务器出现问题，仍能保证集群的正常运行。
- 性能优秀：DLedger Controller 能快速处理大量的日志数据，使得 RocketMQ 能够高效地处理分布式事务。
- 扩展性好：DLedger Controller 可以添加新的节点，以适应不断增长的数据量和用户需求。
- 优点
- 缺点
总结
本文简单介绍了 DLedger 基于 Raft 协议实现的 Leader 选举机制，让大家深入地理解分布式系统中的 Leader 选举过程。然而如何在实际场景下选取、优化和扩展分布式一致性算法，都是非常值得探讨的问题。
参考文献
- Apache RocketMQ
- In Search of an Understandable Consensus Algorithm (Extended Version)
本文参与 腾讯云自媒体同步曝光计划，分享自微信公众号。
原始发表：2023-07-10，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
对象存储
对象存储（Cloud Object Storage，COS）是由腾讯云推出的无目录层次结构、无数据格式限制，可容纳海量数据且支持 HTTP/HTTPS 协议访问的分布式存储服务。腾讯云 COS 的存储桶空间无容量上限，无需分区管理，适用于 CDN 数据分发、数据万象处理或大数据计算与分析的数据湖等多种场景。
腾讯云开发者
