分布式系统设计实战指南：从CAP到高可用

# 四层负载均衡（LVS）
upstream backend {
    server 10.0.0.1:8080;
    server 10.0.0.2:8080;
    server 10.0.0.3:8080 backup;  # 备用
}

# 七层负载均衡（Nginx）
upstream backend {
    least_conn;
    server 10.0.0.1 max_fails=3 fail_timeout=30s;
    server 10.0.0.2 max_fails=3 fail_timeout=30s;
}

# 令牌桶限流
import time
from collections import deque

class TokenBucket:
    def __init__(self, capacity, rate):
        self.capacity = capacity
        self.rate = rate  # tokens per second
        self.tokens = capacity
        self.last_time = time.time()

    def allow(self):
        now = time.time()
        elapsed = now - self.last_time
        self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
        self.last_time = now

        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

bucket = TokenBucket(capacity=100, rate=10)  # 每秒 10 个请求

def handle_request(request):
    if not bucket.allow():
        return {'error': 'Rate limit exceeded'}, 429
    # 正常处理

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=60):
        self.failure_count = 0
        self.failure_threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.state = 'closed'  # closed, open, half_open
        self.last_failure_time = 0

    def call(self, func, *args, **kwargs):
        if self.state == 'open':
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = 'half_open'
            else:
                raise Exception('Circuit breaker is open')

        try:
            result = func(*args, **kwargs)
            if self.state == 'half_open':
                self.state = 'closed'
                self.failure_count = 0
            return result
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            if self.failure_count >= self.failure_threshold:
                self.state = 'open'
            raise

BEGIN TRANSACTION;
UPDATE inventory SET count = count - 1 WHERE product_id = 1;
INSERT INTO orders (user_id, product_id) VALUES (100, 1);
COMMIT;

# Try：预留资源
def try_create_order(order_data):
    # 冻结库存
    inventory.freeze(order_data.items)
    # 预创建订单（待确认状态）
    order = Order.create_pending(order_data)
    return order

# Confirm：确认执行
def confirm_order(order_id):
    # 扣减冻结的库存
    inventory.confirm(order_id)
    # 订单状态改为已确认
    Order.confirm(order_id)

# Cancel：取消
def cancel_order(order_id):
    # 解冻库存
    inventory.unfreeze(order_id)
    # 订单状态改为已取消
    Order.cancel(order_id)

# 订单创建 Saga
def create_order_saga(order_data):
    try:
        # 1. 创建订单
        order = create_order(order_data)

        # 2. 扣库存
        deduct_inventory(order_data.items)

        # 3. 扣余额
        deduct_balance(order_data.user_id, order.total)

        # 4. 发通知
        send_notification(order_data.user_id)

        return order

    except InventoryError:
        # 补偿：取消订单
        cancel_order(order.id)
    except BalanceError:
        # 补偿：恢复库存、取消订单
        restore_inventory(order_data.items)
        cancel_order(order.id)

def create_order_with_message(user_id, items):
    # 本地事务：创建订单 + 写消息表
    with db.transaction():
        order = Order.create(user_id, items)
        Message.create(
            topic='order_created',
            payload={'order_id': order.id}
        )

    # 异步发送消息到 MQ（独立服务）
    return order

# 消息中继服务
def relay_messages():
    while True:
        messages = Message.get_pending()
        for msg in messages:
            try:
                mq.send(msg.topic, msg.payload)
                msg.mark_sent()
            except:
                # 失败重试
                msg.retry_count += 1

from kafka import KafkaProducer, KafkaConsumer

# 生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

producer.send('orders', {'order_id': 123, 'amount': 100})
producer.flush()

# 消费者
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='order_processor',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

for message in consumer:
    process_order(message.value)

# 生产端：开启确认
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    acks='all'  # 所有副本确认
)

# 消费端：手动提交 offset
consumer = KafkaConsumer(
    'orders',
    enable_auto_commit=False
)

for message in consumer:
    try:
        process_order(message.value)
        consumer.commit()  # 处理完再提交
    except:
        # 不提交，下次重试
        pass

def process_order(order_data):
    # 幂等检查
    if Order.exists(order_data['order_id']):
        return  # 已处理，跳过

    Order.create(order_data)

def get_user(user_id):
    # 先查缓存
    user = cache.get(f'user:{user_id}')
    if user:
        return user

    # 查数据库
    user = db.query('SELECT * FROM users WHERE id = %s', user_id)

    # 写缓存
    cache.set(f'user:{user_id}', user, ttl=3600)
    return user

def update_user(user_id, data):
    # 更新数据库
    db.update('users', data, id=user_id)
    # 删除缓存（不是更新）
    cache.delete(f'user:{user_id}')

def get_user(user_id):
    user = cache.get(f'user:{user_id}')
    if user:
        return user if user != 'NULL' else None

    user = db.query('SELECT ...', user_id)
    if user:
        cache.set(f'user:{user_id}', user, ttl=3600)
    else:
        # 缓存空值，防止穿透
        cache.set(f'user:{user_id}', 'NULL', ttl=60)
    return user

# 用互斥锁
def get_hot_data(key):
    data = cache.get(key)
    if data:
        return data

    # 加锁
    with redis.lock(f'lock:{key}', timeout=10):
        # 双重检查
        data = cache.get(key)
        if data:
            return data

        data = db.query(key)
        cache.set(key, data, ttl=3600)
        return data

# 随机过期时间
def cache_with_random_ttl(key, value, base_ttl=3600):
    ttl = base_ttl + random.randint(0, 600)  # 加 0-10 分钟随机
    cache.set(key, value, ttl=ttl)

import redis
import uuid

r = redis.Redis()

def acquire_lock(name, expire=10):
    """获取锁"""
    token = str(uuid.uuid4())
    if r.set(name, token, nx=True, ex=expire):
        return token
    return None

def release_lock(name, token):
    """释放锁（Lua 保证原子性）"""
    script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    return r.eval(script, 1, name, token)

# 使用
token = acquire_lock('process_order:123')
if token:
    try:
        # 业务逻辑
        pass
    finally:
        release_lock('process_order:123', token)

# Consul 服务注册
{
    "service": {
        "name": "user-service",
        "address": "10.0.0.1",
        "port": 8080,
        "check": {
            "http": "http://10.0.0.1:8080/health",
            "interval": "10s"
        }
    }
}

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("process_order")
def process_order(order_id):
    with tracer.start_as_current_span("query_db"):
        order = db.get_order(order_id)

    with tracer.start_as_current_span("call_payment"):
        payment.charge(order)

    return order