系统设计：高并发秒杀架构

一、秒杀系统挑战与设计目标

核心挑战

秒杀场景的特点是短时间内大量用户抢购少量商品，带来以下挑战：

1. 瞬时高并发：数十万甚至数百万用户同时请求
2. 超卖风险：库存控制不准确导致实际售出量超过库存
3. 恶意攻击：刷单、脚本抢购等恶意行为
4. 数据库压力：大量请求直接冲击数据库
5. 用户体验：页面加载缓慢、响应延迟

设计目标

• 高可用：99.99%可用性，支持故障降级
• 高性能：QPS 10万+，响应时间<100ms
• 一致性：不超卖、不少卖
• 可扩展：水平扩展支持更大规模

二、架构设计

整体架构

                        ┌─────────────┐
                        │   用户层    │
                        │ (Web/App)   │
                        └──────┬──────┘
                               │
                        ┌──────▼──────┐
                        │   CDN缓存    │
                        │ (静态资源)   │
                        └──────┬──────┘
                               │
                        ┌──────▼──────┐
                        │  负载均衡    │
                        │ (LVS/Nginx)  │
                        └──────┬──────┘
                               │
              ┌────────────────┼────────────────┐
              │                │                │
        ┌─────▼─────┐    ┌────▼────┐    ┌─────▼─────┐
        │  Web服务器  │    │ Web服务器 │    │  Web服务器  │
        │ (Nginx/Go) │    │ (Nginx/Go)│    │ (Nginx/Go) │
        └─────┬─────┘    └────┬────┘    └─────┬─────┘
              │               │                │
              └───────────────┼────────────────┘
                              │
                    ┌─────────▼─────────┐
                    │    API网关/Gateway │
                    │ (限流/熔断/路由)    │
                    └─────────┬─────────┘
                              │
              ┌───────────────┼───────────────┐
              │               │               │
        ┌─────▼─────┐   ┌────▼────┐   ┌─────▼─────┐
        │ 秒杀服务   │   │秒杀服务  │   │ 秒杀服务   │
        │ (Redis)   │   │ (Redis) │   │ (Redis)   │
        └─────┬─────┘   └────┬────┘   └─────┬─────┘
              │               │               │
              └───────────────┼───────────────┘
                              │
                    ┌─────────▼─────────┐
                    │   消息队列(RMQ)    │
                    │ (异步削峰/解耦)    │
                    └─────────┬─────────┘
                              │
              ┌───────────────┼───────────────┐
              │               │               │
        ┌─────▼─────┐   ┌────▼────┐   ┌─────▼─────┐
        │ 订单服务   │   │订单服务  │   │ 订单服务   │
        │ (MySQL)   │   │ (MySQL) │   │ (MySQL)   │
        └───────────┘   └─────────┘   └───────────┘
​

核心组件设计

1. Redis缓存层

import redis
import json
from threading import Lock

class SeckillRedis:
    def __init__(self):
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            db=0,
            decode_responses=True
        )
        self.local_lock = {}
    
    def init_stock(self, product_id, total_stock):
        """初始化库存到Redis"""
        key = f"stock:{product_id}"
        
        # 使用Lua脚本保证原子性
        lua_script = """
        local key = KEYS[1]
        local total = tonumber(ARGV[1])
        
        if redis.call('exists', key) == 1 then
            return -1  -- 已经初始化
        end
        
        redis.call('set', key, total)
        redis.call('set', key..':init', total)
        redis.call('set', key..':sold', 0)
        
        return 1
        """
        
        result = self.redis_client.eval(
            lua_script, 1,
            f"stock:{product_id}", total_stock
        )
        
        return result
    
    def deduct_stock(self, product_id, quantity=1):
        """扣减库存（原子操作）"""
        key = f"stock:{product_id}"
        
        # Lua脚本：检查库存并扣减
        lua_script = """
        local key = KEYS[1]
        local deduct = tonumber(ARGV[1])
        
        local current = tonumber(redis.call('get', key))
        if current == nil then
            return -1  -- 库存不存在
        end
        
        if current < deduct then
            return 0  -- 库存不足
        end
        
        redis.call('decrby', key, deduct)
        redis.call('incrby', key..':sold', deduct)
        
        return 1  -- 成功
        """
        
        result = self.redis_client.eval(
            lua_script, 1,
            f"stock:{product_id}", quantity
        )
        
        return result
    
    def get_stock_info(self, product_id):
        """获取库存信息"""
        key = f"stock:{product_id}"
        
        stock = self.redis_client.get(key)
        sold = self.redis_client.get(f"{key}:sold")
        
        return {
            "remaining": int(stock) if stock else 0,
            "sold": int(sold) if sold else 0
        }
    
    def add_to_blacklist(self, user_id, product_id, ttl=3600):
        """添加到抢购黑名单"""
        key = f"blacklist:{product_id}:{user_id}"
        self.redis_client.setex(key, ttl, "1")
    
    def is_in_blacklist(self, user_id, product_id):
        """检查是否在黑名单"""
        key = f"blacklist:{product_id}:{user_id}"
        return self.redis_client.exists(key) > 0

# Redis集群配置（生产环境）
def get_redis_cluster():
    from rediscluster import RedisCluster
    
    startup_nodes = [
        {"host": "192.168.1.10", "port": 7000},
        {"host": "192.168.1.11", "port": 7001},
        {"host": "192.168.1.12", "port": 7002},
    ]
    
    return RedisCluster(
        startup_nodes=startup_nodes,
        decode_responses=True,
        skip_full_coverage_check=True,
        max_connections=100
    )
​

2. 限流保护

import time
from collections import deque
from functools import wraps

class RateLimiter:
    """限流器：滑动窗口算法"""
    
    def __init__(self, max_requests, time_window):
        self.max_requests = max_requests
        self.time_window = time_window  # 秒
        self.requests = deque()
        self.lock = Lock()
    
    def is_allowed(self):
        with self.lock:
            now = time.time()
            
            # 清理过期请求
            while self.requests and self.requests[0] < now - self.time_window:
                self.requests.popleft()
            
            # 检查是否超过限制
            if len(self.requests) >= self.max_requests:
                return False
            
            self.requests.append(now)
            return True

# 令牌桶算法限流
class TokenBucketLimiter:
    def __init__(self, capacity, rate):
        """
        capacity: 桶容量
        rate: 令牌生成速率（令牌/秒）
        """
        self.capacity = capacity
        self.rate = rate
        self.tokens = capacity
        self.last_time = time.time()
        self.lock = Lock()
    
    def is_allowed(self, tokens=1):
        with self.lock:
            now = time.time()
            
            # 计算新增令牌
            new_tokens = (now - self.last_time) * self.rate
            self.tokens = min(self.capacity, self.tokens + new_tokens)
            self.last_time = now
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            
            return False

# 分布式限流（基于Redis）
class DistributedRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def is_allowed(self, key, limit, window=60):
        """
        key: 限流标识（如 user_id、ip等）
        limit: 时间窗口内最大请求数
        window: 时间窗口（秒）
        """
        current_time = int(time.time())
        window_start = current_time - window
        
        # 使用有序集合存储请求时间戳
        pipe = self.redis.pipeline()
        
        # 移除窗口外的记录
        pipe.zremrangebyscore(key, 0, window_start)
        
        # 获取当前请求数
        pipe.zcard(key)
        
        # 添加当前请求
        pipe.zadd(key, {str(current_time): current_time})
        
        # 设置过期时间
        pipe.expire(key, window + 1)
        
        results = pipe.execute()
        request_count = results[1]
        
        return request_count < limit

# 使用示例
limiter = DistributedRateLimiter(redis.Redis())

def rate_limit_check(user_id):
    """用户级限流：每秒最多10次请求"""
    return limiter.is_allowed(f"rate:user:{user_id}", 10, 1)

def ip_rate_limit_check(ip):
    """IP级限流：每分钟最多100次请求"""
    return limiter.is_allowed(f"rate:ip:{ip}", 100, 60)
​

3. 订单服务

import uuid
from datetime import datetime

class OrderService:
    def __init__(self, db_connection, redis_client, message_queue):
        self.db = db_connection
        self.redis = redis_client
        self.mq = message_queue
    
    def create_order(self, user_id, product_id, quantity=1):
        """创建订单（异步）"""
        
        # 1. 检查用户是否已购买
        if self._has_purchased(user_id, product_id):
            return {
                "success": False,
                "message": "每人限购一件"
            }
        
        # 2. 检查黑名单
        if self._is_in_blacklist(user_id, product_id):
            return {
                "success": False,
                "message": "操作过于频繁"
            }
        
        # 3. Redis扣减库存
        seckill_redis = SeckillRedis()
        stock_result = seckill_redis.deduct_stock(product_id, quantity)
        
        if stock_result == -1:
            return {
                "success": False,
                "message": "商品不存在"
            }
        
        if stock_result == 0:
            return {
                "success": False,
                "message": "库存不足"
            }
        
        # 4. 创建订单记录（内存中）
        order_id = str(uuid.uuid4())
        order_data = {
            "order_id": order_id,
            "user_id": user_id,
            "product_id": product_id,
            "quantity": quantity,
            "status": "pending",
            "create_time": datetime.now().isoformat()
        }
        
        # 5. 发送到消息队列异步处理
        self.mq.publish(
            "order_created",
            json.dumps(order_data)
        )
        
        # 6. 缓存用户购买记录
        self._mark_purchased(user_id, product_id)
        
        return {
            "success": True,
            "order_id": order_id,
            "message": "抢购成功，订单处理中"
        }
    
    def _has_purchased(self, user_id, product_id):
        """检查是否已购买"""
        key = f"purchased:{product_id}:{user_id}"
        return self.redis.exists(key) > 0
    
    def _mark_purchased(self, user_id, product_id):
        """标记已购买"""
        key = f"purchased:{product_id}:{user_id}"
        self.redis.setex(key, 3600, "1")
    
    def _is_in_blacklist(self, user_id, product_id):
        """检查黑名单"""
        key = f"blacklist:{product_id}:{user_id}"
        return self.redis.exists(key) > 0

# 消息队列消费者（异步处理订单）
class OrderProcessor:
    def __init__(self, db_connection):
        self.db = db_connection
    
    def process_order(self, order_data):
        """处理订单（消费者）"""
        try:
            # 1. 插入订单到数据库
            self.db.insert(
                "INSERT INTO orders "
                "(order_id, user_id, product_id, quantity, status, create_time) "
                "VALUES (%s, %s, %s, %s, 'pending', %s)",
                order_data["order_id"],
                order_data["user_id"],
                order_data["product_id"],
                order_data["quantity"],
                order_data["create_time"]
            )
            
            # 2. 扣减数据库库存（用于对账）
            self.db.execute(
                "UPDATE products SET stock = stock - %s "
                "WHERE product_id = %s AND stock >= %s",
                order_data["quantity"],
                order_data["product_id"],
                order_data["quantity"]
            )
            
            # 3. 更新订单状态
            self.db.execute(
                "UPDATE orders SET status = 'completed' "
                "WHERE order_id = %s",
                order_data["order_id"]
            )
            
            print(f"订单 {order_data['order_id']} 处理成功")
            
        except Exception as e:
            print(f"订单处理失败: {e}")
            # 记录失败日志，后续人工处理或重试
            self.log_failed_order(order_data, e)
    
    def log_failed_order(self, order_data, error):
        """记录失败订单"""
        self.db.insert(
            "INSERT INTO failed_orders "
            "(order_id, user_id, product_id, error, create_time) "
            "VALUES (%s, %s, %s, %s, NOW())",
            order_data["order_id"],
            order_data["user_id"],
            order_data["product_id"],
            str(error)
        )
​

三、性能优化

1. 页面静态化

# 秒杀活动页面静态化
class SeckillPageCache:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def generate_static_page(self, activity_id):
        """生成静态页面"""
        # 查询活动信息
        activity = get_activity_info(activity_id)
        
        # 生成HTML
        html = f"""
        <!DOCTYPE html>
        <html>
        <head>
            <title>{activity['title']}</title>
            <meta name="viewport" content="width=device-width, initial-scale=1">
        </head>
        <body>
            <div class="product-info">
                <h1>{activity['product_name']}</h1>
                <p class="price">¥{activity['price']}</p>
                <p class="stock">库存：<span id="stock">{activity['stock']}</span></p>
                <p class="time">开始时间：{activity['start_time']}</p>
            </div>
            <button onclick="seckill()" id="btn">立即抢购</button>
            <script src="/static/js/seckill.js"></script>
        </body>
        </html>
        """
        
        # 缓存到Redis
        self.redis.setex(
            f"page:activity:{activity_id}",
            3600,
            html
        )
        
        return html
    
    def get_cached_page(self, activity_id):
        """获取缓存的页面"""
        return self.redis.get(f"page:activity:{activity_id}")
​

2. 动态库存展示

# WebSocket推送库存变化
class StockNotifier:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.connections = set()
    
    def subscribe_stock_updates(self, product_id, websocket):
        """订阅库存更新"""
        self.connections.add(websocket)
        
        # 订阅Redis发布订阅
        pubsub = self.redis.pubsub()
        pubsub.subscribe(f"stock_update:{product_id}")
        
        for message in pubsub.listen():
            if message['type'] == 'message':
                # 推送到前端
                websocket.send(message['data'])
    
    def publish_stock_update(self, product_id, remaining):
        """发布库存更新"""
        self.redis.publish(
            f"stock_update:{product_id}",
            json.dumps({"product_id": product_id, "stock": remaining})
        )
​

四、安全防护

1. 防刷机制

class AntiSpam:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def check_user_behavior(self, user_id, ip):
        """检查用户行为"""
        # 1. 检查IP频率
        ip_key = f"spam:ip:{ip}"
        ip_count = self.redis.incr(ip_key)
        self.redis.expire(ip_key, 60)
        
        if ip_count > 100:  # 每分钟超过100次
            return False, "操作过于频繁"
        
        # 2. 检查用户频率
        user_key = f"spam:user:{user_id}"
        user_count = self.redis.incr(user_key)
        self.redis.expire(user_key, 60)
        
        if user_count > 50:  # 每分钟超过50次
            return False, "操作过于频繁"
        
        # 3. 检查User-Agent
        # (省略User-Agent检测逻辑)
        
        return True, "OK"
    
    def add_to_blacklist(self, user_id, reason, ttl=3600):
        """添加到黑名单"""
        key = f"blacklist:user:{user_id}"
        self.redis.setex(key, ttl, reason)
    
    def is_blacklisted(self, user_id):
        """检查是否在黑名单"""
        key = f"blacklist:user:{user_id}"
        return self.redis.exists(key) > 0
​

2. 验证码机制

import random
import string

class CaptchaService:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def generate_captcha(self, session_id):
        """生成验证码"""
        # 生成4位数字验证码
        captcha = ''.join(random.choices(string.digits, k=4))
        
        # 存储到Redis（5分钟有效）
        key = f"captcha:{session_id}"
        self.redis.setex(key, 300, captcha)
        
        return captcha
    
    def verify_captcha(self, session_id, user_input):
        """验证验证码"""
        key = f"captcha:{session_id}"
        stored_captcha = self.redis.get(key)
        
        if not stored_captcha:
            return False, "验证码已过期"
        
        if stored_captcha.lower() != user_input.lower():
            return False, "验证码错误"
        
        # 验证成功后删除
        self.redis.delete(key)
        return True, "OK"
​

五、监控告警

1. 实时监控

from prometheus_client import Counter, Gauge, start_http_server

# 定义监控指标
request_counter = Counter('seckill_requests_total', 'Total requests', ['status'])
stock_gauge = Gauge('seckill_stock_remaining', 'Remaining stock', ['product_id'])
order_counter = Counter('seckill_orders_total', 'Total orders', ['status'])

# 监控装饰器
def monitor_request(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            if result.get("success"):
                request_counter.labels(status='success').inc()
            else:
                request_counter.labels(status='failed').inc()
            return result
        except Exception as e:
            request_counter.labels(status='error').inc()
            raise e
    return wrapper

# 库存监控
def monitor_stock(product_id, stock):
    stock_gauge.labels(product_id=product_id).set(stock)
​

2. 告警规则

# Prometheus告警规则
groups:
- name: seckill_alerts
  rules:
  - alert: HighErrorRate
    expr: rate(seckill_requests_total{status="error"}[5m]) > 0.1
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "秒杀系统错误率过高"
      description: "5分钟内错误率超过10%"
  
  - alert: LowStock
    expr: seckill_stock_remaining < 100
    labels:
      severity: warning
    annotations:
      summary: "库存告急"
      description: "商品 {{ $labels.product_id }} 库存不足100"
​

六、最佳实践

1. 分层限流：用户级、IP级、系统级多层限流
2. 异步处理：订单创建异步化，快速响应用户
3. 缓存优先：所有可缓存的数据都放入Redis
4. 降级预案：准备多个降级方案，保证核心功能可用
5. 压测验证：上线前进行充分压力测试
6. 灰度发布：逐步放开流量，观察系统表现
7. 数据对账：定期核对Redis和MySQL数据一致性

秒杀系统是高并发架构的经典场景，通过合理的设计和优化，可以在保证用户体验的同时，确保系统的稳定性和数据一致性。