1
# 线性化的直觉：一个寄存器的读写
2
# 假设初始值 x = 0
3

4
# 客户端 1          客户端 2          客户端 3
5
# write(x, 1)       read(x)          read(x)
6
# --- 调用 ---      --- 调用 ---
7

8
#                                     --- 调用 ---
9

10
# --- 响应 OK ---
11

12
# 后续所有 read(x) 都必须返回 1（或更新的值）
13
# 不允许出现 read→1, read→0 的回退

1
// 线性化读的代价：Quorum 读必须联系多数节点
2
func LinearizableRead(quorum int, nodes []*Node) (string, error) {
3
    // 必须从多数节点读取，才能确定最新值
4
    results := make(chan ReadResult, len(nodes))
5
    for _, node := range nodes {
6
        go func(n *Node) {
7
            val, ts, err := n.Read()
8
            results <- ReadResult{Value: val, Timestamp: ts, Err: err}
9
        }(node)
10
    }
11

12
    // 收集多数响应
13
    responses := make([]ReadResult, 0, quorum)
14
    for i := 0; i < quorum; i++ {
15
        r := <-results
16
        if r.Err != nil {
17
            continue
18
        }
19
        responses = append(responses, r)
20
    }
21

22
    // 返回时间戳最大的值（最新写入）
23
    latest := responses[0]
24
    for _, r := range responses[1:] {
25
        if r.Timestamp > latest.Timestamp {
26
            latest = r
27
        }
28
    }
29
    return latest.Value, nil
30
}

1
# 因果一致性的经典场景：社交网络的评论
2
#
3
# 用户 A 发帖："今天天气真好！"
4
# 用户 B 看到帖子后评论："确实！"
5
# 用户 C 看到评论后回复："同意！"
6
#
7
# 因果关系链：发帖 → 评论 → 回复
8
# 因果一致性要求：任何节点必须先看到"发帖"，才能看到"评论"
9
#                 先看到"评论"，才能看到"回复"
10
#
11
# 如果某个节点先看到"回复"而没看到"发帖"，就违反了因果一致性
12

13
class CausalClock:
14
    """因果时钟：跟踪已知的因果前驱"""
15
    def __init__(self):
16
        self.clock = {}  # {node_id: counter}
17

18
    def increment(self, node_id):
19
        """本地事件：递增自己的计数器"""
20
        self.clock[node_id] = self.clock.get(node_id, 0) + 1
21

22
    def merge(self, other_clock):
23
        """接收消息：合并对方的时钟，取每个分量的最大值"""
24
        for node_id, counter in other_clock.items():
25
            self.clock[node_id] = max(
26
                self.clock.get(node_id, 0), counter
27
            )
28

29
    def happens_before(self, other):
30
        """判断 self 是否 happens-before other"""
31
        all_less_or_equal = True
32
        has_strictly_less = False
33
        for node_id in set(list(self.clock.keys()) + list(other.keys())):
34
            a = self.clock.get(node_id, 0)
35
            b = other.get(node_id, 0)
36
            if a > b:
37
                all_less_or_equal = False
38
                break
39
            if a < b:
40
                has_strictly_less = True
41
        return all_less_or_equal and has_strictly_less

1
class LamportClock:
2
    """Lamport 逻辑时钟"""
3
    def __init__(self, node_id: int):
4
        self.node_id = node_id
5
        self.counter = 0
6

7
    def local_event(self) -> int:
8
        """本地事件"""
9
        self.counter += 1
10
        return self.counter
11

12
    def send_event(self) -> tuple[int, int]:
13
        """发送事件：返回 (counter, node_id) 作为时间戳"""
14
        self.counter += 1
15
        return (self.counter, self.node_id)
16

17
    def recv_event(self, msg_timestamp: tuple[int, int]) -> int:
18
        """接收事件：取 max 后 +1"""
19
        msg_counter, _ = msg_timestamp
20
        self.counter = max(self.counter, msg_counter) + 1
21
        return self.counter
22

23
    @staticmethod
24
    def compare(ts_a: tuple[int, int], ts_b: tuple[int, int]) -> int:
25
        """比较两个 Lamport 时间戳：先比 counter，再比 node_id"""
26
        if ts_a[0] != ts_b[0]:
27
            return -1 if ts_a[0] < ts_b[0] else 1
28
        return -1 if ts_a[1] < ts_b[1] else (1 if ts_a[1] > ts_b[1] else 0)

1
from typing import Dict
2

3
class VectorClock:
4
    """向量时钟：精确捕获因果关系"""
5
    def __init__(self, node_id: int, all_nodes: list[int]):
6
        self.node_id = node_id
7
        self.clock: Dict[int, int] = {n: 0 for n in all_nodes}
8

9
    def local_event(self):
10
        self.clock[self.node_id] += 1
11
        return dict(self.clock)
12

13
    def send_event(self):
14
        self.clock[self.node_id] += 1
15
        return dict(self.clock)
16

17
    def recv_event(self, msg_clock: Dict[int, int]):
18
        for node, counter in msg_clock.items():
19
            self.clock[node] = max(self.clock.get(node, 0), counter)
20
        self.clock[self.node_id] += 1
21
        return dict(self.clock)
22

23
    @staticmethod
24
    def compare(a: Dict[int, int], b: Dict[int, int]) -> str:
25
        """
26
        比较两个向量时钟：
27
        - 'before': a happens-before b
28
        - 'after': b happens-before a
29
        - 'concurrent': 并发（无因果关系）
30
        - 'equal': 相同
31
        """
32
        a_less = any(a.get(n, 0) < b.get(n, 0) for n in set(a) | set(b))
33
        b_less = any(b.get(n, 0) < a.get(n, 0) for n in set(a) | set(b))
34

35
        if not a_less and not b_less:
36
            return 'equal'
37
        if a_less and not b_less:
38
            return 'before'
39
        if b_less and not a_less:
40
            return 'after'
41
        return 'concurrent'  # 两个方向都有分量更小 → 并发

1
// 全序广播的抽象接口
2
type TotalOrderBroadcast interface {
3
    // 广播一条消息：所有节点以相同顺序接收
4
    Broadcast(msg []byte) error
5

6
    // 接收下一条消息：按全局确定的顺序返回
7
    Receive() (msg []byte, err error)
8
}
9

10
// 基于全序广播实现复制日志
11
type ReplicatedLog struct {
12
    broadcast TotalOrderBroadcast
13
    log       [][]byte
14
    commitIdx int
15
}
16

17
func (rl *ReplicatedLog) Append(entry []byte) error {
18
    // 将条目广播到所有节点，全序广播保证顺序一致
19
    return rl.broadcast.Broadcast(entry)
20
}
21

22
func (rl *ReplicatedLog) Read(offset int) ([]byte, error) {
23
    // 所有节点在相同 offset 读到相同内容
24
    if offset >= len(rl.log) {
25
        return nil, fmt.Errorf("offset %d not yet committed", offset)
26
    }
27
    return rl.log[offset], nil
28
}

1
# 用全序广播实现线性化读写
2
class LinearizableKVStore:
3
    def __init__(self, broadcast):
4
        self.broadcast = broadcast  # 全序广播
5
        self.state = {}             # 本地状态
6
        self.pending_reads = {}     # 等待中的读请求
7

8
    def write(self, key, value):
9
        """线性化写：将写操作追加到日志"""
10
        entry = {"op": "write", "key": key, "value": value}
11
        self.broadcast.broadcast(entry)
12
        # 等待条目被提交（简化示意）
13

14
    def read(self, key):
15
        """线性化读：将读请求也追加到日志"""
16
        request_id = generate_uuid()
17
        entry = {"op": "read", "key": key, "request_id": request_id}
18
        self.broadcast.broadcast(entry)
19
        # 等待该条目被提交后返回结果
20
        return self.pending_reads.pop(request_id)
21

22
    def apply_entry(self, entry):
23
        """按日志顺序应用条目"""
24
        if entry["op"] == "write":
25
            self.state[entry["key"]] = entry["value"]
26
        elif entry["op"] == "read":
27
            # 读操作被排在日志中，保证了线性化语义
28
            self.pending_reads[entry["request_id"]] = \
29
                self.state.get(entry["key"])

1
// Raft 节点状态
2
const (
3
    Follower  = "Follower"
4
    Candidate = "Candidate"
5
    Leader    = "Leader"
6
)
7

8
type RaftNode struct {
9
    nodeID      int
10
    state       string
11
    currentTerm int
12
    votedFor    int          // 当前任期投给了谁
13
    log         []LogEntry
14
    commitIndex int
15
    lastApplied int
16

17
    // 选举相关
18
    electionTimeout time.Duration
19
    heartbeatTimer  *time.Timer
20

21
    // 领导者相关（仅 Leader 使用）
22
    nextIndex  []int // 每个跟随者的下一条日志索引
23
    matchIndex []int // 每个跟随者的已匹配日志索引
24
}
25

26
// 发起选举
27
func (rn *RaftNode) startElection() {
28
    rn.state = Candidate
29
    rn.currentTerm++
30
    rn.votedFor = rn.nodeID // 投给自己
31

32
    votesReceived := 1 // 自己的一票
33

34
    // 向所有其他节点请求投票
35
    for _, peer := range rn.peers {
36
        go func(p int) {
37
            args := RequestVoteArgs{
38
                Term:         rn.currentTerm,
39
                CandidateID:  rn.nodeID,
40
                LastLogIndex: len(rn.log) - 1,
41
                LastLogTerm:  rn.log[len(rn.log)-1].Term,
42
            }
43
            reply := rn.sendRequestVote(p, args)
44
            if reply.VoteGranted {
45
                votesReceived++
46
                if votesReceived > len(rn.peers)/2+1 &&
47
                    rn.state == Candidate {
48
                    rn.becomeLeader()
49
                }
50
            }
51
        }(peer)
52
    }
53
}

1
// AppendEntries RPC 处理（跟随者侧）
2
func (rn *RaftNode) handleAppendEntries(args AppendEntriesArgs) AppendEntriesReply {
3
    reply := AppendEntriesReply{}
4

5
    // 规则 1：任期检查
6
    if args.Term < rn.currentTerm {
7
        reply.Success = false
8
        reply.Term = rn.currentTerm
9
        return reply
10
    }
11

12
    // 规则 2：日志匹配检查
13
    if args.PrevLogIndex >= 0 {
14
        if args.PrevLogIndex >= len(rn.log) ||
15
            rn.log[args.PrevLogIndex].Term != args.PrevLogTerm {
16
            // 日志不一致，拒绝，让领导者回退
17
            reply.Success = false
18
            reply.Term = rn.currentTerm
19
            return reply
20
        }
21
    }
22

23
    // 规则 3：追加新条目（处理冲突：截断不一致的条目）
24
    for i, entry := range args.Entries {
25
        logIdx := args.PrevLogIndex + 1 + i
26
        if logIdx < len(rn.log) {
27
            if rn.log[logIdx].Term != entry.Term {
28
                // 冲突：截断从此处开始的所有条目
29
                rn.log = rn.log[:logIdx]
30
                rn.log = append(rn.log, entry)
31
            }
32
            // 否则：已存在且一致，跳过
33
        } else {
34
            rn.log = append(rn.log, entry)
35
        }
36
    }
37

38
    // 规则 4：更新提交索引
39
    if args.LeaderCommit > rn.commitIndex {
40
        lastNewIdx := args.PrevLogIndex + len(args.Entries)
41
        rn.commitIndex = min(args.LeaderCommit, lastNewIdx)
42
    }
43

44
    reply.Success = true
45
    reply.Term = rn.currentTerm
46
    return reply
47
}

1
// 投票者的选举限制检查
2
func (rn *RaftNode) handleRequestVote(args RequestVoteArgs) RequestVoteReply {
3
    reply := RequestVoteReply{Term: rn.currentTerm}
4

5
    // 任期检查
6
    if args.Term < rn.currentTerm {
7
        reply.VoteGranted = false
8
        return reply
9
    }
10

11
    // 投票限制：每个任期最多投一票
12
    if rn.votedFor != -1 && rn.votedFor != args.CandidateID {
13
        reply.VoteGranted = false
14
        return reply
15
    }
16

17
    // 选举限制：候选人的日志必须至少和自己一样新
18
    lastLogIdx := len(rn.log) - 1
19
    lastLogTerm := rn.log[lastLogIdx].Term
20
    if args.LastLogTerm < lastLogTerm ||
21
        (args.LastLogTerm == lastLogTerm && args.LastLogIndex < lastLogIdx) {
22
        reply.VoteGranted = false
23
        return reply
24
    }
25

26
    // 通过所有检查，授予投票
27
    rn.votedFor = args.CandidateID
28
    rn.currentTerm = args.Term
29
    reply.VoteGranted = true
30
    return reply
31
}

1
# Basic Paxos 的 Acceptor 实现
2
class Acceptor:
3
    def __init__(self, acceptor_id: int):
4
        self.acceptor_id = acceptor_id
5
        self.promised_n = -1       # 承诺过的最大提案号
6
        self.accepted_n = -1       # 已接受的最大提案号
7
        self.accepted_value = None # 已接受的值
8

9
    def prepare(self, n: int) -> dict:
10
        """处理 Prepare 请求"""
11
        if n > self.promised_n:
12
            self.promised_n = n
13
            return {
14
                "ok": True,
15
                "accepted_n": self.accepted_n,
16
                "accepted_value": self.accepted_value
17
            }
18
        return {"ok": False}  # 拒绝：已承诺更高编号
19

20
    def accept(self, n: int, value) -> dict:
21
        """处理 Accept 请求"""
22
        if n >= self.promised_n:
23
            self.promised_n = n
24
            self.accepted_n = n
25
            self.accepted_value = value
26
            return {"ok": True}
27
        return {"ok": False}  # 拒绝：已承诺更高编号
28

29
# Basic Paxos 的 Proposer 实现
30
class Proposer:
31
    def __init__(self, proposer_id: int, acceptors: list):
32
        self.proposer_id = proposer_id
33
        self.acceptors = acceptors
34

35
    def propose(self, value) -> bool:
36
        """两阶段提案"""
37
        n = self.generate_proposal_number()
38

39
        # 阶段一：Prepare
40
        promises = []
41
        for acc in self.acceptors:
42
            resp = acc.prepare(n)
43
            if resp["ok"]:
44
                promises.append(resp)
45

46
        if len(promises) < len(self.acceptors) // 2 + 1:
47
            return False  # 未获多数承诺
48

49
        # 如果有已接受的值，必须使用编号最大的那个
50
        max_accepted_n = -1
51
        for p in promises:
52
            if p["accepted_n"] > max_accepted_n:
53
                max_accepted_n = p["accepted_n"]
54
                value = p["accepted_value"]
55

56
        # 阶段二：Accept
57
        accepted = 0
58
        for acc in self.acceptors:
59
            resp = acc.accept(n, value)
60
            if resp["ok"]:
61
                accepted += 1
62

63
        return accepted >= len(self.acceptors) // 2 + 1
64

65
    def generate_proposal_number(self) -> int:
66
        """生成全局唯一且递增的提案号"""
67
        # 简化实现：时间戳 * 节点数 + proposer_id
68
        import time
69
        return int(time.time() * 1000) * 100 + self.proposer_id

1
// Fencing Token 防脑裂
2
type FencingService struct {
3
    currentToken int64
4
    mu           sync.Mutex
5
}
6

7
// 领导者获得新的 Fencing Token（由共识服务颁发）
8
func (fs *FencingService) NewToken() int64 {
9
    fs.mu.Lock()
10
    defer fs.mu.Unlock()
11
    fs.currentToken++
12
    return fs.currentToken
13
}
14

15
// 共享资源（如分布式锁服务、存储系统）
16
type SharedResource struct {
17
    lastSeenToken int64
18
    mu            sync.Mutex
19
}
20

21
// 所有请求必须携带 Fencing Token
22
func (sr *SharedResource) Write(token int64, key string, value string) error {
23
    sr.mu.Lock()
24
    defer sr.mu.Unlock()
25

26
    if token < sr.lastSeenToken {
27
        return fmt.Errorf("fencing error: token %d < last seen %d, "+
28
            "possible stale leader", token, sr.lastSeenToken)
29
    }
30

31
    sr.lastSeenToken = token
32
    // 执行写入
33
    sr.data[key] = value
34
    return nil
35
}

1
# ZooKeeper 分布式锁 + Fencing Token
2
class ZKDistributedLock:
3
    def __init__(self, zk_client, lock_path="/locks"):
4
        self.zk = zk_client
5
        self.lock_path = lock_path
6
        self.my_node = None
7
        self.fencing_token = None
8

9
    def acquire(self) -> int:
10
        """获取锁，返回 Fencing Token"""
11
        # 1. 创建临时顺序节点
12
        self.my_node = self.zk.create(
13
            f"{self.lock_path}/lock-",
14
            ephemeral=True,
15
            sequential=True
16
        )
17
        # my_node = "/locks/lock-0000000042"
18

19
        # 2. 提取序号作为 Fencing Token
20
        self.fencing_token = int(self.my_node.split("-")[-1])
21

22
        # 3. 检查是否获得锁（序号最小）
23
        children = self.zk.get_children(self.lock_path)
24
        children.sort()
25

26
        if self.my_node == f"{self.lock_path}/{children[0]}":
27
            return self.fencing_token  # 获得锁
28

29
        # 4. 等待前一个节点删除（Watch 机制）
30
        prev_idx = children.index(self.my_node.split('/')[-1]) - 1
31
        prev_node = f"{self.lock_path}/{children[prev_idx]}"
32
        self.zk.wait_for_deletion(prev_node)
33
        return self.fencing_token
34

35
    def release(self):
36
        """释放锁：删除临时节点"""
37
        if self.my_node:
38
            self.zk.delete(self.my_node)
39
            self.my_node = None