京东秒杀系统崩溃？三招Redis缓存策略带你扛住百万并发（附python源码）

一、 秒杀崩溃的根源与架构总览

为什么传统架构会崩？

• 数据库瓶颈：瞬间百万QPS直接打满数据库连接池，导致服务雪崩。

• 超卖现象：单纯的 if stock > 0: stock -= 1在并发下会扣成负数。

• 库存热点：单行数据（如 product_id=666）成为热点，所有请求都在竞争同一把锁。

抗崩架构核心：分层过滤

graph TD
    A[100万并发请求] --> B[网关层: 限流/风控]
    B --> C[秒杀服务: Redis预扣库存]
    C --> D[消息队列: 削峰填谷]
    D --> E[数据库: 最终落库]

二、 扛住百万并发的三招Redis策略

第一招：原子防超卖（Lua脚本）

import redis

# 连接Redis（生产环境用连接池）
redis_client = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 初始化库存（秒杀开始前执行）
redis_client.set("seckill:stock:1001", 1000)

# 定义Lua脚本（核心：判断+扣减在Redis端原子执行）
SECKILL_SCRIPT = """
local stock_key = KEYS[1]
local stock = tonumber(redis.call('GET', stock_key))
if stock and stock > 0 then
    redis.call('DECR', stock_key)
    return 1 -- 成功
else
    return 0 -- 失败
end
"""
seckill_script = redis_client.register_script(SECKILL_SCRIPT)
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
def handle_seckill(user_id, product_id):
    """处理秒杀请求（核心逻辑）"""
    stock_key = f"seckill:stock:{product_id}"
    
    # 执行Lua脚本（原子操作）
    result = seckill_script(keys=[stock_key])
    
    if result == 1:
        # 秒杀成功，发送消息到MQ，异步创建订单
        # send_to_mq(user_id, product_id)
        return {"code": 200, "msg": "抢购成功"}
    else:
        return {"code": 400, "msg": "已售罄"}

第二招：库存分片（解决热点Key）

def get_shard_key(product_id, user_id, shard_count=10):
    """根据用户ID哈希取模，路由到不同的库存分片"""
    shard_index = hash(user_id) % shard_count
    return f"seckill:stock:{product_id}:shard_{shard_index}"

def init_shard_stock(product_id, total_stock, shard_count=10):
    """初始化分片库存（总库存1000，分10片，每片100）"""
    base_stock = total_stock // shard_count
    for i in range(shard_count):
        key = f"seckill:stock:{product_id}:shard_{i}"
        redis_client.set(key, base_stock)

def sharded_seckill(user_id, product_id):
    """分片秒杀逻辑"""
    shard_key = get_shard_key(product_id, user_id)
    
    # 使用同样的Lua脚本，但操作的是分片Key
    result = seckill_script(keys=[shard_key])
    
    if result == 1:
        return {"code": 200, "msg": "抢购成功"}
    else:
        # 该分片没了，可以尝试其他分片或直接返回售罄
        return {"code": 400, "msg": "已售罄"}

第三招：令牌桶限流（保护Redis）

import time

def token_bucket_limiter(user_id, max_requests=100, refill_rate=10):
    """令牌桶限流（Redis实现）"""
    key = f"rate_limit:{user_id}"
    now = time.time()
    
    # 使用Pipeline减少网络往返
    pipe = redis_client.pipeline()
    pipe.hgetall(key)
    pipe.hset(key, 'last_time', now)
    pipe.expire(key, 60)
    data, _, _ = pipe.execute()
    
    if not data:
        # 第一次请求，初始化
        tokens = max_requests - 1
        redis_client.hset(key, 'tokens', tokens)
        return True
    
    last_time = float(data.get('last_time', now))
    tokens = float(data.get('tokens', max_requests))
    
    # 计算这段时间应该补充的令牌
    elapsed = now - last_time
    tokens = min(max_requests, tokens + elapsed * refill_rate)
    
    if tokens >= 1:
        tokens -= 1
        redis_client.hset(key, 'tokens', tokens)
        return True  # 放行
    else:
        return False  # 限流

# 在秒杀入口添加限流
def seckill_entry(user_id, product_id):
    if not token_bucket_limiter(user_id):
        return {"code": 429, "msg": "手速太快，请稍后再试"}
    return handle_seckill(user_id, product_id)

三、 完整秒杀流程与压测建议

1. 完整秒杀时序图

sequenceDiagram
    participant U as 用户
    participant G as 网关(限流)
    participant A as 秒杀服务(Python)
    participant R as Redis(Lua)
    participant M as 消息队列
    participant D as 数据库

    U->>G: 点击秒杀
    G->>A: 放行(令牌桶)
    A->>R: 执行Lua脚本(原子扣减)
    R->>A: 成功/失败
    A->>M: 发送成功消息(异步)
    A->>U: 返回“抢购成功”
    M->>D: 消费消息，创建订单

2. 压测与部署建议

• 压测工具：使用 locust或 wrk模拟百万并发。

• Redis配置：必须开启持久化（AOF），防止重启丢数据。

• 连接池：Python端务必使用 redis.ConnectionPool，避免频繁创建连接。

• 监控：实时监控Redis的 QPS和 内存，设置库存告警。

四、 避坑指南与总结

⚠️ 三大坑点

1. 不要用事务：Redis事务（MULTI）不是原子回滚，高并发下性能差，必须用Lua。

2. 不要先查后写：任何“先查数据库再更新”的逻辑都会超卖。

3. 不要忘记预热：秒杀开始前，必须把库存加载到Redis，不能临时查DB。

✅ 三招总结