分布式系统基础：拜占庭将军问题深度解析

1
2
3
4
5

    将军A ←→ 将军B ←→ 将军C
      ↑         ×         ↓
      └──── 将军D ────┘

   (× 表示将军B可能发送虚假信息)

1
2
3
4
5

节点A (诚实) ←→ 节点B (恶意)
    ↕              ↕
节点C (诚实) ←→ 节点D (诚实)

问题：如何确保诚实节点对交易达成共识？

1
2
3
4
5

数据库1 ←→ 数据库2 (故障)
    ↕           ↕
数据库3 ←→ 数据库4

问题：如何保证数据的一致性和完整性？

1
2
3
4
5

诚实节点(H) + 拜占庭节点(B) = 总节点数(n)
        ↓            ↓              ↓
        └────────────┴──────→ 共识值(v)
                                   ↓
                    满足：一致性、正确性、终止性

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

系统参数：
• n：总节点数
• f：最大拜占庭故障节点数
• H：诚实节点集合，|H| = n - f
• B：拜占庭节点集合，|B| ≤ f

共识属性：
• Agreement：∀i,j ∈ H，decision_i = decision_j
• Validity：如果所有诚实节点输入相同值v，则决策值为v
• Termination：所有诚实节点最终都会产生决策

1
2
3

诚实节点数 > 拜占庭节点数
n - f > f
因此：n > 2f

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

情况1：网络分区
┌─────────────┐    ┌─────────────┐
│   组 X      │    │    组 Y     │
│ 诚实: a个   │    │  诚实: b个  │
│ 拜占庭: f个 │    │  拜占庭: 0个│
└─────────────┘    └─────────────┘

其中：a + b = n - f, a ≥ f (组X能够检测拜占庭行为)

为了保证每组都能正确工作：
a ≥ f 且 b ≥ f
因此：a + b ≥ 2f
即：n - f ≥ 2f
所以：n ≥ 3f

 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

import java.util.*;
import java.util.concurrent.ThreadLocalRandom;

public class ByzantineGeneralsOM {
    private int n; // 总节点数
    private int f; // 最大拜占庭节点数
    private Set<Integer> byzantineNodes;

    public ByzantineGeneralsOM(int n, int f) {
        this.n = n;
        this.f = f;
        this.byzantineNodes = new HashSet<>();
    }

    /**
     * OM(m)算法实现
     * @param commanderValue 指挥官的初始值
     * @param m 轮数，通常设置为f
     * @return 共识结果
     */
    public int oralMessage(int commanderValue, int m) {
        if (m == 0) {
            // 基础情况：直接返回指挥官的值
            return commanderValue;
        }

        // 第一轮：指挥官向所有副官发送值
        Map<Integer, Integer> messages = new HashMap<>();
        for (int i = 1; i < n; i++) {
            // 模拟可能的拜占庭行为
            if (isByzantine(0)) { // 如果指挥官是拜占庭的
                messages.put(i, byzantineValue(i));
            } else {
                messages.put(i, commanderValue);
            }
        }

        // 后续轮次：递归处理
        Map<Integer, Integer> results = new HashMap<>();
        for (int i = 1; i < n; i++) {
            if (!isByzantine(i)) {
                // 诚实节点运行OM(m-1)
                results.put(i, oralMessage(messages.get(i), m - 1));
            }
        }

        return majorityVote(results);
    }

    /**
     * 多数投票决策
     */
    private int majorityVote(Map<Integer, Integer> values) {
        Map<Integer, Integer> voteCount = new HashMap<>();
        for (Integer value : values.values()) {
            voteCount.put(value, voteCount.getOrDefault(value, 0) + 1);
        }

        return voteCount.entrySet().stream()
            .max(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey)
            .orElse(0);
    }

    /**
     * 检查节点是否为拜占庭节点
     */
    private boolean isByzantine(int nodeId) {
        return byzantineNodes.contains(nodeId);
    }

    /**
     * 拜占庭节点发送的恶意值
     */
    private int byzantineValue(int targetNode) {
        // 拜占庭节点可能发送不同的值给不同节点
        return ThreadLocalRandom.current().nextInt(2);
    }

    public void setByzantineNodes(Set<Integer> nodes) {
        this.byzantineNodes = nodes;
    }

    // 使用示例
    public static void main(String[] args) {
        int n = 4, f = 1; // 4个节点，最多1个拜占庭节点
        ByzantineGeneralsOM bg = new ByzantineGeneralsOM(n, f);
        bg.setByzantineNodes(Set.of(3)); // 节点3是拜占庭节点

        int result = bg.oralMessage(1, f);
        System.out.println("共识结果: " + result);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

OM(m)算法流程：

轮次0：指挥官 → 所有副官
┌─────┐    1    ┌─────┐
│ 将军0│ ──────→ │ 将军1│
│(指挥)│    1    │     │
└─────┘ ──────→ ┌─────┐
   │       1    │ 将军2│
   └──────────→ │     │
              └─────┘

轮次1：每个副官作为指挥官
将军1 → {将军2}: 值1
将军2 → {将军1}: 值1

决策阶段：
将军1收到: [1(来自指挥官), 1(来自将军2)]
将军2收到: [1(来自指挥官), 1(来自将军1)]
→ 共识：攻击(1)

  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

import java.security.*;
import java.security.spec.PKCS8EncodedKeySpec;
import java.security.spec.X509EncodedKeySpec;
import java.util.*;
import java.util.stream.Collectors;
import javax.crypto.Cipher;

class SignatureEntry {
    private int signerId;
    private String signature;

    public SignatureEntry(int signerId, String signature) {
        this.signerId = signerId;
        this.signature = signature;
    }

    public int getSignerId() { return signerId; }
    public String getSignature() { return signature; }
}

class SignedMessage {
    private int value;
    private int senderId;
    private List<SignatureEntry> signatureChain;

    public SignedMessage(int value, int senderId) {
        this.value = value;
        this.senderId = senderId;
        this.signatureChain = new ArrayList<>();
    }

    public SignedMessage(int value, int senderId, List<SignatureEntry> chain) {
        this.value = value;
        this.senderId = senderId;
        this.signatureChain = new ArrayList<>(chain);
    }

    /**
     * 为消息添加数字签名
     */
    public SignedMessage sign(PrivateKey privateKey, int signerId) throws Exception {
        // 创建消息摘要
        String messageData = String.format("{value:%d,senderId:%d,chainSize:%d}",
                                          value, senderId, signatureChain.size());

        // 生成签名
        Signature signature = Signature.getInstance("SHA256withRSA");
        signature.initSign(privateKey);
        signature.update(messageData.getBytes());
        byte[] signatureBytes = signature.sign();

        // 添加到签名链
        signatureChain.add(new SignatureEntry(signerId,
                          Base64.getEncoder().encodeToString(signatureBytes)));

        return this;
    }

    public int getValue() { return value; }
    public int getSenderId() { return senderId; }
    public List<SignatureEntry> getSignatureChain() { return signatureChain; }
}

public class ByzantineGeneralsSM {
    private int n;
    private int f;
    private Map<Integer, KeyPair> keyPairs;
    private Set<Integer> byzantineNodes;

    public ByzantineGeneralsSM(int n, int f) throws Exception {
        this.n = n;
        this.f = f;
        this.byzantineNodes = new HashSet<>();
        generateKeys();
    }

    /**
     * 为每个节点生成公私钥对
     */
    private void generateKeys() throws Exception {
        keyPairs = new HashMap<>();
        KeyPairGenerator keyGen = KeyPairGenerator.getInstance("RSA");
        keyGen.initialize(2048);

        for (int i = 0; i < n; i++) {
            KeyPair keyPair = keyGen.generateKeyPair();
            keyPairs.put(i, keyPair);
        }
    }

    /**
     * SM算法实现
     */
    public Map<Integer, Integer> signedMessageAlgorithm(int commanderValue, int commanderId)
            throws Exception {

        // 第一阶段：指挥官签名并广播
        SignedMessage initialMessage = new SignedMessage(commanderValue, commanderId);
        initialMessage.sign(keyPairs.get(commanderId).getPrivate(), commanderId);

        // 模拟消息传播
        Map<Integer, List<SignedMessage>> nodeMessages = new HashMap<>();
        for (int i = 0; i < n; i++) {
            nodeMessages.put(i, new ArrayList<>());
        }

        // 指挥官向所有节点发送签名消息
        for (int i = 0; i < n; i++) {
            if (i != commanderId) {
                nodeMessages.get(i).add(initialMessage);
            }
        }

        // 多轮转发和签名
        for (int round = 0; round < f; round++) {
            Map<Integer, List<SignedMessage>> newMessages = new HashMap<>();
            for (int i = 0; i < n; i++) {
                newMessages.put(i, new ArrayList<>());
            }

            for (int nodeId = 0; nodeId < n; nodeId++) {
                if (nodeId == commanderId || isByzantine(nodeId)) {
                    continue;
                }

                // 诚实节点转发收到的消息
                for (SignedMessage msg : nodeMessages.get(nodeId)) {
                    if (msg.getSignatureChain().size() <= f) {
                        // 添加自己的签名并转发
                        SignedMessage forwardedMsg = new SignedMessage(
                            msg.getValue(), msg.getSenderId(), msg.getSignatureChain());
                        forwardedMsg.sign(keyPairs.get(nodeId).getPrivate(), nodeId);

                        // 发送给其他节点
                        Set<Integer> signerIds = msg.getSignatureChain().stream()
                            .map(SignatureEntry::getSignerId)
                            .collect(Collectors.toSet());

                        for (int targetId = 0; targetId < n; targetId++) {
                            if (targetId != nodeId && !signerIds.contains(targetId)) {
                                newMessages.get(targetId).add(forwardedMsg);
                            }
                        }
                    }
                }
            }

            // 更新消息集合
            for (int nodeId = 0; nodeId < n; nodeId++) {
                nodeMessages.get(nodeId).addAll(newMessages.get(nodeId));
            }
        }

        // 决策阶段
        Map<Integer, Integer> decisions = new HashMap<>();
        for (int nodeId = 0; nodeId < n; nodeId++) {
            if (!isByzantine(nodeId)) {
                decisions.put(nodeId, makeDecision(nodeMessages.get(nodeId)));
            }
        }

        return decisions;
    }

    /**
     * 验证签名链的有效性
     */
    private boolean verifySignatureChain(SignedMessage message) {
        // 简化实现，实际中需要完整的签名验证
        Set<Integer> signerIds = message.getSignatureChain().stream()
            .map(SignatureEntry::getSignerId)
            .collect(Collectors.toSet());
        return signerIds.size() >= f + 1;
    }

    /**
     * 基于收到的消息做出决策
     */
    private Integer makeDecision(List<SignedMessage> messages) {
        List<SignedMessage> validMessages = messages.stream()
            .filter(this::verifySignatureChain)
            .collect(Collectors.toList());

        if (validMessages.isEmpty()) {
            return null; // 默认值
        }

        // 选择具有最多有效签名的值
        Map<Integer, Integer> valueCounts = new HashMap<>();
        for (SignedMessage msg : validMessages) {
            int value = msg.getValue();
            int weight = msg.getSignatureChain().size();
            valueCounts.put(value, valueCounts.getOrDefault(value, 0) + weight);
        }

        return valueCounts.entrySet().stream()
            .max(Map.Entry.comparingByValue())
            .map(Map.Entry::getKey)
            .orElse(null);
    }

    /**
     * 检查是否为拜占庭节点
     */
    private boolean isByzantine(int nodeId) {
        return byzantineNodes.contains(nodeId);
    }

    public void setByzantineNodes(Set<Integer> nodes) {
        this.byzantineNodes = nodes;
    }

    // 使用示例
    public static void main(String[] args) throws Exception {
        ByzantineGeneralsSM bgSm = new ByzantineGeneralsSM(4, 1);
        bgSm.setByzantineNodes(Set.of(3));

        Map<Integer, Integer> decisions = bgSm.signedMessageAlgorithm(1, 0);
        System.out.println("各节点决策结果: " + decisions);
    }
}

  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

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.time.Instant;

class BlockProposal {
    private int view;
    private int sequence;
    private String blockData;
    private long timestamp;

    public BlockProposal(int view, int sequence, String blockData) {
        this.view = view;
        this.sequence = sequence;
        this.blockData = blockData;
        this.timestamp = Instant.now().toEpochMilli();
    }

    // Getters
    public int getView() { return view; }
    public int getSequence() { return sequence; }
    public String getBlockData() { return blockData; }
    public long getTimestamp() { return timestamp; }
}

class Validator {
    private int id;
    private boolean isByzantine;

    public Validator(int id, boolean isByzantine) {
        this.id = id;
        this.isByzantine = isByzantine;
    }

    public boolean validateProposal(BlockProposal proposal) {
        // 简化的验证逻辑
        return !isByzantine && proposal.getBlockData() != null;
    }

    public String signPreprepare(BlockProposal proposal) {
        return "preprepare_" + id + "_" + proposal.getSequence();
    }

    public String signPrepare(BlockProposal proposal) {
        return "prepare_" + id + "_" + proposal.getSequence();
    }

    public String signCommit(BlockProposal proposal) {
        return "commit_" + id + "_" + proposal.getSequence();
    }

    public int getId() { return id; }
}

public class BlockchainBFT {
    private List<Validator> validators;
    private int f; // 最大拜占庭节点数
    private int currentView;
    private int sequenceNumber;

    public BlockchainBFT(List<Validator> validators) {
        this.validators = validators;
        this.f = (validators.size() - 1) / 3;
        this.currentView = 0;
        this.sequenceNumber = 0;
    }

    /**
     * 提议新区块
     */
    public boolean proposeBlock(String blockData) {
        // PBFT三阶段协议

        // 1. Pre-prepare阶段
        BlockProposal proposal = new BlockProposal(currentView, sequenceNumber++, blockData);
        List<String> preprepareVotes = broadcastPreprepare(proposal);

        // 2. Prepare阶段
        if (preprepareVotes.size() >= 2 * f + 1) {
            List<String> prepareVotes = broadcastPrepare(proposal);

            // 3. Commit阶段
            if (prepareVotes.size() >= 2 * f + 1) {
                List<String> commitVotes = broadcastCommit(proposal);

                if (commitVotes.size() >= 2 * f + 1) {
                    return finalizeBlock(blockData);
                }
            }
        }

        return false;
    }

    /**
     * 广播pre-prepare消息
     */
    private List<String> broadcastPreprepare(BlockProposal proposal) {
        List<String> votes = new ArrayList<>();
        for (Validator validator : validators) {
            if (validator.validateProposal(proposal)) {
                String vote = validator.signPreprepare(proposal);
                votes.add(vote);
            }
        }
        return votes;
    }

    /**
     * 广播prepare消息
     */
    private List<String> broadcastPrepare(BlockProposal proposal) {
        List<String> votes = new ArrayList<>();
        for (Validator validator : validators) {
            if (validator.validateProposal(proposal)) {
                String vote = validator.signPrepare(proposal);
                votes.add(vote);
            }
        }
        return votes;
    }

    /**
     * 广播commit消息
     */
    private List<String> broadcastCommit(BlockProposal proposal) {
        List<String> votes = new ArrayList<>();
        for (Validator validator : validators) {
            if (validator.validateProposal(proposal)) {
                String vote = validator.signCommit(proposal);
                votes.add(vote);
            }
        }
        return votes;
    }

    /**
     * 最终确认区块
     */
    private boolean finalizeBlock(String blockData) {
        System.out.println("区块已确认: " + blockData);
        return true;
    }

    // 使用示例
    public static void main(String[] args) {
        List<Validator> validators = Arrays.asList(
            new Validator(0, false),
            new Validator(1, false),
            new Validator(2, false),
            new Validator(3, true)  // 拜占庭节点
        );

        BlockchainBFT bft = new BlockchainBFT(validators);
        boolean success = bft.proposeBlock("新区块数据");
        System.out.println("区块提议结果: " + success);
    }
}

  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

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class Transaction {
    private String id;
    private List<String> operations;
    private long timestamp;

    public Transaction(String id, List<String> operations) {
        this.id = id;
        this.operations = operations;
        this.timestamp = System.currentTimeMillis();
    }

    public String getId() { return id; }
    public List<String> getOperations() { return operations; }
    public long getTimestamp() { return timestamp; }
}

class TransactionProposal {
    private String txId;
    private List<String> operations;
    private long timestamp;

    public TransactionProposal(String txId, List<String> operations, long timestamp) {
        this.txId = txId;
        this.operations = operations;
        this.timestamp = timestamp;
    }

    public String getTxId() { return txId; }
    public List<String> getOperations() { return operations; }
    public long getTimestamp() { return timestamp; }
}

class DatabaseReplica {
    private int id;
    private boolean isByzantine;
    private Map<String, String> data;

    public DatabaseReplica(int id, boolean isByzantine) {
        this.id = id;
        this.isByzantine = isByzantine;
        this.data = new ConcurrentHashMap<>();
    }

    public boolean validateTransaction(Transaction transaction) {
        // 简化的事务验证逻辑
        return !isByzantine && transaction.getOperations() != null && !transaction.getOperations().isEmpty();
    }

    public String voteCommit(TransactionProposal proposal) {
        if (isByzantine) {
            // 拜占庭节点可能发送恶意投票
            return "byzantine_vote_" + id;
        }
        return "commit_vote_" + id + "_" + proposal.getTxId();
    }

    public boolean commitTransaction(Transaction transaction) {
        if (isByzantine) {
            return false; // 拜占庭节点可能拒绝提交
        }

        // 模拟事务提交
        for (String operation : transaction.getOperations()) {
            data.put("tx_" + transaction.getId(), operation);
        }
        System.out.println("副本 " + id + " 提交事务: " + transaction.getId());
        return true;
    }

    public void abortTransaction(Transaction transaction) {
        System.out.println("副本 " + id + " 中止事务: " + transaction.getId());
    }

    public int getId() { return id; }
}

public class DistributedDatabase {
    private List<DatabaseReplica> replicas;
    private int f; // 最大拜占庭节点数
    private ExecutorService executor;

    public DistributedDatabase(List<DatabaseReplica> replicas) {
        this.replicas = replicas;
        this.f = (replicas.size() - 1) / 3;
        this.executor = Executors.newCachedThreadPool();
    }

    /**
     * 执行事务
     */
    public boolean executeTransaction(Transaction transaction) {
        // 使用拜占庭容错确保事务一致性

        // 1. 事务提议阶段
        TransactionProposal proposal = new TransactionProposal(
            transaction.getId(),
            transaction.getOperations(),
            transaction.getTimestamp()
        );

        // 2. 投票阶段
        List<String> votes = new ArrayList<>();
        List<CompletableFuture<String>> voteFutures = new ArrayList<>();

        for (DatabaseReplica replica : replicas) {
            CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
                if (replica.validateTransaction(transaction)) {
                    return replica.voteCommit(proposal);
                }
                return null;
            }, executor);
            voteFutures.add(future);
        }

        // 收集投票结果
        for (CompletableFuture<String> future : voteFutures) {
            try {
                String vote = future.get();
                if (vote != null && !vote.startsWith("byzantine")) {
                    votes.add(vote);
                }
            } catch (Exception e) {
                System.err.println("投票收集异常: " + e.getMessage());
            }
        }

        // 3. 决策阶段
        if (votes.size() >= 2 * f + 1) {
            // 提交事务
            System.out.println("事务 " + transaction.getId() + " 获得足够投票，开始提交");
            for (DatabaseReplica replica : replicas) {
                replica.commitTransaction(transaction);
            }
            return true;
        } else {
            // 回滚事务
            System.out.println("事务 " + transaction.getId() + " 投票不足，开始回滚");
            for (DatabaseReplica replica : replicas) {
                replica.abortTransaction(transaction);
            }
            return false;
        }
    }

    public void shutdown() {
        executor.shutdown();
    }

    // 使用示例
    public static void main(String[] args) {
        List<DatabaseReplica> replicas = Arrays.asList(
            new DatabaseReplica(0, false),
            new DatabaseReplica(1, false),
            new DatabaseReplica(2, false),
            new DatabaseReplica(3, true)  // 拜占庭节点
        );

        DistributedDatabase db = new DistributedDatabase(replicas);

        Transaction tx = new Transaction("tx_001",
            Arrays.asList("INSERT user1", "UPDATE balance"));

        boolean success = db.executeTransaction(tx);
        System.out.println("事务执行结果: " + success);

        db.shutdown();
    }
}

1
2
3
4
5
6
7
8