分布式系统基础：Raft算法深度解析与实现

1
2
3
4
5
6
7
8
9

┌─────────────────────────────────────┐
│           Raft设计目标              │
├─────────────────────────────────────┤
│ • 易于理解：简化复杂的共识问题       │
│ • 易于实现：提供清晰的实现指导       │
│ • 高可用性：容忍少数节点故障         │
│ • 强一致性：保证数据一致性           │
│ • 高性能：优化日志复制和选举过程     │
└─────────────────────────────────────┘

1
2
3
4
5

A[Raft算法] → B[节点状态]
A → C[核心机制]
A → D[安全性保证]
B → E[Leader 领导者]
B → F[Follower 跟随者]

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Raft算法实现
 */
public class RaftAlgorithm {

    /**
     * 节点状态枚举
     */
    public enum NodeState {
        FOLLOWER("跟随者"),
        CANDIDATE("候选者"),
        LEADER("领导者");

        private final String description;

        NodeState(String description) {
            this.description = description;
        }

        public String getDescription() { return description; }
    }

    /**
     * 日志条目
     */
    public static class LogEntry {
        private final int index;
        private final long term;
        private final String command;
        private final long timestamp;

        public LogEntry(int index, long term, String command) {
            this.index = index;
            this.term = term;
            this.command = command;
            this.timestamp = System.currentTimeMillis();
        }

        public int getIndex() { return index; }
        public long getTerm() { return term; }
        public String getCommand() { return command; }
        public long getTimestamp() { return timestamp; }

        @Override
        public boolean equals(Object obj) {
            if (this == obj) return true;
            if (!(obj instanceof LogEntry)) return false;
            LogEntry other = (LogEntry) obj;
            return index == other.index && term == other.term;
        }

        @Override
        public int hashCode() {
            return Objects.hash(index, term);
        }

        @Override
        public String toString() {
            return String.format("LogEntry{index=%d, term=%d, cmd='%s'}", index, term, command);
        }
    }

    /**
     * 投票请求
     */
    public static class VoteRequest {
        private final long term;
        private final String candidateId;
        private final int lastLogIndex;
        private final long lastLogTerm;

        public VoteRequest(long term, String candidateId, int lastLogIndex, long lastLogTerm) {
            this.term = term;
            this.candidateId = candidateId;
            this.lastLogIndex = lastLogIndex;
            this.lastLogTerm = lastLogTerm;
        }

        public long getTerm() { return term; }
        public String getCandidateId() { return candidateId; }
        public int getLastLogIndex() { return lastLogIndex; }
        public long getLastLogTerm() { return lastLogTerm; }
    }

    /**
     * 投票响应
     */
    public static class VoteResponse {
        private final long term;
        private final boolean voteGranted;
        private final String voterId;

        public VoteResponse(long term, boolean voteGranted, String voterId) {
            this.term = term;
            this.voteGranted = voteGranted;
            this.voterId = voterId;
        }

        public long getTerm() { return term; }
        public boolean isVoteGranted() { return voteGranted; }
        public String getVoterId() { return voterId; }
    }

    /**
     * 追加条目请求
     */
    public static class AppendEntriesRequest {
        private final long term;
        private final String leaderId;
        private final int prevLogIndex;
        private final long prevLogTerm;
        private final List<LogEntry> entries;
        private final int leaderCommit;

        public AppendEntriesRequest(long term, String leaderId, int prevLogIndex,
                                  long prevLogTerm, List<LogEntry> entries, int leaderCommit) {
            this.term = term;
            this.leaderId = leaderId;
            this.prevLogIndex = prevLogIndex;
            this.prevLogTerm = prevLogTerm;
            this.entries = entries != null ? new ArrayList<>(entries) : new ArrayList<>();
            this.leaderCommit = leaderCommit;
        }

        public long getTerm() { return term; }
        public String getLeaderId() { return leaderId; }
        public int getPrevLogIndex() { return prevLogIndex; }
        public long getPrevLogTerm() { return prevLogTerm; }
        public List<LogEntry> getEntries() { return entries; }
        public int getLeaderCommit() { return leaderCommit; }
        public boolean isHeartbeat() { return entries.isEmpty(); }
    }

    /**
     * 追加条目响应
     */
    public static class AppendEntriesResponse {
        private final long term;
        private final boolean success;
        private final String followerId;
        private final int matchIndex;

        public AppendEntriesResponse(long term, boolean success, String followerId, int matchIndex) {
            this.term = term;
            this.success = success;
            this.followerId = followerId;
            this.matchIndex = matchIndex;
        }

        public long getTerm() { return term; }
        public boolean isSuccess() { return success; }
        public String getFollowerId() { return followerId; }
        public int getMatchIndex() { return matchIndex; }
    }

    /**
     * Raft节点实现
     */
    public static class RaftNode {
        // 持久化状态
        private volatile long currentTerm = 0;
        private volatile String votedFor = null;
        private final List<LogEntry> log = Collections.synchronizedList(new ArrayList<>());

        // 易失状态
        private volatile int commitIndex = 0;
        private volatile int lastApplied = 0;

        // Leader状态
        private final Map<String, Integer> nextIndex = new ConcurrentHashMap<>();
        private final Map<String, Integer> matchIndex = new ConcurrentHashMap<>();

        // 节点信息
        private final String nodeId;
        private volatile NodeState state = NodeState.FOLLOWER;
        private volatile String currentLeader = null;
        private volatile long lastHeartbeat = System.currentTimeMillis();

        // 网络和定时器
        private final List<RaftNode> peers;
        private final ScheduledExecutorService scheduler;
        private final ExecutorService executor;
        private ScheduledFuture<?> electionTimer;
        private ScheduledFuture<?> heartbeatTimer;

        // 配置参数
        private final int electionTimeoutMin = 150; // ms
        private final int electionTimeoutMax = 300; // ms
        private final int heartbeatInterval = 50; // ms

        public RaftNode(String nodeId, List<RaftNode> peers) {
            this.nodeId = nodeId;
            this.peers = new ArrayList<>(peers);
            this.scheduler = Executors.newScheduledThreadPool(2);
            this.executor = Executors.newCachedThreadPool();

            // 初始化日志（索引0为空条目）
            this.log.add(new LogEntry(0, 0, ""));

            // 启动选举定时器
            resetElectionTimer();
        }

        /**
         * 启动Leader选举
         */
        private void startElection() {
            if (state == NodeState.LEADER) {
                return; // Leader不参与选举
            }

            synchronized (this) {
                state = NodeState.CANDIDATE;
                currentTerm++;
                votedFor = nodeId;
                resetElectionTimer();

                System.out.printf("节点 %s 开始选举，任期 %d%n", nodeId, currentTerm);
            }

            // 向所有节点发送投票请求
            List<CompletableFuture<VoteResponse>> voteFutures = new ArrayList<>();

            for (RaftNode peer : peers) {
                if (!peer.nodeId.equals(nodeId)) {
                    CompletableFuture<VoteResponse> future = CompletableFuture
                        .supplyAsync(() -> peer.requestVote(createVoteRequest()), executor);
                    voteFutures.add(future);
                }
            }

            // 自己给自己投票
            AtomicInteger voteCount = new AtomicInteger(1);
            int majoritySize = (peers.size() + 1) / 2 + 1;

            // 处理投票响应
            CompletableFuture.allOf(voteFutures.toArray(new CompletableFuture[0]))
                .thenRun(() -> {
                    if (state != NodeState.CANDIDATE) {
                        return; // 状态已改变
                    }

                    for (CompletableFuture<VoteResponse> future : voteFutures) {
                        try {
                            VoteResponse response = future.get();
                            if (response.getTerm() > currentTerm) {
                                becomeFollower(response.getTerm());
                                return;
                            }
                            if (response.isVoteGranted()) {
                                voteCount.incrementAndGet();
                            }
                        } catch (Exception e) {
                            // 忽略投票失败
                        }
                    }

                    if (voteCount.get() >= majoritySize && state == NodeState.CANDIDATE) {
                        becomeLeader();
                    }
                });
        }

        /**
         * 创建投票请求
         */
        private VoteRequest createVoteRequest() {
            int lastLogIndex = log.size() - 1;
            long lastLogTerm = lastLogIndex > 0 ? log.get(lastLogIndex).getTerm() : 0;
            return new VoteRequest(currentTerm, nodeId, lastLogIndex, lastLogTerm);
        }

        /**
         * 处理投票请求
         */
        public VoteResponse requestVote(VoteRequest request) {
            synchronized (this) {
                // 检查任期
                if (request.getTerm() > currentTerm) {
                    becomeFollower(request.getTerm());
                }

                boolean voteGranted = false;

                // 投票条件检查
                if (request.getTerm() == currentTerm &&
                    (votedFor == null || votedFor.equals(request.getCandidateId())) &&
                    isLogUpToDate(request.getLastLogIndex(), request.getLastLogTerm())) {

                    votedFor = request.getCandidateId();
                    voteGranted = true;
                    resetElectionTimer();

                    System.out.printf("节点 %s 投票给 %s，任期 %d%n",
                                     nodeId, request.getCandidateId(), currentTerm);
                }

                return new VoteResponse(currentTerm, voteGranted, nodeId);
            }
        }

        /**
         * 检查日志是否足够新
         */
        private boolean isLogUpToDate(int lastLogIndex, long lastLogTerm) {
            int myLastLogIndex = log.size() - 1;
            long myLastLogTerm = myLastLogIndex > 0 ? log.get(myLastLogIndex).getTerm() : 0;

            // 比较最后一条日志的任期和索引
            if (lastLogTerm != myLastLogTerm) {
                return lastLogTerm > myLastLogTerm;
            }
            return lastLogIndex >= myLastLogIndex;
        }

        /**
         * 成为Leader
         */
        private void becomeLeader() {
            if (state != NodeState.CANDIDATE) {
                return;
            }

            synchronized (this) {
                state = NodeState.LEADER;
                currentLeader = nodeId;

                // 初始化Leader状态
                int lastLogIndex = log.size() - 1;
                for (RaftNode peer : peers) {
                    if (!peer.nodeId.equals(nodeId)) {
                        nextIndex.put(peer.nodeId, lastLogIndex + 1);
                        matchIndex.put(peer.nodeId, 0);
                    }
                }

                System.out.printf("节点 %s 成为Leader，任期 %d%n", nodeId, currentTerm);
            }

            stopElectionTimer();
            startHeartbeat();

            // 发送空的AppendEntries作为心跳
            sendHeartbeat();
        }

        /**
         * 成为Follower
         */
        private void becomeFollower(long term) {
            synchronized (this) {
                state = NodeState.FOLLOWER;
                currentTerm = term;
                votedFor = null;
                currentLeader = null;
            }

            stopHeartbeatTimer();
            resetElectionTimer();
        }

        /**
         * 发送心跳
         */
        private void sendHeartbeat() {
            if (state != NodeState.LEADER) {
                return;
            }

            for (RaftNode peer : peers) {
                if (!peer.nodeId.equals(nodeId)) {
                    executor.submit(() -> sendAppendEntries(peer, true));
                }
            }
        }

        /**
         * 发送AppendEntries RPC
         */
        private void sendAppendEntries(RaftNode peer, boolean isHeartbeat) {
            if (state != NodeState.LEADER) {
                return;
            }

            Integer peerNextIndex = nextIndex.get(peer.nodeId);
            if (peerNextIndex == null) {
                peerNextIndex = log.size();
                nextIndex.put(peer.nodeId, peerNextIndex);
            }

            int prevLogIndex = peerNextIndex - 1;
            long prevLogTerm = prevLogIndex > 0 ? log.get(prevLogIndex).getTerm() : 0;

            List<LogEntry> entries = new ArrayList<>();
            if (!isHeartbeat && peerNextIndex < log.size()) {
                // 发送需要复制的日志条目
                entries = log.subList(peerNextIndex, log.size());
            }

            AppendEntriesRequest request = new AppendEntriesRequest(
                currentTerm, nodeId, prevLogIndex, prevLogTerm, entries, commitIndex);

            try {
                AppendEntriesResponse response = peer.appendEntries(request);
                handleAppendEntriesResponse(peer, request, response);
            } catch (Exception e) {
                System.err.printf("发送AppendEntries到 %s 失败: %s%n", peer.nodeId, e.getMessage());
            }
        }

        /**
         * 处理AppendEntries响应
         */
        private void handleAppendEntriesResponse(RaftNode peer, AppendEntriesRequest request,
                                               AppendEntriesResponse response) {
            if (state != NodeState.LEADER || response.getTerm() > currentTerm) {
                if (response.getTerm() > currentTerm) {
                    becomeFollower(response.getTerm());
                }
                return;
            }

            if (response.isSuccess()) {
                // 更新nextIndex和matchIndex
                int newMatchIndex = request.getPrevLogIndex() + request.getEntries().size();
                matchIndex.put(peer.nodeId, newMatchIndex);
                nextIndex.put(peer.nodeId, newMatchIndex + 1);

                // 尝试更新commitIndex
                updateCommitIndex();
            } else {
                // 日志不匹配，递减nextIndex重试
                Integer currentNext = nextIndex.get(peer.nodeId);
                if (currentNext != null && currentNext > 1) {
                    nextIndex.put(peer.nodeId, currentNext - 1);
                    executor.submit(() -> sendAppendEntries(peer, false));
                }
            }
        }

        /**
         * 更新commitIndex
         */
        private void updateCommitIndex() {
            for (int n = commitIndex + 1; n < log.size(); n++) {
                if (log.get(n).getTerm() == currentTerm) {
                    int replicationCount = 1; // 包括Leader自己

                    for (Integer match : matchIndex.values()) {
                        if (match >= n) {
                            replicationCount++;
                        }
                    }

                    if (replicationCount > (peers.size() + 1) / 2) {
                        commitIndex = n;
                        System.out.printf("Leader %s 提交日志到索引 %d%n", nodeId, commitIndex);
                    }
                }
            }
        }

        /**
         * 处理AppendEntries请求
         */
        public AppendEntriesResponse appendEntries(AppendEntriesRequest request) {
            synchronized (this) {
                lastHeartbeat = System.currentTimeMillis();

                // 检查任期
                if (request.getTerm() > currentTerm) {
                    becomeFollower(request.getTerm());
                } else if (request.getTerm() < currentTerm) {
                    return new AppendEntriesResponse(currentTerm, false, nodeId, 0);
                }

                // 重置选举定时器
                if (state == NodeState.CANDIDATE) {
                    becomeFollower(currentTerm);
                }
                resetElectionTimer();

                currentLeader = request.getLeaderId();

                // 检查日志一致性
                if (request.getPrevLogIndex() > 0) {
                    if (request.getPrevLogIndex() >= log.size() ||
                        log.get(request.getPrevLogIndex()).getTerm() != request.getPrevLogTerm()) {
                        return new AppendEntriesResponse(currentTerm, false, nodeId, 0);
                    }
                }

                // 追加新条目
                if (!request.getEntries().isEmpty()) {
                    int insertIndex = request.getPrevLogIndex() + 1;

                    // 删除冲突的条目
                    if (insertIndex < log.size()) {
                        log.subList(insertIndex, log.size()).clear();
                    }

                    // 追加新条目
                    log.addAll(request.getEntries());

                    System.out.printf("Follower %s 追加 %d 条日志条目%n",
                                     nodeId, request.getEntries().size());
                }

                // 更新commitIndex
                if (request.getLeaderCommit() > commitIndex) {
                    commitIndex = Math.min(request.getLeaderCommit(), log.size() - 1);
                }

                return new AppendEntriesResponse(currentTerm, true, nodeId, log.size() - 1);
            }
        }

        /**
         * 客户端提交命令
         */
        public CompletableFuture<String> submitCommand(String command) {
            if (state != NodeState.LEADER) {
                return CompletableFuture.completedFuture("NOT_LEADER");
            }

            synchronized (this) {
                LogEntry entry = new LogEntry(log.size(), currentTerm, command);
                log.add(entry);

                System.out.printf("Leader %s 接收命令: %s (索引 %d)%n",
                                 nodeId, command, entry.getIndex());
            }

            // 立即复制到所有Follower
            for (RaftNode peer : peers) {
                if (!peer.nodeId.equals(nodeId)) {
                    executor.submit(() -> sendAppendEntries(peer, false));
                }
            }

            // 等待日志被提交
            return waitForCommit(log.size() - 1);
        }

        /**
         * 等待日志提交
         */
        private CompletableFuture<String> waitForCommit(int logIndex) {
            return CompletableFuture.supplyAsync(() -> {
                long startTime = System.currentTimeMillis();
                long timeout = 5000; // 5秒超时

                while (commitIndex < logIndex && System.currentTimeMillis() - startTime < timeout) {
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return "INTERRUPTED";
                    }
                }

                if (commitIndex >= logIndex) {
                    return "COMMITTED";
                } else {
                    return "TIMEOUT";
                }
            }, executor);
        }

        /**
         * 定时器管理
         */
        private void resetElectionTimer() {
            stopElectionTimer();
            long timeout = ThreadLocalRandom.current().nextLong(electionTimeoutMin, electionTimeoutMax);
            electionTimer = scheduler.schedule(this::startElection, timeout, TimeUnit.MILLISECONDS);
        }

        private void stopElectionTimer() {
            if (electionTimer != null) {
                electionTimer.cancel(false);
            }
        }

        private void startHeartbeat() {
            heartbeatTimer = scheduler.scheduleWithFixedDelay(
                this::sendHeartbeat, 0, heartbeatInterval, TimeUnit.MILLISECONDS);
        }

        private void stopHeartbeatTimer() {
            if (heartbeatTimer != null) {
                heartbeatTimer.cancel(false);
            }
        }

        // Getter方法
        public String getNodeId() { return nodeId; }
        public NodeState getState() { return state; }
        public long getCurrentTerm() { return currentTerm; }
        public String getCurrentLeader() { return currentLeader; }
        public int getLogSize() { return log.size(); }
        public int getCommitIndex() { return commitIndex; }

        public void shutdown() {
            stopElectionTimer();
            stopHeartbeatTimer();
            scheduler.shutdown();
            executor.shutdown();
        }
    }
}

/**
 * Raft集群演示
 */
class RaftClusterDemo {
    public static void main(String[] args) throws Exception {
        System.out.println("=== Raft算法演示 ===");

        // 创建5个节点的Raft集群
        List<RaftAlgorithm.RaftNode> nodes = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            nodes.add(new RaftAlgorithm.RaftNode("Node" + i, nodes));
        }

        // 更新每个节点的peers列表
        for (RaftAlgorithm.RaftNode node : nodes) {
            // 使用反射或其他方式更新peers（简化演示）
        }

        try {
            // 等待Leader选举
            System.out.println("等待Leader选举...");
            Thread.sleep(1000);

            // 查找Leader
            RaftAlgorithm.RaftNode leader = nodes.stream()
                .filter(node -> node.getState() == RaftAlgorithm.NodeState.LEADER)
                .findFirst()
                .orElse(null);

            if (leader != null) {
                System.out.printf("Leader选举完成: %s (任期 %d)%n",
                                 leader.getNodeId(), leader.getCurrentTerm());

                // 提交一些命令
                List<CompletableFuture<String>> futures = new ArrayList<>();
                for (int i = 1; i <= 5; i++) {
                    String command = "SET key" + i + " value" + i;
                    CompletableFuture<String> future = leader.submitCommand(command);
                    futures.add(future);
                }

                // 等待命令提交
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();

                System.out.println("\n--- 集群状态 ---");
                for (RaftAlgorithm.RaftNode node : nodes) {
                    System.out.printf("节点 %s: 状态=%s, 任期=%d, 日志大小=%d, 提交索引=%d%n",
                                     node.getNodeId(),
                                     node.getState().getDescription(),
                                     node.getCurrentTerm(),
                                     node.getLogSize(),
                                     node.getCommitIndex());
                }
            } else {
                System.out.println("Leader选举失败");
            }

        } finally {
            // 关闭所有节点
            for (RaftAlgorithm.RaftNode node : nodes) {
                node.shutdown();
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108

/**
 * 批量日志复制优化
 */
public class BatchLogReplication {

    /**
     * 批量日志条目
     */
    public static class BatchLogEntries {
        private final List<RaftAlgorithm.LogEntry> entries;
        private final int batchId;
        private final long timestamp;

        public BatchLogEntries(List<RaftAlgorithm.LogEntry> entries, int batchId) {
            this.entries = new ArrayList<>(entries);
            this.batchId = batchId;
            this.timestamp = System.currentTimeMillis();
        }

        public List<RaftAlgorithm.LogEntry> getEntries() { return entries; }
        public int getBatchId() { return batchId; }
        public long getTimestamp() { return timestamp; }
        public int size() { return entries.size(); }
    }

    /**
     * 批量复制管理器
     */
    public static class BatchReplicationManager {
        private final RaftAlgorithm.RaftNode node;
        private final List<RaftAlgorithm.LogEntry> pendingEntries;
        private final int batchSize;
        private final long batchTimeoutMs;
        private final ScheduledExecutorService scheduler;
        private ScheduledFuture<?> batchTimer;

        public BatchReplicationManager(RaftAlgorithm.RaftNode node, int batchSize, long batchTimeoutMs) {
            this.node = node;
            this.pendingEntries = Collections.synchronizedList(new ArrayList<>());
            this.batchSize = batchSize;
            this.batchTimeoutMs = batchTimeoutMs;
            this.scheduler = Executors.newScheduledThreadPool(1);
        }

        /**
         * 添加待复制的日志条目
         */
        public synchronized void addEntry(RaftAlgorithm.LogEntry entry) {
            pendingEntries.add(entry);

            // 检查是否达到批量大小
            if (pendingEntries.size() >= batchSize) {
                flushBatch();
            } else if (batchTimer == null) {
                // 启动批量定时器
                batchTimer = scheduler.schedule(this::flushBatch, batchTimeoutMs, TimeUnit.MILLISECONDS);
            }
        }

        /**
         * 刷新批量条目
         */
        private synchronized void flushBatch() {
            if (pendingEntries.isEmpty()) {
                return;
            }

            List<RaftAlgorithm.LogEntry> batch = new ArrayList<>(pendingEntries);
            pendingEntries.clear();

            if (batchTimer != null) {
                batchTimer.cancel(false);
                batchTimer = null;
            }

            // 执行批量复制
            executeBatchReplication(batch);
        }

        /**
         * 执行批量复制
         */
        private void executeBatchReplication(List<RaftAlgorithm.LogEntry> batch) {
            BatchLogEntries batchEntries = new BatchLogEntries(batch, generateBatchId());

            System.out.printf("Leader %s 开始批量复制: %d 条日志条目%n",
                             node.getNodeId(), batch.size());

            // 并行发送到所有Follower
            // 这里简化实现，实际中需要集成到RaftNode中
            CompletableFuture.runAsync(() -> {
                // 批量复制逻辑
                System.out.printf("批量复制完成: 批次ID %d%n", batchEntries.getBatchId());
            });
        }

        private int generateBatchId() {
            return (int) (System.currentTimeMillis() % Integer.MAX_VALUE);
        }

        public void shutdown() {
            if (batchTimer != null) {
                batchTimer.cancel(false);
            }
            scheduler.shutdown();
        }
    }
}