分布式一致性算法详解

在分布式系统中，保证多个节点数据的一致性是核心挑战。本文深入探讨主流一致性算法的原理与应用。

1. 基本概念

一致性级别

# 强一致性（线性一致性）
def linearizable_write(key, value):
    # 任何读操作都能读到最新写入的值
    consensus.propose(key, value)
    await quorum_ack()
    return SUCCESS

# 最终一致性
def eventually_consistent_write(key, value):
    # 写入立即返回，后台异步复制
    local_storage[key] = value
    async_replicate(key, value)
    return SUCCESS  # 可能读到旧值
​

CAP定理

一致性（Consistency）
    |
    |   可用性（Availability）
    |       |
    |       |   分区容错（Partition Tolerance）
    └───────┴─────
         |
       AP（放弃强一致）或
       CP（放弃可用性）
​

2. Paxos算法

基本角色

# 提议者（Proposer）
class Proposer:
    def __init__(self, proposal_id):
        self.id = proposal_id
        self.promised_count = 0
        self.accepted_count = 0
    
    def prepare(self):
        # Phase 1: Prepare
        for acceptor in acceptors:
            response = acceptor.on_prepare(self.id)
            if response.promise:
                self.promised_count += 1
                # 记录已接受的值（如果有）
        
        if self.promised_count > len(acceptors) // 2:
            return True
        return False
    
    def accept(self, value):
        # Phase 2: Accept
        for acceptor in acceptors:
            response = acceptor.on_accept(self.id, value)
            if response.accepted:
                self.accepted_count += 1
        
        return self.accepted_count > len(acceptors) // 2

# 接受者（Acceptor）
class Acceptor:
    def __init__(self):
        self.promised_id = None
        self.accepted_id = None
        self.accepted_value = None
    
    def on_prepare(self, proposal_id):
        if proposal_id > self.promised_id:
            self.promised_id = proposal_id
            return PrepareResponse(
                promise=True,
                accepted_id=self.accepted_id,
                accepted_value=self.accepted_value
            )
        return PrepareResponse(promise=False)
    
    def on_accept(self, proposal_id, value):
        if proposal_id >= self.promised_id:
            self.promised_id = proposal_id
            self.accepted_id = proposal_id
            self.accepted_value = value
            return AcceptResponse(accepted=True)
        return AcceptResponse(accepted=False)
​

Multi-Paxos优化

# 优化：跳过Prepare阶段
class MultiPaxosProposer:
    def __init__(self):
        self.current_leader = None
    
    def fast_accept(self, instance_id, value):
        # 如果是Leader，直接进入Accept阶段
        if self.is_leader():
            return self.accept(instance_id, value)
        else:
            # 否则完整Paxos流程
            return self.full_paxos(instance_id, value)
​

3. Raft算法

核心机制

# Raft状态机
class RaftNode:
    def __init__(self, node_id):
        self.node_id = node_id
        self.state = FOLLOWER  # FOLLOWER, CANDIDATE, LEADER
        self.current_term = 0
        self.voted_for = None
        self.log = []
        self.commit_index = 0
        self.last_applied = 0
        
        # Leader专用
        self.next_index = {}
        self.match_index = {}
    
    def become_follower(self, term):
        self.state = FOLLOWER
        self.current_term = term
        self.voted_for = None
    
    def become_candidate(self):
        self.state = CANDIDATE
        self.current_term += 1
        self.voted_for = self.node_id
        # 发送RequestVote RPC
    
    def become_leader(self):
        self.state = LEADER
        # 初始化next_index和match_index
        for peer in peers:
            self.next_index[peer] = len(self.log) + 1
            self.match_index[peer] = 0
​

日志复制

# Leader处理客户端请求
def append_entries(self, command):
    # 1. 添加到本地日志
    entry = LogEntry(
        term=self.current_term,
        command=command,
        index=len(self.log) + 1
    )
    self.log.append(entry)
    
    # 2. 并行复制到Follower
    for follower in self.followers:
        self.send_append_entries(follower)
    
    # 3. 等待大多数Follower确认
    while True:
        replicated = 1  # 自己
        for follower in self.followers:
            if self.match_index[follower] >= entry.index:
                replicated += 1
        
        if replicated > (len(self.nodes) + 1) // 2:
            # 提交
            self.commit_index = entry.index
            self.apply_command(command)
            break
        
        time.sleep(0.01)

# Follower处理AppendEntries
def on_append_entries(self, request):
    # 1. 检查term
    if request.term < self.current_term:
        return AppendEntriesResponse(success=False, term=self.current_term)
    
    # 2. 更新term
    if request.term > self.current_term:
        self.current_term = request.term
        self.state = FOLLOWER
    
    # 3. 检查日志一致性
    if len(self.log) < request.prev_log_index:
        return AppendEntriesResponse(success=False)
    
    if self.log[request.prev_log_index - 1].term != request.prev_log_term:
        return AppendEntriesResponse(success=False)
    
    # 4. 追加日志
    self.log = self.log[:request.prev_log_index]
    self.log.extend(request.entries)
    
    # 5. 更新commit_index
    if request.leader_commit > self.commit_index:
        self.commit_index = min(request.leader_commit, len(self.log))
    
    return AppendEntriesResponse(success=True)
​

4. ZAB协议（Zookeeper）

协议阶段

# 阶段1：发现（Discovery）
def discovery_phase(self):
    # Follower连接Leader，获取最新事务
    leader = self.find_leader()
    response = leader.get_last_snapshot()
    
    if response.last_zxid > self.last_zxid:
        # 需要同步
        self.sync_state(leader)

# 阶段2：同步（Synchronization）
def sync_phase(self, leader):
    # 接收Leader的快照和事务
    snapshot = leader.get_snapshot()
    self.restore_snapshot(snapshot)
    
    for txn in leader.get_new_transactions():
        self.apply_transaction(txn)

# 阶段3：广播（Broadcast）
def broadcast_phase(self):
    # 正常运行，Leader广播事务
    while True:
        txn = self.receive_transaction()
        self.propose_transaction(txn)
        self.wait_for_quorum()
        self.commit_transaction(txn)
​

5. 应用场景

分布式锁

# 使用Raft实现分布式锁
class DistributedLock:
    def __init__(self, raft_node, lock_name):
        self.raft = raft_node
        self.lock_name = lock_name
    
    def acquire(self, timeout=30):
        command = {
            "type": "LOCK_ACQUIRE",
            "lock": self.lock_name,
            "owner": self.raft.node_id,
            "timeout": timeout
        }
        return self.raft.propose(command)
    
    def release(self):
        command = {
            "type": "LOCK_RELEASE",
            "lock": self.lock_name,
            "owner": self.raft.node_id
        }
        return self.raft.propose(command)
​

配置中心

# 使用Raft存储配置
class ConfigCenter:
    def __init__(self, raft_cluster):
        self.raft = raft_cluster
    
    def set_config(self, key, value):
        command = {
            "type": "SET",
            "key": key,
            "value": value
        }
        return self.raft.propose(command)
    
    def get_config(self, key):
        # 从状态机读取
        return self.raft.state_machine.get(key)
​

6. 性能优化

批量提案

# 批量处理命令
def batch_propose(self, commands):
    # 打包多个命令
    batch_entry = {
        "type": "BATCH",
        "commands": commands
    }
    return self.propose(batch_entry)
​

Pipeline复制

# Pipeline优化：不等待前一个日志确认
def pipeline_replicate(self, entries):
    for entry in entries:
        self.send_append_entries(entry)
    
    # 异步等待确认
    while not all_confirmed(entries):
        time.sleep(0.001)
​

分布式一致性算法是构建可靠分布式系统的基石，选择合适的算法需要权衡性能、复杂度和可靠性。