3.1.1 CAP / PACELC / BASE 的工程含义

# 一个直观的演示:分区真的发生时,你在 C 和 A 之间只能选一个
def on_network_partition(node_a, node_b, write_request):
    # node_a 收到写请求,但 node_b 因网络分区不可达
    if strategy == "CP":  # 选择一致性,放弃可用性
        return {"error": "service unavailable, refusing write"}, 503
    elif strategy == "AP":  # 选择可用性,放弃一致性
        node_a.accept_write_locally()  # 先写本地,返回 200
        node_b.will_sync_when_partition_heals()  # 分区恢复后再同步
        return {"status": "accepted, eventual consistency"}, 200

# Gilbert-Lynch 2002 的核心论证(简化版)
# 假设:分布式系统有两个节点 N1 / N2,网络可能分区
# 目标:同时满足 Consistency 和 Availability

class CAPProof:
    def __init__(self, n1, n2, network):
        self.n1, self.n2 = n1, n2
        self.network = network  # 可能分区

    def write(self, key, value):
        # 写 N1
        self.n1.set(key, value)
        # 如果网络已分区,这条消息无法到达 N2
        if self.network.is_partitioned():
            # 想继续服务(Availability)→ 必须返回,但 N2 看不到最新值
            # 想保证读到的都是最新值(Consistency)→ 必须拒绝服务
            raise InconsistencyOrUnavailabilityError(
                "Brewer 2000 / Gilbert-Lynch 2002: 无法两者兼得"
            )

# ZooKeeper (CP) 行为演示
from kazoo.client import KazooClient

zk = KazooClient(hosts="zk1:2181,zk2:2181,zk3:2181")
zk.start()
# 在 5 节点集群中,如果形成 2-3 分区:
# - 少数派(2 节点)侧:leader 退出,只读不写 → 体现 CP
# - 多数派(3 节点)侧:继续接受写,选举新 leader

# Redis Cluster (AP) 行为演示
import redis
rc = redis.RedisCluster(startup_nodes=[
    {"host": "redis-1", "port": 6379},
    {"host": "redis-2", "port": 6379},
])
# 主从分区时:主可能在 A 侧,从在 B 侧
# 客户端仍可写(主在线),从切换后那部分数据可能丢失
rc.set("counter:user_42", 1)  # 不会因为分区而拒绝

# 演示 PACELC 的「EL」维度重要性
# 即使分区没发生,你也得在延迟和一致性之间做日常取舍

class PACELCDemo:
    def __init__(self, mode):
        self.mode = mode  # "EL" or "EC"

    def write(self, key, value):
        if self.mode == "EL":  # Else: 选 Latency
            # 单 leader + 异步复制 follower,几百微秒返回
            self.local_leader.commit(key, value)
            self.replicate_async_to_followers(key, value)  # 后台慢慢同步
            return {"status": "committed locally", "latency_ms": 1}
        elif self.mode == "EC":  # Else: 选 Consistency
            # Quorum write:必须等多数节点 ack,几毫秒返回
            self.write_to_quorum(key, value, quorum=3)
            return {"status": "committed to quorum", "latency_ms": 10}

import time
import threading
import random
from typing import Dict, List

# 模拟一个 3 节点的最终一致 KV 存储
class EventuallyConsistentStore:
    def __init__(self, nodes: List[str]):
        self.nodes = {n: {} for n in nodes}
        self.locks = {n: threading.Lock() for n in nodes}
        self.vector_clock = {n: {} for n in nodes}

    def write(self, node: str, key: str, value):
        """本地写 + 异步广播"""
        with self.locks[node]:
            self.nodes[node][key] = value
            self.vector_clock[node][key] = self.vector_clock[node].get(key, 0) + 1
            my_version = self.vector_clock[node][key]

        # 异步同步(实际工程里用 gossip / 推拉复制)
        threading.Thread(target=self._gossip, args=(node, key, value, my_version), daemon=True).start()
        return my_version

    def _gossip(self, src: str, key: str, value, version: int):
        """模拟 gossip 传播,有延迟"""
        for other in self.nodes:
            if other == src:
                continue
            time.sleep(random.uniform(0.01, 0.1))  # 网络延迟
            with self.locks[other]:
                existing = self.vector_clock[other].get(key, 0)
                if version > existing:
                    self.nodes[other][key] = value
                    self.vector_clock[other][key] = version
                    print(f"  gossip {src}→{other}: {key}={value} (v{version})")

    def read(self, node: str, key: str):
        """本地读,可能读到旧值"""
        with self.locks[node]:
            return self.nodes[node].get(key), self.vector_clock[node].get(key, 0)


# 演示:写 1 次到 A,观察所有节点最终一致
if __name__ == "__main__":
    store = EventuallyConsistentStore(["A", "B", "C"])
    print("=== 初始状态 ===")
    for n in store.nodes:
        print(f"  {n}: {store.nodes[n]}")

    print("\n=== A 写入 key=balance → 200 ===")
    store.write("A", "balance", 200)
    time.sleep(0.5)  # 等 gossip 传播

    print("\n=== 最终状态 ===")
    for n in store.nodes:
        v, ver = store.read(n, "balance")
        print(f"  {n}: balance={v} (v{ver})")

# DynamoDB 的最终一致读 vs 强一致读
import boto3

ddb = boto3.client("dynamodb")

# 强一致读(ConsistentRead=True) — 走 quorum,贵但新
ddb.get_item(
    TableName="Users",
    Key={"UserId": {"S": "u_42"}},
    ConsistentRead=True   # 走 leader,延迟高
)

# 最终一致读(默认)
ddb.get_item(
    TableName="Users",
    Key={"UserId": {"S": "u_42"}},
    ConsistentRead=False  # 可能读到 stale,延迟低
)

class ConsistencyLevels:
    """演示不同一致性级别的行为差异"""

    def strong_consistent_write_and_read(self):
        """强一致:写完所有后续读都看得到"""
        # Raft / ZAB:写必须等多数 ack,读必须走 leader
        write("x", 1, wait_quorum=True)
        assert read("x") == 1  # 立即读到 1

    def eventual_consistent_read(self):
        """最终一致:写后立即读可能读不到"""
        write("x", 1, async_replication=True)
        # 本地立刻能读到(本地读)
        assert read_local("x") == 1
        # 远端节点可能还是 0(stale read)
        # 但经过异步复制后,远端也会变成 1

    def read_your_writes_demo(self):
        """读己之写:同一 session 立刻能读自己的写"""
        session = Session(user_id="u_42")
        session.write("cart_total", 999)
        # 同一 session 必须立刻读到 999,即使底层是最终一致
        assert session.read("cart_total") == 999

    def monotonic_read_demo(self):
        """单调读:T1 读到 5,T2 时刻读到的不应该比 5 更旧"""
        read_1 = read("user_score")  # 读到 5
        time.sleep(1)
        read_2 = read("user_score")  # 必须是 5 或更大,不能是 3
        assert read_2 >= read_1

# 简化数据流:下单链路
def place_order(user_id, sku, qty):
    # 第 1 步:订单服务(CP)
    order = order_db.create_order(user_id, sku, qty)  # 强一致

    # 第 2 步:异步扣库存(AP,允许短暂超卖)
    @retry(max_attempts=3)
    def deduct_inventory():
        stock = redis.decr(f"stock:{sku}", qty)
        if stock < 0:
            redis.incr(f"stock:{sku}", qty)  # 还回去
            raise OutOfStock()
        return stock

    # 第 3 步:调用支付(CP)
    pay_result = payment.charge(user_id, order.amount)
    if not pay_result.ok:
        order_db.cancel(order.id, reason="支付失败")

    # 第 4 步:异步发消息(MQ 事务消息保证最终一致)
    mq.send("order_created", order.id)

# Redis Cluster 客户端配置:决定副本读取策略
from redis.cluster import RedisCluster, ReadFrom

client = RedisCluster(
    startup_nodes=[{"host": "redis1", "port": 6379}, ...],
    # 默认从主读(强一致)→ 但性能差;可以从 replica 读(可能 stale)
    read_from_replicas=False  # 关键开关
)

# 一致性级别动态调整
def get_user_session(user_id):
    # 关键业务:从主读(慢但新)
    return client.execute_command("GET", f"session:{user_id}", target_nodes="master")

def get_recommendation_list(user_id):
    # 推荐:从副本读(快,允许 stale)
    return client.execute_command("GET", f"rec:{user_id}", target_nodes="replica")

# Nacos 客户端配置
from nacos import NacosClient

# 配置走 CP(强一致)
config_client = NacosClient(
    server_addresses="nacos1:8848",
    namespace="prod",
    # 默认走 AP 服务发现
)

# 关键配置变更(必须 CP):监听并强一致获取
def on_config_change(args):
    new_config = config_client.get_config(
        data_id="db.password",
        group="DEFAULT_GROUP"
    )
    # 默认就是 CP 模式,Raft 提交后所有客户端才看到
    apply_password_change(new_config)

# TiDB 通过 TiKV 存储层使用 Multi-Raft(每个 Region 一个 Raft Group)
# 不同 Region 的副本可以分散在多机房,实现异地多活
class TiDBTopology:
    def __init__(self):
        self.tikv_nodes = {
            "dc1": ["tikv1", "tikv2", "tikv3"],
            "dc2": ["tikv4", "tikv5", "tikv6"],
        }

    def write_with_quorum(self, key, value):
        # Region 的 5 个副本分布在 2 个机房
        # 写必须等多数(3 个),即「同机房为主 + 跨机房 1」
        # 这就是 EC(选一致性)牺牲了部分延迟
        leader = self.find_leader(key)
        return leader.commit(key, value, replicas=3)

class ConsistencyPolicy:
    PARTITION_STRATEGY = "AP"  # 明确写出来,不是默认行为
    ELSE_STRATEGY = "EL"       # PACELC 的 E 维度

# 误选 CP 的代价:Raft 的 quorum 写延迟,秒杀场景扛不住
# 误选 AP 的代价:账务对账不平,可能被监管处罚

# 订单幂等:依赖强一致 DB 的唯一约束
def create_order(user_id, sku, qty, idempotency_key):
    try:
        with db.transaction():
            # 唯一约束:同一个 idempotency_key 只能插入一次
            db.execute(
                "INSERT INTO orders (user_id, sku, qty, key) VALUES (?, ?, ?, ?)",
                user_id, sku, qty, idempotency_key
            )
    except UniqueConstraintViolation:
        return existing_order  # 幂等返回

# 错误示范:ZK 做注册中心
zk.register_service("order-service", "10.0.1.5:8080")
# 一旦 ZK minority 分区,所有服务发现全部失败

# 正确做法:Nacos 用 AP
nacos_client.register_instance(
    service_name="order-service",
    ip="10.0.1.5",
    port=8080,
    ephemeral=True  # AP 模式
)

# 错误:Redis SETNX 当分布式锁
def acquire_lock_redis(key, ttl=30):
    # 主从切换时,旧 master 上的锁还没同步,新 master 没这把锁
    return redis.set(key, "locked", nx=True, ex=ttl)

# 正确:etcd 强一致锁(Lease + Revision)
import etcd3
client = etcd3.client(host="etcd1", port=2379)

def acquire_lock_etcd(key, ttl=30):
    # Raft 强一致,Lease 自动续约,Revision 防止误删
    lease = client.lease(ttl)
    return client.put(key, b"locked", lease=lease)

# 错误:一致性"一刀切"
class WrongDesign:
    def order(self): return redis.set(...)    # 订单用 Redis → 错
    def counter(self): return mysql.insert(...)  # 点赞用 MySQL → 错

# 正确:分级
class RightDesign:
    def order(self): return mysql.insert(...)  # 订单强一致
    def counter(self): return redis.incr(...)  # 点赞最终一致

# etcd 批量写优化(ETCD 3.4+)
import etcd3
client = etcd3.client()

# 单写:每个 Key 都要一次 Raft 往返
for k, v in items:
    client.put(k, v)  # 50ms × 100 keys = 5s

# 批量:一次 Raft 提交多个 Key
with client.transaction() as txn:
    for k, v in items:
        txn.put(k, v)
# 50ms × 1 = 50ms(快 100 倍)