02. 共识算法：Raft 与 Paxos

class RaftNode:
    def __init__(self, node_id, peers):
        self.id = node_id
        self.peers = peers  # 其他节点列表
        self.state = 'Follower'
        self.current_term = 0
        self.voted_for = None
        self.log = []  # (term, command)
        
        # 超时设置（随机化避免活锁）
        self.election_timeout = random.randint(150, 300)  # 150-300ms
        self.last_heartbeat = time.time()
    
    def start_election(self):
        """Follower 超时后转为 Candidate 发起选举"""
        self.state = 'Candidate'
        self.current_term += 1
        self.voted_for = self.id
        votes = 1  # 自己投自己
        
        # 并行发送 RequestVote RPC
        for peer in self.peers:
            if self.request_vote(peer):
                votes += 1
        
        # 获得多数票则成为 Leader
        if votes > len(self.peers) // 2:
            self.become_leader()
    
    def request_vote(self, peer):
        """请求投票 RPC"""
        args = {
            'term': self.current_term,
            'candidate_id': self.id,
            'last_log_index': len(self.log) - 1,
            'last_log_term': self.log[-1][0] if self.log else 0
        }
        # 发送 RPC 并等待响应
        response = peer.handle_request_vote(args)
        return response.get('vote_granted', False)
    
    def handle_request_vote(self, args):
        """处理投票请求"""
        if args['term'] < self.current_term:
            return {'term': self.current_term, 'vote_granted': False}
        
        if (self.voted_for is None or self.voted_for == args['candidate_id']) \
           and self.is_log_at_least_up_to_date(args):
            self.voted_for = args['candidate_id']
            return {'term': self.current_term, 'vote_granted': True}
        
        return {'term': self.current_term, 'vote_granted': False}

    def append_entries(self, peer):
        """Leader 发送心跳/日志复制 RPC"""
        prev_log_index = self.next_index[peer.id] - 1
        entries = self.log[prev_log_index + 1:]
        
        args = {
            'term': self.current_term,
            'leader_id': self.id,
            'prev_log_index': prev_log_index,
            'prev_log_term': self.log[prev_log_index][0] if prev_log_index >= 0 else 0,
            'entries': entries,
            'leader_commit': self.commit_index
        }
        
        response = peer.handle_append_entries(args)
        
        if response['success']:
            self.next_index[peer.id] = len(self.log)
            self.match_index[peer.id] = len(self.log) - 1
        else:
            # 日志不一致，回退 next_index 重试
            self.next_index[peer.id] -= 1
    
    def handle_append_entries(self, args):
        """处理日志复制请求"""
        if args['term'] < self.current_term:
            return {'term': self.current_term, 'success': False}
        
        self.last_heartbeat = time.time()
        self.current_term = args['term']
        self.state = 'Follower'
        self.voted_for = None
        
        # 检查日志一致性
        if args['prev_log_index'] >= 0:
            if args['prev_log_index'] >= len(self.log):
                return {'term': self.current_term, 'success': False}
            if self.log[args['prev_log_index']][0] != args['prev_log_term']:
                return {'term': self.current_term, 'success': False}
        
        # 追加新条目
        self.log = self.log[:args['prev_log_index'] + 1] + args['entries']
        
        # 更新 commit_index
        if args['leader_commit'] > self.commit_index:
            self.commit_index = min(args['leader_commit'], len(self.log) - 1)
        
        return {'term': self.current_term, 'success': True}

    def is_log_at_least_up_to_date(self, args):
        """判断候选者日志是否至少一样新"""
        my_last_index = len(self.log) - 1
        my_last_term = self.log[-1][0] if self.log else 0
        
        candidate_last_index = args['last_log_index']
        candidate_last_term = args['last_log_term']
        
        if candidate_last_term != my_last_term:
            return candidate_last_term > my_last_term
        return candidate_last_index >= my_last_index

    def handle_configuration_change(self, new_config):
        """Joint Consensus：联合共识保证安全成员变更"""
        # 第一阶段：Cold + New 联合配置
        joint_config = self.old_config.union(new_config)
        self.configuration = joint_config
        
        # 在联合配置中达到多数，才认为日志提交
        # 第二阶段：完全切换到新配置
        if self.is_majority_in_joint_config():
            self.configuration = new_config

class Acceptor:
    def __init__(self):
        self.promised_n = -1  # 已承诺的最大提案号
        self.accepted_n = -1  # 已接受的最大提案号
        self.accepted_v = None  # 已接受的值
    
    def prepare(self, n):
        if n > self.promised_n:
            self.promised_n = n
            return {'promised': True, 'accepted_n': self.accepted_n, 
                    'accepted_v': self.accepted_v}
        return {'promised': False}
    
    def accept(self, n, v):
        if n >= self.promised_n:
            self.promised_n = n
            self.accepted_n = n
            self.accepted_v = v
            return {'accepted': True}
        return {'accepted': False}

class Proposer:
    def __init__(self, acceptors):
        self.acceptors = acceptors
        self.proposal_n = 0
    
    def propose(self, value):
        self.proposal_n += 1
        
        # Phase 1: Prepare
        promises = []
        for ac in self.acceptors:
            resp = ac.prepare(self.proposal_n)
            if resp['promised']:
                promises.append(resp)
        
        if len(promises) <= len(self.acceptors) // 2:
            return False  # 未获多数
        
        # 选择值：如果有已接受的值，选最大的 n 对应的值
        accepted_values = [(p['accepted_n'], p['accepted_v']) 
                          for p in promises if p['accepted_n'] >= 0]
        if accepted_values:
            value = max(accepted_values, key=lambda x: x[0])[1]
        
        # Phase 2: Accept
        accepts = 0
        for ac in self.acceptors:
            resp = ac.accept(self.proposal_n, value)
            if resp['accepted']:
                accepts += 1
        
        return accepts > len(self.acceptors) // 2

# 写数据
etcdctl put /config/database/host "db.example.com"
etcdctl put /config/database/port "5432"

# 读数据
etcdctl get /config/database/host

# 带前缀查询
etcdctl get /config/ --prefix

# 监听变更（Watch）
etcdctl watch /config/database/ --prefix

# 事务操作
etcdctl txn -i
put("/user/balance", "90")

# 集群状态
etcdctl endpoint status --cluster -w table

# 三节点集群
etcd --name node1 \
  --initial-advertise-peer-urls http://192.168.1.1:2380 \
  --listen-peer-urls http://192.168.1.1:2380 \
  --listen-client-urls http://192.168.1.1:2379,http://127.0.0.1:2379 \
  --advertise-client-urls http://192.168.1.1:2379 \
  --initial-cluster-token etcd-cluster-1 \
  --initial-cluster "node1=http://192.168.1.1:2380,node2=http://192.168.1.2:2380,node3=http://192.168.1.3:2380" \
  --initial-cluster-state new

# 查看集群成员
etcdctl member list -w table

import etcd3

client = etcd3.Client(host='localhost', port=2379)

# 分布式锁（基于租约 + 前缀机制）
lock = client.lock('my-resource-lock', ttl=10)
lock.acquire()
try:
    # 执行业务逻辑
    process_critical_section()
finally:
    lock.release()