第2章：分布式系统基础¶

# CP系统示例：ZooKeeper
from kazoo.client import KazooClient

zk = KazooClient(hosts='127.0.0.1:2181')
zk.start()

# 创建节点（强一致性）
zk.create('/my_node', b'my_data')

# 读取节点（保证一致性）
data, stat = zk.get('/my_node')

# AP系统示例：Cassandra
from cassandra.cluster import Cluster

cluster = Cluster(['127.0.0.1'])
session = cluster.connect()

# 写入数据（最终一致性）
session.execute(
    "INSERT INTO users (id, name) VALUES (%s, %s)",
    (1, 'John')
)

# 读取数据（可能读到旧数据）
rows = session.execute("SELECT * FROM users WHERE id = 1")

# 2PC简化实现
class TwoPhaseCommit:
    def __init__(self, participants):
        self.participants = participants
        self.coordinator = Coordinator()

    def commit(self, transaction):
        # 准备阶段
        prepared = []
        for participant in self.participants:
            if participant.prepare(transaction):
                prepared.append(participant)

        # 如果所有参与者都准备好，提交
        if len(prepared) == len(self.participants):
            for participant in prepared:
                participant.commit(transaction)
            return True
        else:
            # 回滚
            for participant in prepared:
                participant.rollback(transaction)
            return False

# Paxos简化实现
class PaxosNode:
    def __init__(self, node_id):
        self.node_id = node_id
        self.proposed_value = None
        self.accepted_value = None
        self.promised_n = 0

    def prepare(self, n):
        if n > self.promised_n:
            self.promised_n = n
            return (True, self.accepted_value)
        return (False, None)

    def accept(self, n, value):
        if n >= self.promised_n:
            self.promised_n = n
            self.accepted_value = value
            return True
        return False

# 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

    def start_election(self):
        self.state = 'candidate'
        self.current_term += 1
        self.voted_for = self.node_id
        # 请求投票
        self.request_votes()

    def request_votes(self):
        # 向其他节点请求投票
        pass

    def append_entries(self, entries):
        # Leader追加日志
        pass

# TCC实现示例
class TCCService:
    def try_operation(self, data):
        # 预留资源
        if self.check_resource(data):
            self.reserve_resource(data)
            return True
        return False

    def confirm_operation(self, data):
        # 确认提交
        self.commit_resource(data)

    def cancel_operation(self, data):
        # 取消操作
        self.release_resource(data)

# Saga实现示例
class Saga:
    def __init__(self):
        self.transactions = []
        self.compensations = []

    def add_transaction(self, transaction, compensation):
        self.transactions.append(transaction)
        self.compensations.append(compensation)

    def execute(self):
        executed = []
        try:
            for i, transaction in enumerate(self.transactions):  # enumerate同时获取索引和值
                transaction()
                executed.append(i)
            return True
        except Exception as e:
            # 执行补偿
            for i in reversed(executed):
                self.compensations[i]()
            return False

# 本地消息表实现
class LocalMessageTable:
    def __init__(self, db):
        self.db = db

    def send_message(self, message):
        # 在事务中写入消息
        with self.db.transaction():
            self.execute_business_logic()
            self.db.insert('message_table', {
                'message': message,
                'status': 'pending'
            })

    def process_messages(self):
        # 定时处理消息
        messages = self.db.query(
            "SELECT * FROM message_table WHERE status = 'pending'"
        )
        for msg in messages:
            if self.send_to_remote(msg['message']):
                self.db.update(
                    'message_table',
                    {'status': 'sent'},
                    {'id': msg['id']}
                )

# 事务消息实现（以RocketMQ为例）
from rocketmq.client import Producer, Message

producer = Producer('transaction_producer')
producer.set_name_server_address('localhost:9876')
producer.start()

def send_transaction_message(message, local_transaction):
    # 发送事务消息
    msg = Message('test_topic', message.encode('utf-8'))
    send_result = producer.send_message_in_transaction(
        msg, local_transaction, None
    )
    return send_result

def local_transaction_execute(msg, arg):
    # 执行本地事务
    try:
        execute_business_logic()
        return TransactionStatus.COMMIT
    except Exception as e:
        return TransactionStatus.ROLLBACK

import redis
import time
import uuid

class RedisLock:
    def __init__(self, redis_client, lock_name, expire_time=30):
        self.redis = redis_client
        self.lock_name = f"lock:{lock_name}"
        self.expire_time = expire_time
        self.identifier = str(uuid.uuid4())

    def acquire(self, timeout=10):
        end_time = time.time() + timeout
        while time.time() < end_time:
            # 使用SET命令的NX选项
            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
        )

# 使用示例
redis_client = redis.Redis(host='localhost', port=6379)
lock = RedisLock(redis_client, 'my_resource')

if lock.acquire():
    try:
        # 执行需要加锁的操作
        do_something()
    finally:
        lock.release()

from kazoo.client import KazooClient
from kazoo.exceptions import NodeExistsError

class ZKLock:
    def __init__(self, zk_client, lock_path):
        self.zk = zk_client
        self.lock_path = lock_path
        self.lock_node = None

    def acquire(self, timeout=10):
        start_time = time.time()
        while time.time() - start_time < timeout:
            try:  # try/except捕获异常
                # 创建临时顺序节点
                self.lock_node = self.zk.create(
                    f"{self.lock_path}/lock-",
                    ephemeral=True,
                    sequence=True
                )

                # 检查是否是最小的节点
                children = self.zk.get_children(self.lock_path)
                children.sort()

                if self.lock_node.endswith(children[0]):
                    return True

                # 等待前一个节点释放
                self.wait_for_predecessor(children)

            except NodeExistsError:
                pass
        return False

    def release(self):
        if self.lock_node:
            self.zk.delete(self.lock_node)
            self.lock_node = None