分布式系统一致性算法深度解析：从Paxos到Raft

 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

from enum import Enum
from typing import Dict, List, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta

class CAPProperty(Enum):
    CONSISTENCY = "一致性"    # 所有节点同时看到相同数据
    AVAILABILITY = "可用性"    # 每个请求都能得到响应
    PARTITION_TOLERANCE = "分区容错"  # 系统在网络分区时仍能运行

@dataclass
class SystemState:
    """系统状态"""
    nodes: List[str]
    leader: Optional[str]
    term: int
    log: List[dict]
    commit_index: int
    last_applied: int

class DistributedSystem:
    """分布式系统基类"""

    def __init__(self, nodes: List[str]):
        self.nodes = nodes
        self.state = SystemState(
            nodes=nodes,
            leader=None,
            term=0,
            log=[],
            commit_index=0,
            last_applied=0
        )
        self.rpc_timeout = 1.0  # RPC超时时间（秒）

    async def send_rpc(self, from_node: str, to_node: str, method: str, **kwargs) -> dict:
        """发送RPC消息"""
        # 模拟网络延迟
        await self._simulate_network_delay()

        # 检查节点是否可达
        if to_node not in self.nodes:
            return {"error": "node_not_found"}

        # 模拟可能的网络分区
        if self._is_partitioned(from_node, to_node):
            return {"error": "network_partition"}

        return {"success": True}

    async def _simulate_network_delay(self):
        """模拟网络延迟"""
        import random
        delay = random.uniform(0.01, 0.1)
        await asyncio.sleep(delay)

    def _is_partitioned(self, node1: str, node2: str) -> bool:
        """检查两个节点之间是否存在网络分区"""
        # 模拟分区场景
        return False

 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

from typing import Optional
from dataclasses import dataclass, field
from enum import Enum

class NodeState(Enum):
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"

@dataclass
class LogEntry:
    """日志条目"""
    index: int
    term: int
    command: str
    is_committed: bool = False

@dataclass
class RaftNode:
    """Raft节点"""
    node_id: str
    state: NodeState = NodeState.FOLLOWER
    current_term: int = 0
    voted_for: Optional[str] = None
    log: List[LogEntry] = field(default_factory=list)
    commit_index: int = 0
    last_applied: int = 0

    # Leader特有状态
    next_index: Dict[str, int] = field(default_factory=dict)
    match_index: Dict[str, int] = field(default_factory=dict)

    # 超时设置
    election_timeout: float = 5.0
    heartbeat_interval: float = 1.0

    # 计时器
    last_heartbeat: datetime = field(default_factory=datetime.now)
    last_election_attempt: datetime = field(default_factory=datetime.now)

    def __post_init__(self):
        """初始化后处理"""
        self.peers: List[RaftNode] = []
        self.vote_count: int = 0

    def get_last_log_index(self) -> int:
        """获取最后一条日志的索引"""
        return len(self.log) - 1

    def get_last_log_term(self) -> int:
        """获取最后一条日志的任期"""
        if not self.log:
            return 0
        return self.log[-1].term

    def get_log_term(self, index: int) -> int:
        """获取指定索引的日志任期"""
        if index < 0 or index >= len(self.log):
            return 0
        return self.log[index].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

import asyncio
import random

class RaftElection:
    """Raft选举机制"""

    def __init__(self, node: RaftNode):
        self.node = node

    async def start_election(self) -> bool:
        """开始选举"""
        # 转换为候选者状态
        self.node.state = NodeState.CANDIDATE
        self.node.current_term += 1
        self.node.voted_for = self.node.node_id
        self.node.vote_count = 1  # 自己投给自己
        self.node.last_election_attempt = datetime.now()

        print(f"[Node {self.node.node_id}] 开始选举, Term: {self.node.current_term}")

        # 向所有其他节点发送RequestVote RPC
        votes_needed = len(self.node.peers) // 2 + 1
        request_vote = RequestVoteRPC(
            term=self.node.current_term,
            candidate_id=self.node.node_id,
            last_log_index=self.node.get_last_log_index(),
            last_log_term=self.node.get_last_log_term()
        )

        # 并发发送投票请求
        tasks = [
            self._request_vote(peer, request_vote)
            for peer in self.node.peers
            if peer.node_id != self.node.node_id
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        # 统计选票
        for result in results:
            if isinstance(result, bool) and result:
                self.node.vote_count += 1

        # 检查是否获得多数票
        if self.node.vote_count >= votes_needed:
            self.node.state = NodeState.LEADER
            print(f"[Node {self.node.node_id}] 当选为Leader, Term: {self.node.current_term}")
            await self._become_leader()
            return True

        print(f"[Node {self.node.node_id}] 选举失败, 得票: {self.node.vote_count}/{votes_needed}")
        self.node.state = NodeState.FOLLOWER
        return False

    async def _request_vote(self, peer: RaftNode, rpc: 'RequestVoteRPC') -> bool:
        """请求投票"""
        try:
            # 模拟网络延迟
            await asyncio.sleep(random.uniform(0.01, 0.05))

            # 对方的决策逻辑
            if rpc.term < peer.current_term:
                return False  # 对方的任期更大

            if rpc.term > peer.current_term:
                # 发现更大的任期，更新并投票
                peer.current_term = rpc.term
                peer.voted_for = rpc.candidate_id
                peer.state = NodeState.FOLLOWER
                return True

            # 任期相同，检查是否已投票
            if peer.voted_for is None or peer.voted_for == rpc.candidate_id:
                # 检查日志是否至少一样新
                if (rpc.last_log_term > peer.get_last_log_term() or
                    (rpc.last_log_term == peer.get_last_log_term() and
                     rpc.last_log_index >= peer.get_last_log_index())):
                    peer.voted_for = rpc.candidate_id
                    return True

            return False

        except Exception as e:
            print(f"[Node {self.node.node_id}] 请求 {peer.node_id} 投票失败: {e}")
            return False

    async def _become_leader(self):
        """成为Leader后的初始化"""
        # 初始化next_index和match_index
        for peer in self.node.peers:
            if peer.node_id != self.node.node_id:
                self.node.next_index[peer.node_id] = self.node.get_last_log_index() + 1
                self.node.match_index[peer.node_id] = 0

        # 立即发送心跳
        await self._send_heartbeat()

    async def _send_heartbeat(self):
        """发送心跳（空的AppendEntries RPC）"""
        while self.node.state == NodeState.LEADER:
            for peer in self.node.peers:
                if peer.node_id != self.node.node_id:
                    await self._send_append_entries(peer, empty=True)

            await asyncio.sleep(self.node.heartbeat_interval)

@dataclass
class RequestVoteRPC:
    """RequestVote RPC参数"""
    term: int
    candidate_id: str
    last_log_index: int
    last_log_term: int

  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

class RaftLogReplication:
    """Raft日志复制"""

    def __init__(self, node: RaftNode):
        self.node = node

    async def replicate_log(self, command: str) -> bool:
        """复制日志到所有节点"""
        if self.node.state != NodeState.LEADER:
            return False

        # 添加到本地日志
        new_entry = LogEntry(
            index=self.node.get_last_log_index() + 1,
            term=self.node.current_term,
            command=command
        )
        self.node.log.append(new_entry)

        print(f"[Leader {self.node.node_id}] 添加日志: Index {new_entry.index}, Term {new_entry.term}")

        # 复制到所有节点
        success_count = 1  # 包括Leader自己
        tasks = []

        for peer in self.node.peers:
            if peer.node_id != self.node.node_id:
                tasks.append(self._replicate_to_peer(peer))

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for result in results:
            if isinstance(result, bool) and result:
                success_count += 1

        # 检查是否被多数节点接受
        if success_count > len(self.node.peers) // 2:
            # 更新commit_index
            self.node.commit_index = new_entry.index
            new_entry.is_committed = True
            print(f"[Leader {self.node.node_id}] 日志 {new_entry.index} 已提交")
            return True

        return False

    async def _replicate_to_peer(self, peer: RaftNode) -> bool:
        """复制日志到指定节点"""

        # 如果next_index表明存在不一致，回退
        next_idx = self.node.next_index.get(peer.node_id, 0)

        if next_idx > len(self.node.log):
            return False

        # 构建AppendEntries RPC
        prev_log_index = next_idx - 1
        prev_log_term = self.node.log[prev_log_index].term if prev_log_index >= 0 else 0

        entries = self.node.log[next_idx:]

        rpc = AppendEntriesRPC(
            term=self.node.current_term,
            leader_id=self.node.node_id,
            prev_log_index=prev_log_index,
            prev_log_term=prev_log_term,
            entries=entries,
            leader_commit=self.node.commit_index
        )

        return await self._send_append_entries(peer, empty=False, rpc=rpc)

    async def _send_append_entries(
        self,
        peer: RaftNode,
        empty: bool = False,
        rpc: Optional[AppendEntriesRPC] = None
    ) -> bool:
        """发送AppendEntries RPC"""

        if not rpc:
            rpc = AppendEntriesRPC(
                term=self.node.current_term,
                leader_id=self.node.node_id,
                prev_log_index=self.node.get_last_log_index(),
                prev_log_term=self.node.get_last_log_term(),
                entries=[],
                leader_commit=self.node.commit_index
            )

        try:
            # 模拟网络延迟
            await asyncio.sleep(random.uniform(0.01, 0.05))

            # 如果term过期，转为follower
            if rpc.term > peer.current_term:
                peer.current_term = rpc.term
                peer.state = NodeState.FOLLOWER
                return False

            # 如果term较小，拒绝
            if rpc.term < peer.current_term:
                return False

            # 检查日志一致性
            if rpc.prev_log_index >= 0:
                if (rpc.prev_log_index >= len(peer.log) or
                    peer.get_log_term(rpc.prev_log_index) != rpc.prev_log_term):
                    # 日志不一致，返回失败
                    return False

            # 日志一致，添加新条目
            if rpc.entries:
                # 删除冲突的条目
                peer.log = peer.log[:rpc.prev_log_index + 1]
                # 添加新条目
                peer.log.extend(rpc.entries)

            # 更新commit_index
            if rpc.leader_commit > peer.commit_index:
                peer.commit_index = min(rpc.leader_commit, len(peer.log) - 1)

            # 更新Leader的next_index和match_index
            if not empty:
                last_new_entry = rpc.prev_log_index + len(rpc.entries)
                self.node.next_index[peer.node_id] = last_new_entry + 1
                self.node.match_index[peer.node_id] = last_new_entry

            return True

        except Exception as e:
            print(f"[Leader {self.node.node_id}] 复制到 {peer.node_id} 失败: {e}")
            return False

@dataclass
class AppendEntriesRPC:
    """AppendEntries RPC参数"""
    term: int
    leader_id: str
    prev_log_index: int
    prev_log_term: int
    entries: List[LogEntry]
    leader_commit: int

  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

class PaxosProposer:
    """Paxos提议者"""

    def __init__(self, proposer_id: str):
        self.proposer_id = proposer_id
        self.proposal_number = 0
        self.accepted_value = None
        self.promises_received = set()
        self.accepts_received = set()

    async def propose(self, value: str, quorum_size: int) -> Optional[str]:
        """发起提议"""

        # Phase 1: Prepare
        self.proposal_number += 1
        self.promises_received.clear()

        print(f"[Proposer {self.proposer_id}] Phase 1: Prepare (N={self.proposal_number})")

        prepare_msg = PrepareMessage(
            proposal_number=self.proposal_number,
            proposer_id=self.proposer_id
        )

        # 发送Prepare到所有Acceptors
        tasks = [
            self._send_prepare(acceptor, prepare_msg)
            for acceptor in self.acceptors
        ]

        responses = await asyncio.gather(*tasks, return_exceptions=True)

        # 统计Promise响应
        for response in responses:
            if isinstance(response, PromiseMessage):
                self.promises_received.add(response.acceptor_id)
                # 如果有已接受的值，使用它
                if response.accepted_number > 0:
                    self.accepted_value = response.accepted_value

        # 检查是否获得多数Promise
        if len(self.promises_received) < quorum_size:
            print(f"[Proposer {self.proposer_id}] Phase 1 失败: 未获得多数Promise")
            return None

        print(f"[Proposer {self.proposer_id}] Phase 1 成功: 获得 {len(self.promises_received)} 个Promise")

        # Phase 2: Accept
        self.accepts_received.clear()

        # 使用已接受的值或提议的新值
        value_to_propose = self.accepted_value if self.accepted_value else value

        accept_msg = AcceptMessage(
            proposal_number=self.proposal_number,
            value=value_to_propose,
            proposer_id=self.proposer_id
        )

        print(f"[Proposer {self.proposer_id}] Phase 2: Accept (Value={value_to_propose})")

        # 发送Accept到所有Acceptors
        tasks = [
            self._send_accept(acceptor, accept_msg)
            for acceptor in self.acceptors
        ]

        responses = await asyncio.gather(*tasks, return_exceptions=True)

        # 统计Accepted响应
        for response in responses:
            if isinstance(response, AcceptedMessage):
                self.accepts_received.add(response.acceptor_id)

        # 检查是否获得多数Accepted
        if len(self.accepts_received) >= quorum_size:
            print(f"[Proposer {self.proposer_id}] Phase 2 成功: 值 '{value_to_propose}' 已达成共识")
            return value_to_propose

        print(f"[Proposer {self.proposer_id}] Phase 2 失败: 未获得多数Accepted")
        return None

    async def _send_prepare(self, acceptor: 'PaxosAcceptor', msg: PrepareMessage) -> Optional['PromiseMessage']:
        """发送Prepare消息"""
        try:
            await asyncio.sleep(random.uniform(0.01, 0.05))
            return await acceptor.receive_prepare(msg)
        except Exception as e:
            print(f"[Proposer {self.proposer_id}] Prepare失败: {e}")
            return None

    async def _send_accept(self, acceptor: 'PaxosAcceptor', msg: AcceptMessage) -> Optional['AcceptedMessage']:
        """发送Accept消息"""
        try:
            await asyncio.sleep(random.uniform(0.01, 0.05))
            return await acceptor.receive_accept(msg)
        except Exception as e:
            print(f"[Proposer {self.proposer_id}] Accept失败: {e}")
            return None

@dataclass
class PrepareMessage:
    proposal_number: int
    proposer_id: str

@dataclass
class PromiseMessage:
    acceptor_id: str
    proposal_number: int
    accepted_number: int = 0
    accepted_value: Optional[str] = None

@dataclass
class AcceptMessage:
    proposal_number: int
    value: str
    proposer_id: str

@dataclass
class AcceptedMessage:
    acceptor_id: str
    proposal_number: int

 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

class EPaxosNode:
    """EPaxOS节点 - 优化Leaderless场景"""

    def __init__(self, node_id: str, cluster: List[str]):
        self.node_id = node_id
        self.cluster = cluster
        self.instance_number = 0
        self.dependencies: Dict[int, set] = {}
        self.log: Dict[int, str] = {}

    async def fast_write(self, key: str, value: str) -> bool:
        """快速写入（单轮RPC）"""

        self.instance_number += 1
        instance = self.instance_number

        # 选择依赖集合
        deps = await self._get_dependencies(key)

        # 发送PreAccept到所有节点
        preaccept_msg = PreAcceptMessage(
            instance=instance,
            key=key,
            value=value,
            dependencies=deps,
            from_node=self.node_id
        )

        print(f"[EPaxOS {self.node_id}] Fast Write: {key}={value}")

        # 并发发送
        tasks = [
            self._send_preaccept(node, preaccept_msg)
            for node in self.cluster
            if node != self.node_id
        ]

        responses = await asyncio.gather(*tasks, return_exceptions=True)

        # 统计响应
        success_count = 1  # 包括自己
        all_deps_match = True

        for response in responses:
            if isinstance(response, PreAcceptResponse):
                success_count += 1
                # 检查依赖是否一致
                if response.dependencies != deps:
                    all_deps_match = False

        # 如果多数节点同意且依赖一致，快速提交
        quorum = len(self.cluster) // 2 + 1
        if success_count >= quorum and all_deps_match:
            self.log[instance] = f"{key}={value}"
            print(f"[EPaxOS {self.node_id}] Fast Commit: {key}={value}")
            return True

        # 否则回退到标准Paxos
        return await self._slow_write(instance, key, value, deps)

    async def _get_dependencies(self, key: str) -> set:
        """获取依赖集合"""
        # 简化实现：查找同一key的最新实例
        deps = set()
        for inst, val in self.log.items():
            if val.startswith(f"{key}="):
                deps.add(inst)
        return deps

    async def _send_preaccept(self, node: str, msg: PreAcceptMessage) -> Optional[PreAcceptResponse]:
        """发送PreAccept消息"""
        # 实际实现中这里是RPC调用
        await asyncio.sleep(random.uniform(0.01, 0.05))
        return PreAcceptResponse(
            from_node=node,
            dependencies=msg.dependencies.copy()
        )

    async def _slow_write(self, instance: int, key: str, value: str, deps: set) -> bool:
        """慢速写入（标准Paxos）"""
        print(f"[EPaxOS {self.node_id}] Slow Write: {key}={value}")
        # 实现标准Paxos Accept阶段
        return True

@dataclass
class PreAcceptMessage:
    instance: int
    key: str
    value: str
    dependencies: set
    from_node: str

@dataclass
class PreAcceptResponse:
    from_node: str
    dependencies: set