分布式系统一致性：从 CAP 到 Raft

分布式系统基础

什么是分布式系统？

分布式系统是多个独立计算机节点协同工作的系统，通过网络通信。

核心挑战

1. 并发：多个操作同时执行
2. 局部故障：部分节点可能失败
3. 消息延迟：网络不可靠
4. 时钟同步：节点时钟不一致

CAP 定理

CAP 三个属性

一致性（Consistency）：

• 所有节点同时看到相同的数据
• 线性一致性（Linearizability）

可用性（Availability）：

• 每个请求都能得到响应
• 响应成功或失败，但不超时

分区容错性（Partition Tolerance）：

• 系统在网络分区时仍能继续运行

CAP 权衡

           一致性（C）
              ┌────┐
              │    │
              │    │
     可用性（A）─────┴ 分区容错（P）
​

现实选择：

• CA：不考虑分区（单机系统）
• CP：保证一致性，牺牲可用性
• AP：保证可用性，牺牲一致性

一致性级别

1. 强一致性

• 读操作总是返回最新写入
• 例子：Raft、Paxos

2. 弱一致性

• 不保证立即一致
• 例子：DNS

3. 最终一致性

• 系统最终达到一致
• 例子：Cassandra、Dynamo

分布式共识算法

1. Paxos（基础算法）

角色：

• Proposer（提案者）
• Acceptor（接受者）
• Learner（学习者）

两阶段提交：

阶段 1：Prepare/Promise
Proposer → Acceptor: Prepare(n)
Acceptor → Proposer: Promise(n, accepted_value)

阶段 2：Accept/Accepted
Proposer → Acceptor: Accept(n, value)
Acceptor → Proposer: Accepted(n, value)
​

缺点：

• 难以理解和实现
• 性能开销大

2. Raft（易于理解）

核心思想：领导者选举 + 日志复制

角色划分

领导者（Leader）
    ├── 处理所有写入请求
    └── 管理日志复制

跟随者（Follower）
    ├── 接收 Leader 的请求
    └── 转换为 Candidate（选举时）

候选人（Candidate）
    ├── 请求投票
    └── 转换为 Leader 或 Follower
​

领导者选举

选举触发条件：

• Follower 在选举超时内未收到 Leader 心跳

选举过程：

# 伪代码
def start_election():
    current_term += 1
    voted_for = self
    votes_received = 1  # 投给自己
    
    # 向所有节点请求投票
    for server in servers:
        send_request_vote(server, current_term, last_log_index, last_log_term)
    
    # 等待投票结果
    while votes_received <= majority:
        if received_append_entries():
            become_follower()
            return
        elif received_vote_response():
            votes_received += 1
    
    # 获得多数票，成为 Leader
    become_leader()
​

投票规则：

• 如果 candidate 的 term 更新，投票
• 如果 candidate 的日志至少和自己一样新，投票
• 每个任期只能投一次票

日志复制

# Leader 处理客户端请求
def handle_client_request(command):
    # 添加到本地日志
    log.append(LogEntry(current_term, command))
    
    # 并行发送给所有 Follower
    for server in servers:
        send_append_entries(server, log)
    
    # 等待大多数节点复制
    while commit_index < log.index:
        if replicated_on_majority(log.index):
            commit_index = log.index
            apply_to_state_machine(log[commit_index])
    
    # 响应客户端
    return success

# Follower 处理 AppendEntries
def handle_append_entries(leader_term, prev_log_index, prev_log_term, entries, leader_commit):
    # 检查 term
    if leader_term < current_term:
        return False
    
    # 检查日志一致性
    if log[prev_log_index].term != prev_log_term:
        return False
    
    # 追加日志
    log.append(entries)
    
    # 更新提交索引
    if leader_commit > commit_index:
        commit_index = min(leader_commit, log.index)
    
    return True
​

安全性保证

1. 选举安全性：每个任期最多一个 Leader
2. 日志匹配特性：如果两个日志包含相同的索引和任期，则之前的所有条目都相同
3. 领导者完备性：如果日志条目在某任期被提交，则出现在所有更高任期的领导者日志中
4. 日志单调性：一旦提交，永不改变

3. ZAB（Zookeeper 原子广播）

特点：

• 保证事务顺序
• 崩溃恢复
• 主备切换

阶段：

1. 恢复阶段：领导者选举和日志恢复
2. 广播阶段：消息广播和同步

分布式事务

1. 两阶段提交（2PC）

阶段 1：准备阶段
协调者 → 所有参与者：准备事务
参与者 → 协调者：同意/拒绝

阶段 2：提交阶段
协调者 → 所有参与者：提交/回滚
参与者 → 协调者：确认
​

缺点：

• 阻塞协议
• 单点故障
• 数据锁定

2. 三阶段提交（3PC）

改进：

• 引入超时机制
• CanCommit 阶段

仍然有问题：

• 分区情况下无法保证一致性

3. Saga（长事务）

# Saga 实现
class Saga:
    def __init__(self):
        self.transactions = []
    
    def add_transaction(self, transaction):
        self.transactions.append(transaction)
    
    def execute(self):
        completed = []
        
        # 执行所有事务
        for tx in self.transactions:
            try:
                tx.execute()
                completed.append(tx)
            except Exception as e:
                # 补偿已执行的事务
                for tx in reversed(completed):
                    tx.compensate()
                raise e

# 使用示例
class OrderSaga(Saga):
    def __init__(self):
        super().__init__()
        self.add_transaction(CreateOrder())
        self.add_transaction(ReserveInventory())
        self.add_transaction(ProcessPayment())
        self.add_transaction(ConfirmOrder())
​

一致性哈希

问题背景

在分布式缓存中，如何均匀分配数据到不同节点？

一致性哈希算法

import hashlib

class ConsistentHashing:
    def __init__(self, nodes=None, replicas=3):
        self.replicas = replicas
        self.ring = {}
        self.sorted_keys = []
        
        if nodes:
            for node in nodes:
                self.add_node(node)
    
    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
    
    def add_node(self, node):
        for i in range(self.replicas):
            virtual_node = f"{node}:{i}"
            hash_value = self._hash(virtual_node)
            self.ring[hash_value] = node
            self.sorted_keys.append(hash_value)
        
        self.sorted_keys.sort()
    
    def remove_node(self, node):
        for i in range(self.replicas):
            virtual_node = f"{node}:{i}"
            hash_value = self._hash(virtual_node)
            del self.ring[hash_value]
            self.sorted_keys.remove(hash_value)
    
    def get_node(self, key):
        if not self.sorted_keys:
            return None
        
        hash_value = self._hash(key)
        
        # 顺时针查找第一个节点
        for ring_key in self.sorted_keys:
            if ring_key >= hash_value:
                return self.ring[ring_key]
        
        # 环绕到第一个节点
        return self.ring[self.sorted_keys[0]]
​

优势：

• 最小化数据迁移
• 负载均衡
• 容错性强

分布式锁

Redis 实现

import redis
import uuid
import time

class DistributedLock:
    def __init__(self, redis_client, lock_name, expire_time=10):
        self.redis = redis_client
        self.lock_name = f"lock:{lock_name}"
        self.expire_time = expire_time
        self.identifier = str(uuid.uuid4())
    
    def acquire(self):
        # 使用 SET NX EX 原子操作
        end_time = time.time() + self.expire_time
        
        while time.time() < end_time:
            if self.redis.set(self.lock_name, self.identifier, 
                             nx=True, ex=self.expire_time):
                return True
            time.sleep(0.001)
        
        return False
    
    def release(self):
        # Lua 脚本确保只有锁的持有者才能释放
        lua_script = '''
        if redis.call("get", KEYS[1]) == ARGV[1] then
            return redis.call("del", KEYS[1])
        else
            return 0
        end
        '''
        
        self.redis.eval(lua_script, 1, self.lock_name, self.identifier)
​

Zookeeper 实现

from kazoo.client import KazooClient

class ZKLocket:
    def __init__(self, zk, lock_path):
        self.zk = zk
        self.lock_path = lock_path
        self.lock = zk.Lock(lock_path)
    
    def acquire(self):
        self.lock.acquire()
    
    def release(self):
        self.lock.release()
​

实际应用案例

1. 分布式配置中心

特点：

• 配置推送
• 版本管理
• 灰度发布

实现：

• 使用一致性协议保证配置一致
• 长轮询或 WebSocket 推送

2. 分布式任务调度

挑战：

• 任务分片
• 故障转移
• 去重

解决方案：

• 主节点负责任务分配
• 心跳检测节点状态
• 任务状态持久化

3. 分布式数据库

分片策略：

• 水平分片
• 垂直分片
• 地理分片

一致性保证：

• 两阶段提交
• Paxos/Raft
• 最终一致性

监控与调试

关键指标

1. 延迟：P50、P95、P99
2. 吞吐量：QPS、TPS
3. 可用性：正常运行时间
4. 一致性：数据不一致率

监控工具

• Prometheus：指标收集
• Grafana：可视化
• Jaeger：分布式追踪
• ELK：日志分析

总结

分布式系统是一致性和可用性的权衡：

1. CAP 定理：理解权衡取舍
2. 共识算法：Raft 是较好选择
3. 事务处理：Saga 适合长事务
4. 数据分片：一致性哈希是基础
5. 监控告警：及时发现和处理问题

记住：没有银弹，根据业务场景选择合适的方案！