分布式系统理论与实践 - Hashicorp Raft：生产级Raft实现详解

1
2
3
4
5

A[状态机] → B[快照触发器]
B → C[增量快照]
B → D[全量快照]
C → E[快照合并]
D → F[快照存储]

1
2
3
4

A[批量处理] → B[管道化]
B → C[压缩传输]
C → D[连接复用]
D → E[异步IO]

  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

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.time.Instant;
import java.nio.ByteBuffer;
import java.util.logging.Logger;

public class HashicorpRaft {
    private static final Logger logger = Logger.getLogger(HashicorpRaft.class.getName());

    // 节点状态
    public enum NodeState {
        FOLLOWER, CANDIDATE, LEADER, SHUTDOWN
    }

    // 配置参数
    public static class Config {
        public final Duration heartbeatTimeout = Duration.ofMillis(150);
        public final Duration electionTimeout = Duration.ofMillis(300);
        public final Duration leaderLeaseTimeout = Duration.ofMillis(500);
        public final Duration commitTimeout = Duration.ofSeconds(5);
        public final int snapshotThreshold = 1024;
        public final int maxAppendEntries = 64;
        public final boolean enablePreVote = true;
        public final boolean enableLearner = true;
    }

    // 日志条目
    public static class LogEntry {
        public final long term;
        public final long index;
        public final byte[] data;
        public final EntryType type;
        public final Instant timestamp;

        public enum EntryType {
            NORMAL, CONFIGURATION, NOOP, BARRIER
        }

        public LogEntry(long term, long index, byte[] data, EntryType type) {
            this.term = term;
            this.index = index;
            this.data = data != null ? data.clone() : new byte[0];
            this.type = type;
            this.timestamp = Instant.now();
        }
    }

    // 快照元数据
    public static class SnapshotMeta {
        public final String id;
        public final long index;
        public final long term;
        public final Set<String> peers;
        public final long size;
        public final Instant timestamp;

        public SnapshotMeta(String id, long index, long term, Set<String> peers, long size) {
            this.id = id;
            this.index = index;
            this.term = term;
            this.peers = new HashSet<>(peers);
            this.size = size;
            this.timestamp = Instant.now();
        }
    }

    private final String nodeId;
    private final Config config;
    private final RaftStorage storage;
    private final Transport transport;
    private final StateMachine stateMachine;
    private final ScheduledExecutorService scheduler;

    // Raft状态
    private volatile NodeState state = NodeState.FOLLOWER;
    private volatile long currentTerm = 0;
    private volatile String votedFor = null;
    private volatile String leaderId = null;
    private volatile long commitIndex = 0;
    private volatile long lastApplied = 0;

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

    // 性能统计
    private final AtomicLong totalRequests = new AtomicLong();
    private final AtomicLong successfulRequests = new AtomicLong();
    private final AtomicLong leadershipChanges = new AtomicLong();

    public HashicorpRaft(String nodeId, Config config, RaftStorage storage,
                         Transport transport, StateMachine stateMachine) {
        this.nodeId = nodeId;
        this.config = config;
        this.storage = storage;
        this.transport = transport;
        this.stateMachine = stateMachine;
        this.scheduler = Executors.newScheduledThreadPool(4);

        initializeFromStorage();
        startPeriodicTasks();
    }

    private void initializeFromStorage() {
        try {
            this.currentTerm = storage.getCurrentTerm();
            this.votedFor = storage.getVotedFor();
            this.commitIndex = storage.getCommitIndex();
            this.lastApplied = storage.getLastApplied();

            // 恢复配置
            LogEntry lastConfig = storage.getLastConfigurationEntry();
            if (lastConfig != null) {
                updateConfiguration(parseConfiguration(lastConfig.data));
            }

        } catch (Exception e) {
            logger.severe("Failed to initialize from storage: " + e.getMessage());
            throw new RuntimeException(e);
        }
    }

    // 启动周期性任务
    private void startPeriodicTasks() {
        // 选举超时检查
        scheduler.scheduleWithFixedDelay(this::checkElectionTimeout,
            50, 50, TimeUnit.MILLISECONDS);

        // 心跳发送
        scheduler.scheduleWithFixedDelay(this::sendHeartbeats,
            config.heartbeatTimeout.toMillis() / 3,
            config.heartbeatTimeout.toMillis() / 3, TimeUnit.MILLISECONDS);

        // 日志压缩
        scheduler.scheduleWithFixedDelay(this::compactLogs,
            5, 5, TimeUnit.SECONDS);

        // 性能统计
        scheduler.scheduleWithFixedDelay(this::logPerformanceMetrics,
            30, 30, TimeUnit.SECONDS);
    }
}

  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

public class AppendEntriesHandler {
    private final HashicorpRaft raft;
    private final BatchProcessor batchProcessor;

    // 批量处理器
    public static class BatchProcessor {
        private final Queue<AppendEntriesRequest> pendingRequests = new ConcurrentLinkedQueue<>();
        private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
        private volatile boolean processingBatch = false;

        public void addRequest(AppendEntriesRequest request) {
            pendingRequests.offer(request);
            if (!processingBatch) {
                scheduleProcessing();
            }
        }

        private void scheduleProcessing() {
            processingBatch = true;
            executor.schedule(() -> {
                processBatch();
                processingBatch = false;
                if (!pendingRequests.isEmpty()) {
                    scheduleProcessing();
                }
            }, 1, TimeUnit.MILLISECONDS);
        }

        private void processBatch() {
            List<AppendEntriesRequest> batch = new ArrayList<>();
            AppendEntriesRequest request;
            while ((request = pendingRequests.poll()) != null && batch.size() < 64) {
                batch.add(request);
            }

            if (!batch.isEmpty()) {
                processBatchedRequests(batch);
            }
        }

        private void processBatchedRequests(List<AppendEntriesRequest> requests) {
            // 批量处理请求以提高性能
            for (AppendEntriesRequest request : requests) {
                processAppendEntries(request);
            }
        }
    }

    // AppendEntries请求
    public static class AppendEntriesRequest {
        public final long term;
        public final String leaderId;
        public final long prevLogIndex;
        public final long prevLogTerm;
        public final List<LogEntry> entries;
        public final long leaderCommit;
        public final CompletableFuture<AppendEntriesResponse> future;

        public AppendEntriesRequest(long term, String leaderId, long prevLogIndex,
                                  long prevLogTerm, List<LogEntry> entries,
                                  long leaderCommit) {
            this.term = term;
            this.leaderId = leaderId;
            this.prevLogIndex = prevLogIndex;
            this.prevLogTerm = prevLogTerm;
            this.entries = new ArrayList<>(entries);
            this.leaderCommit = leaderCommit;
            this.future = new CompletableFuture<>();
        }
    }

    // AppendEntries响应
    public static class AppendEntriesResponse {
        public final long term;
        public final boolean success;
        public final long lastLogIndex;
        public final long conflictIndex;
        public final long conflictTerm;

        public AppendEntriesResponse(long term, boolean success, long lastLogIndex,
                                   long conflictIndex, long conflictTerm) {
            this.term = term;
            this.success = success;
            this.lastLogIndex = lastLogIndex;
            this.conflictIndex = conflictIndex;
            this.conflictTerm = conflictTerm;
        }
    }

    public AppendEntriesHandler(HashicorpRaft raft) {
        this.raft = raft;
        this.batchProcessor = new BatchProcessor();
    }

    public CompletableFuture<AppendEntriesResponse> handleAppendEntries(AppendEntriesRequest request) {
        batchProcessor.addRequest(request);
        return request.future;
    }

    private void processAppendEntries(AppendEntriesRequest request) {
        try {
            AppendEntriesResponse response = doAppendEntries(request);
            request.future.complete(response);
        } catch (Exception e) {
            request.future.completeExceptionally(e);
        }
    }

    private AppendEntriesResponse doAppendEntries(AppendEntriesRequest request) {
        // 1. 检查term
        if (request.term < raft.getCurrentTerm()) {
            return new AppendEntriesResponse(raft.getCurrentTerm(), false,
                raft.getLastLogIndex(), 0, 0);
        }

        // 2. 更新term和leader
        if (request.term > raft.getCurrentTerm()) {
            raft.setCurrentTerm(request.term);
            raft.setVotedFor(null);
            raft.becomeFollower();
        }

        raft.setLeaderId(request.leaderId);
        raft.resetElectionTimeout();

        // 3. 检查日志一致性
        if (request.prevLogIndex > 0) {
            LogEntry prevEntry = raft.getStorage().getLogEntry(request.prevLogIndex);
            if (prevEntry == null || prevEntry.term != request.prevLogTerm) {
                // 查找冲突点
                long conflictIndex = Math.min(request.prevLogIndex, raft.getLastLogIndex());
                long conflictTerm = 0;
                if (conflictIndex > 0) {
                    LogEntry conflictEntry = raft.getStorage().getLogEntry(conflictIndex);
                    if (conflictEntry != null) {
                        conflictTerm = conflictEntry.term;
                        // 查找该term的第一个条目
                        while (conflictIndex > 1) {
                            LogEntry entry = raft.getStorage().getLogEntry(conflictIndex - 1);
                            if (entry == null || entry.term != conflictTerm) {
                                break;
                            }
                            conflictIndex--;
                        }
                    }
                }

                return new AppendEntriesResponse(raft.getCurrentTerm(), false,
                    raft.getLastLogIndex(), conflictIndex, conflictTerm);
            }
        }

        // 4. 删除冲突的条目并追加新条目
        if (!request.entries.isEmpty()) {
            long appendStartIndex = request.prevLogIndex + 1;
            raft.getStorage().deleteLogEntriesFrom(appendStartIndex);

            for (LogEntry entry : request.entries) {
                raft.getStorage().appendLogEntry(entry);
            }
        }

        // 5. 更新commit index
        if (request.leaderCommit > raft.getCommitIndex()) {
            long newCommitIndex = Math.min(request.leaderCommit, raft.getLastLogIndex());
            raft.setCommitIndex(newCommitIndex);
            raft.applyLogEntries();
        }

        return new AppendEntriesResponse(raft.getCurrentTerm(), true,
            raft.getLastLogIndex(), 0, 0);
    }
}

  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

public class SnapshotManager {
    private final HashicorpRaft raft;
    private final SnapshotStorage snapshotStorage;
    private final ScheduledExecutorService executor;
    private volatile boolean snapshotInProgress = false;

    // 快照存储接口
    public interface SnapshotStorage {
        void saveSnapshot(SnapshotMeta meta, byte[] data) throws Exception;
        byte[] loadSnapshot(String snapshotId) throws Exception;
        List<SnapshotMeta> listSnapshots() throws Exception;
        void deleteSnapshot(String snapshotId) throws Exception;
    }

    // 快照任务
    public static class SnapshotTask {
        public final CompletableFuture<SnapshotMeta> future = new CompletableFuture<>();
        public final long index;
        public final long term;
        public final boolean forced;

        public SnapshotTask(long index, long term, boolean forced) {
            this.index = index;
            this.term = term;
            this.forced = forced;
        }
    }

    // 增量快照信息
    public static class IncrementalSnapshot {
        public final SnapshotMeta baseMeta;
        public final List<LogEntry> incrementalEntries;
        public final long newIndex;
        public final long newTerm;

        public IncrementalSnapshot(SnapshotMeta baseMeta, List<LogEntry> incrementalEntries,
                                 long newIndex, long newTerm) {
            this.baseMeta = baseMeta;
            this.incrementalEntries = new ArrayList<>(incrementalEntries);
            this.newIndex = newIndex;
            this.newTerm = newTerm;
        }
    }

    public SnapshotManager(HashicorpRaft raft, SnapshotStorage snapshotStorage) {
        this.raft = raft;
        this.snapshotStorage = snapshotStorage;
        this.executor = Executors.newSingleThreadScheduledExecutor();
    }

    // 检查是否需要创建快照
    public boolean shouldCreateSnapshot() {
        long lastSnapshotIndex = raft.getLastSnapshotIndex();
        long currentIndex = raft.getLastLogIndex();

        return (currentIndex - lastSnapshotIndex) >= raft.getConfig().snapshotThreshold;
    }

    // 异步创建快照
    public CompletableFuture<SnapshotMeta> createSnapshotAsync() {
        if (snapshotInProgress) {
            return CompletableFuture.failedFuture(
                new IllegalStateException("Snapshot already in progress"));
        }

        SnapshotTask task = new SnapshotTask(raft.getLastApplied(),
            raft.getCurrentTerm(), false);

        executor.submit(() -> performSnapshot(task));
        return task.future;
    }

    // 强制创建快照
    public CompletableFuture<SnapshotMeta> forceSnapshot() {
        SnapshotTask task = new SnapshotTask(raft.getLastApplied(),
            raft.getCurrentTerm(), true);

        executor.submit(() -> performSnapshot(task));
        return task.future;
    }

    private void performSnapshot(SnapshotTask task) {
        try {
            snapshotInProgress = true;

            // 1. 检查是否可以创建增量快照
            SnapshotMeta incrementalBase = findIncrementalBase(task.index);
            SnapshotMeta newSnapshot;

            if (incrementalBase != null && !task.forced) {
                newSnapshot = createIncrementalSnapshot(incrementalBase, task);
            } else {
                newSnapshot = createFullSnapshot(task);
            }

            // 2. 清理旧快照
            cleanupOldSnapshots();

            // 3. 压缩日志
            if (newSnapshot != null) {
                raft.getStorage().compactLogs(newSnapshot.index);
            }

            task.future.complete(newSnapshot);

        } catch (Exception e) {
            task.future.completeExceptionally(e);
        } finally {
            snapshotInProgress = false;
        }
    }

    // 查找增量快照基础
    private SnapshotMeta findIncrementalBase(long targetIndex) {
        try {
            List<SnapshotMeta> snapshots = snapshotStorage.listSnapshots();
            return snapshots.stream()
                .filter(s -> s.index <= targetIndex)
                .max(Comparator.comparing(s -> s.index))
                .orElse(null);
        } catch (Exception e) {
            logger.warning("Failed to find incremental base: " + e.getMessage());
            return null;
        }
    }

    // 创建增量快照
    private SnapshotMeta createIncrementalSnapshot(SnapshotMeta base, SnapshotTask task) {
        try {
            // 获取增量数据
            List<LogEntry> incrementalEntries = raft.getStorage()
                .getLogEntries(base.index + 1, task.index);

            if (incrementalEntries.size() > raft.getConfig().snapshotThreshold / 2) {
                // 增量太大，创建全量快照
                return createFullSnapshot(task);
            }

            // 序列化增量数据
            IncrementalSnapshot incremental = new IncrementalSnapshot(
                base, incrementalEntries, task.index, task.term);

            byte[] data = serializeIncrementalSnapshot(incremental);

            String snapshotId = generateSnapshotId(task.index, task.term, true);
            SnapshotMeta meta = new SnapshotMeta(snapshotId, task.index, task.term,
                raft.getCurrentPeers(), data.length);

            snapshotStorage.saveSnapshot(meta, data);

            logger.info(String.format("Created incremental snapshot %s (base: %s, entries: %d)",
                snapshotId, base.id, incrementalEntries.size()));

            return meta;

        } catch (Exception e) {
            logger.warning("Failed to create incremental snapshot: " + e.getMessage());
            return createFullSnapshot(task);
        }
    }

    // 创建全量快照
    private SnapshotMeta createFullSnapshot(SnapshotTask task) {
        try {
            // 获取状态机快照
            byte[] stateMachineData = raft.getStateMachine().createSnapshot();

            // 获取配置信息
            Set<String> peers = raft.getCurrentPeers();

            // 创建快照数据
            FullSnapshot fullSnapshot = new FullSnapshot(stateMachineData, peers,
                task.index, task.term);

            byte[] data = serializeFullSnapshot(fullSnapshot);

            String snapshotId = generateSnapshotId(task.index, task.term, false);
            SnapshotMeta meta = new SnapshotMeta(snapshotId, task.index, task.term,
                peers, data.length);

            snapshotStorage.saveSnapshot(meta, data);

            logger.info(String.format("Created full snapshot %s (size: %d bytes)",
                snapshotId, data.length));

            return meta;

        } catch (Exception e) {
            logger.severe("Failed to create full snapshot: " + e.getMessage());
            throw new RuntimeException(e);
        }
    }

    // 清理旧快照
    private void cleanupOldSnapshots() {
        try {
            List<SnapshotMeta> snapshots = snapshotStorage.listSnapshots();

            // 保留最新的3个快照
            snapshots.stream()
                .sorted(Comparator.comparing((SnapshotMeta s) -> s.index).reversed())
                .skip(3)
                .forEach(snapshot -> {
                    try {
                        snapshotStorage.deleteSnapshot(snapshot.id);
                        logger.info("Deleted old snapshot: " + snapshot.id);
                    } catch (Exception e) {
                        logger.warning("Failed to delete snapshot " + snapshot.id + ": " + e.getMessage());
                    }
                });

        } catch (Exception e) {
            logger.warning("Failed to cleanup old snapshots: " + e.getMessage());
        }
    }

    // 恢复快照
    public void restoreSnapshot(SnapshotMeta meta) {
        try {
            byte[] data = snapshotStorage.loadSnapshot(meta.id);

            if (meta.id.contains("incremental")) {
                restoreIncrementalSnapshot(data, meta);
            } else {
                restoreFullSnapshot(data, meta);
            }

            raft.setLastSnapshotIndex(meta.index);
            raft.setLastSnapshotTerm(meta.term);

            logger.info("Successfully restored snapshot: " + meta.id);

        } catch (Exception e) {
            logger.severe("Failed to restore snapshot " + meta.id + ": " + e.getMessage());
            throw new RuntimeException(e);
        }
    }

    private void restoreIncrementalSnapshot(byte[] data, SnapshotMeta meta) throws Exception {
        IncrementalSnapshot incremental = deserializeIncrementalSnapshot(data);

        // 1. 恢复基础快照
        restoreSnapshot(incremental.baseMeta);

        // 2. 应用增量数据
        for (LogEntry entry : incremental.incrementalEntries) {
            raft.getStateMachine().apply(entry);
        }
    }

    private void restoreFullSnapshot(byte[] data, SnapshotMeta meta) throws Exception {
        FullSnapshot fullSnapshot = deserializeFullSnapshot(data);

        // 恢复状态机
        raft.getStateMachine().restoreSnapshot(fullSnapshot.stateMachineData);

        // 更新配置
        raft.updateConfiguration(fullSnapshot.peers);
    }

    // 序列化和反序列化方法
    private byte[] serializeIncrementalSnapshot(IncrementalSnapshot snapshot) {
        // 实现序列化逻辑
        ByteBuffer buffer = ByteBuffer.allocate(8192);
        // ... 序列化实现
        return buffer.array();
    }

    private IncrementalSnapshot deserializeIncrementalSnapshot(byte[] data) {
        // 实现反序列化逻辑
        ByteBuffer buffer = ByteBuffer.wrap(data);
        // ... 反序列化实现
        return null; // 示例返回
    }

    private byte[] serializeFullSnapshot(FullSnapshot snapshot) {
        // 实现序列化逻辑
        return new byte[0]; // 示例返回
    }

    private FullSnapshot deserializeFullSnapshot(byte[] data) {
        // 实现反序列化逻辑
        return null; // 示例返回
    }

    private String generateSnapshotId(long index, long term, boolean incremental) {
        String type = incremental ? "incremental" : "full";
        return String.format("%s_%d_%d_%d", type, index, term, System.currentTimeMillis());
    }

    // 全量快照数据结构
    private static class FullSnapshot {
        public final byte[] stateMachineData;
        public final Set<String> peers;
        public final long index;
        public final long term;

        public FullSnapshot(byte[] stateMachineData, Set<String> peers, long index, long term) {
            this.stateMachineData = stateMachineData.clone();
            this.peers = new HashSet<>(peers);
            this.index = index;
            this.term = term;
        }
    }
}

  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

public class OptimizedTransport {
    private final String nodeId;
    private final Map<String, Connection> connections = new ConcurrentHashMap<>();
    private final ScheduledExecutorService executor;
    private final ByteBuffer sendBuffer = ByteBuffer.allocateDirect(64 * 1024);
    private final ByteBuffer receiveBuffer = ByteBuffer.allocateDirect(64 * 1024);

    // 连接对象
    public static class Connection {
        public final String peerId;
        public final SocketChannel channel;
        public final AtomicLong lastUsed = new AtomicLong(System.currentTimeMillis());
        public final Queue<Message> sendQueue = new ConcurrentLinkedQueue<>();
        public final AtomicBoolean sending = new AtomicBoolean(false);

        public Connection(String peerId, SocketChannel channel) {
            this.peerId = peerId;
            this.channel = channel;
        }
    }

    // 消息类型
    public static class Message {
        public final MessageType type;
        public final byte[] payload;
        public final CompletableFuture<byte[]> response;

        public enum MessageType {
            APPEND_ENTRIES, REQUEST_VOTE, INSTALL_SNAPSHOT, HEARTBEAT
        }

        public Message(MessageType type, byte[] payload) {
            this.type = type;
            this.payload = payload.clone();
            this.response = new CompletableFuture<>();
        }
    }

    // 批量发送器
    public static class BatchSender {
        private final Queue<Message> batchQueue = new ConcurrentLinkedQueue<>();
        private final ScheduledExecutorService batchExecutor;
        private volatile boolean batchInProgress = false;

        public BatchSender() {
            this.batchExecutor = Executors.newSingleThreadScheduledExecutor();
            startBatchProcessing();
        }

        public void addMessage(Message message) {
            batchQueue.offer(message);
        }

        private void startBatchProcessing() {
            batchExecutor.scheduleWithFixedDelay(() -> {
                if (!batchInProgress && !batchQueue.isEmpty()) {
                    processBatch();
                }
            }, 1, 1, TimeUnit.MILLISECONDS);
        }

        private void processBatch() {
            batchInProgress = true;
            try {
                List<Message> batch = new ArrayList<>();
                Message message;
                while ((message = batchQueue.poll()) != null && batch.size() < 32) {
                    batch.add(message);
                }

                if (!batch.isEmpty()) {
                    sendBatch(batch);
                }
            } finally {
                batchInProgress = false;
            }
        }

        private void sendBatch(List<Message> messages) {
            // 批量发送消息
            for (Message message : messages) {
                // 实际发送逻辑
                try {
                    sendMessage(message);
                } catch (Exception e) {
                    message.response.completeExceptionally(e);
                }
            }
        }

        private void sendMessage(Message message) throws Exception {
            // 实现消息发送逻辑
        }
    }

    // 压缩器
    public static class MessageCompressor {
        public static byte[] compress(byte[] data) {
            if (data.length < 1024) {
                return data; // 小消息不压缩
            }

            try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                try (java.util.zip.GZIPOutputStream gzos = new java.util.zip.GZIPOutputStream(baos)) {
                    gzos.write(data);
                    gzos.finish();
                }
                return baos.toByteArray();
            } catch (Exception e) {
                return data; // 压缩失败返回原数据
            }
        }

        public static byte[] decompress(byte[] compressedData) {
            try {
                ByteArrayInputStream bais = new ByteArrayInputStream(compressedData);
                try (java.util.zip.GZIPInputStream gzis = new java.util.zip.GZIPInputStream(bais);
                     ByteArrayOutputStream baos = new ByteArrayOutputStream()) {

                    byte[] buffer = new byte[4096];
                    int bytesRead;
                    while ((bytesRead = gzis.read(buffer)) != -1) {
                        baos.write(buffer, 0, bytesRead);
                    }
                    return baos.toByteArray();
                }
            } catch (Exception e) {
                return compressedData; // 解压缩失败返回原数据
            }
        }
    }

    public OptimizedTransport(String nodeId) {
        this.nodeId = nodeId;
        this.executor = Executors.newScheduledThreadPool(4);
        startConnectionMaintenance();
    }

    // 发送AppendEntries请求（带优化）
    public CompletableFuture<AppendEntriesResponse> sendAppendEntries(
            String peerId, AppendEntriesRequest request) {

        return CompletableFuture.supplyAsync(() -> {
            try {
                // 1. 获取或创建连接
                Connection conn = getOrCreateConnection(peerId);

                // 2. 序列化请求
                byte[] payload = serializeAppendEntries(request);

                // 3. 压缩消息
                byte[] compressedPayload = MessageCompressor.compress(payload);

                // 4. 创建消息
                Message message = new Message(Message.MessageType.APPEND_ENTRIES, compressedPayload);

                // 5. 发送消息
                sendMessage(conn, message);

                // 6. 等待响应
                byte[] responseData = message.response.get(5, TimeUnit.SECONDS);

                // 7. 解压并反序列化响应
                byte[] decompressed = MessageCompressor.decompress(responseData);
                return deserializeAppendEntriesResponse(decompressed);

            } catch (Exception e) {
                throw new RuntimeException("Failed to send AppendEntries", e);
            }
        }, executor);
    }

    // 获取或创建连接
    private Connection getOrCreateConnection(String peerId) throws Exception {
        return connections.computeIfAbsent(peerId, id -> {
            try {
                SocketChannel channel = SocketChannel.open();
                channel.configureBlocking(false);
                channel.connect(new InetSocketAddress(getPeerAddress(id), getPeerPort(id)));

                // 等待连接完成
                while (!channel.finishConnect()) {
                    Thread.sleep(10);
                }

                return new Connection(id, channel);
            } catch (Exception e) {
                throw new RuntimeException("Failed to create connection to " + id, e);
            }
        });
    }

    // 异步发送消息
    private void sendMessage(Connection conn, Message message) {
        conn.sendQueue.offer(message);

        if (conn.sending.compareAndSet(false, true)) {
            executor.submit(() -> {
                try {
                    processMessageQueue(conn);
                } finally {
                    conn.sending.set(false);
                }
            });
        }
    }

    // 处理消息队列
    private void processMessageQueue(Connection conn) {
        while (!conn.sendQueue.isEmpty()) {
            Message message = conn.sendQueue.poll();
            if (message != null) {
                try {
                    sendMessageDirectly(conn, message);
                    conn.lastUsed.set(System.currentTimeMillis());
                } catch (Exception e) {
                    message.response.completeExceptionally(e);
                }
            }
        }
    }

    // 直接发送消息
    private void sendMessageDirectly(Connection conn, Message message) throws Exception {
        // 构造消息头
        synchronized (sendBuffer) {
            sendBuffer.clear();
            sendBuffer.putInt(message.type.ordinal());
            sendBuffer.putInt(message.payload.length);
            sendBuffer.put(message.payload);
            sendBuffer.flip();

            // 写入通道
            while (sendBuffer.hasRemaining()) {
                conn.channel.write(sendBuffer);
            }
        }

        // 异步等待响应
        executor.submit(() -> {
            try {
                byte[] response = readResponse(conn);
                message.response.complete(response);
            } catch (Exception e) {
                message.response.completeExceptionally(e);
            }
        });
    }

    // 读取响应
    private byte[] readResponse(Connection conn) throws Exception {
        synchronized (receiveBuffer) {
            receiveBuffer.clear();

            // 读取消息头
            while (receiveBuffer.position() < 8) {
                int bytesRead = conn.channel.read(receiveBuffer);
                if (bytesRead == -1) {
                    throw new IOException("Connection closed");
                }
                if (bytesRead == 0) {
                    Thread.sleep(1);
                }
            }

            receiveBuffer.flip();
            int messageType = receiveBuffer.getInt();
            int payloadLength = receiveBuffer.getInt();

            // 读取消息体
            byte[] payload = new byte[payloadLength];
            ByteBuffer payloadBuffer = ByteBuffer.wrap(payload);

            while (payloadBuffer.hasRemaining()) {
                int bytesRead = conn.channel.read(payloadBuffer);
                if (bytesRead == -1) {
                    throw new IOException("Connection closed");
                }
                if (bytesRead == 0) {
                    Thread.sleep(1);
                }
            }

            return payload;
        }
    }

    // 连接维护
    private void startConnectionMaintenance() {
        executor.scheduleWithFixedDelay(() -> {
            long now = System.currentTimeMillis();
            connections.entrySet().removeIf(entry -> {
                Connection conn = entry.getValue();
                if (now - conn.lastUsed.get() > 300000) { // 5分钟超时
                    try {
                        conn.channel.close();
                    } catch (Exception e) {
                        // 忽略关闭错误
                    }
                    return true;
                }
                return false;
            });
        }, 60, 60, TimeUnit.SECONDS);
    }

    // 序列化和反序列化方法
    private byte[] serializeAppendEntries(AppendEntriesRequest request) {
        // 实现序列化
        return new byte[0];
    }

    private AppendEntriesResponse deserializeAppendEntriesResponse(byte[] data) {
        // 实现反序列化
        return null;
    }

    private String getPeerAddress(String peerId) {
        // 获取节点地址
        return "localhost";
    }

    private int getPeerPort(String peerId) {
        // 获取节点端口
        return 8080;
    }
}