2025-01-15 約 1000 字預計閱讀 5 分鐘

08-2. 秒殺系統設計

📌 什麼是秒殺系統？

秒殺系統是電商平台在特定時間段內，以極低價格限量銷售商品的場景。

典型特徵

商品：iPhone 15 Pro Max 256GB
原價：NT$ 42,900
秒殺價：NT$ 9,999
數量：10 台
時間：2025-01-20 12:00:00

預期參與人數：100萬+

⚡ 秒殺系統的挑戰

1. 瞬間流量暴增

正常流量：1,000 QPS（每秒請求數）
秒殺流量：100,000 QPS（100倍）
持續時間：幾秒到幾分鐘

2. 讀寫比例失衡

讀操作（查詢庫存）：99.99%
寫操作（實際購買）：0.01%

10萬人搶10台手機，只有10個人能成功。

3. 惡意請求

機器人刷單
腳本自動下單
惡意佔用資源

🏗️ 系統架構設計

整體架構圖

                    ┌─────────────┐
                    │   用戶端    │
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │   CDN +     │
                    │   靜態資源  │
                    └──────┬──────┘
                           │
                    ┌──────▼──────┐
                    │  Nginx      │
                    │  限流 + 防攻擊│
                    └──────┬──────┘
                           │
         ┌─────────────────┼─────────────────┐
         │                 │                 │
    ┌────▼────┐      ┌────▼────┐      ┌────▼────┐
    │ Django  │      │ Django  │      │ Django  │
    │ Server  │      │ Server  │      │ Server  │
    └────┬────┘      └────┬────┘      └────┬────┘
         │                 │                 │
         └─────────────────┼─────────────────┘
                           │
         ┌─────────────────┼─────────────────┐
         │                 │                 │
    ┌────▼────┐      ┌────▼────┐      ┌────▼────┐
    │  Redis  │      │ Message │      │  MySQL  │
    │  緩存   │      │  Queue  │      │  數據庫 │
    └─────────┘      └─────────┘      └─────────┘

🎯 核心設計思路

1. 動靜分離

靜態內容（商品詳情頁）：

使用 CDN 加速
不經過後端服務器
承載 99% 的流量

動態內容（秒殺接口）：

只在秒殺開始時才打開
需要經過後端驗證

2. 限流

在各個層級進行限流：

CDN：地理位置限流
Nginx：IP 限流、總流量限流
應用層：用戶維度限流、接口限流

3. 削峰填谷

使用消息隊列將瞬時流量削峰：

秒殺請求 → 快速響應「排隊中」
         ↓
      消息隊列
         ↓
    異步處理訂單
         ↓
     通知用戶結果

💻 Django 實現

步驟 1：數據模型設計

from django.db import models
from django.contrib.auth.models import User

class FlashSaleProduct(models.Model):
    """秒殺商品"""
    name = models.CharField(max_length=200)
    original_price = models.DecimalField(max_digits=10, decimal_places=2)
    flash_price = models.DecimalField(max_digits=10, decimal_places=2)
    stock = models.IntegerField(default=0)  # 數據庫庫存（冷備份）

    # 秒殺時間
    start_time = models.DateTimeField()
    end_time = models.DateTimeField()

    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = 'flash_sale_product'


class FlashSaleOrder(models.Model):
    """秒殺訂單"""
    STATUS_CHOICES = [
        ('pending', '待處理'),
        ('success', '成功'),
        ('failed', '失敗'),
    ]

    order_no = models.CharField(max_length=32, unique=True)
    product = models.ForeignKey(FlashSaleProduct, on_delete=models.CASCADE)
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending')

    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        db_table = 'flash_sale_order'
        # 防止重複購買
        unique_together = ['product', 'user']

步驟 2：Redis 緩存層

import redis
from django.conf import settings

class FlashSaleCache:
    """秒殺緩存管理"""

    def __init__(self):
        self.redis_client = redis.StrictRedis(
            host=settings.REDIS_HOST,
            port=settings.REDIS_PORT,
            db=settings.REDIS_DB,
            decode_responses=True
        )

    def init_stock(self, product_id, stock):
        """初始化庫存到 Redis"""
        key = f'flash_sale:stock:{product_id}'
        self.redis_client.set(key, stock)

    def get_stock(self, product_id):
        """獲取當前庫存"""
        key = f'flash_sale:stock:{product_id}'
        stock = self.redis_client.get(key)
        return int(stock) if stock else 0

    def deduct_stock(self, product_id, quantity=1):
        """
        扣減庫存（原子操作）
        返回扣減後的庫存，如果庫存不足返回 -1
        """
        key = f'flash_sale:stock:{product_id}'

        # 使用 Lua 腳本確保原子性
        lua_script = """
        local stock = redis.call('GET', KEYS[1])
        if not stock then
            return -1
        end
        stock = tonumber(stock)
        if stock >= tonumber(ARGV[1]) then
            return redis.call('DECRBY', KEYS[1], ARGV[1])
        else
            return -1
        end
        """

        result = self.redis_client.eval(lua_script, 1, key, quantity)
        return result

    def check_user_purchased(self, product_id, user_id):
        """檢查用戶是否已經購買過"""
        key = f'flash_sale:purchased:{product_id}'
        return self.redis_client.sismember(key, user_id)

    def mark_user_purchased(self, product_id, user_id):
        """標記用戶已購買"""
        key = f'flash_sale:purchased:{product_id}'
        self.redis_client.sadd(key, user_id)

步驟 3：限流裝飾器

import time
from functools import wraps
from django.core.cache import cache
from django.http import JsonResponse

def rate_limit(key_prefix, max_requests, time_window):
    """
    限流裝飾器
    :param key_prefix: 緩存鍵前綴
    :param max_requests: 時間窗口內的最大請求數
    :param time_window: 時間窗口（秒）
    """
    def decorator(func):
        @wraps(func)
        def wrapper(request, *args, **kwargs):
            # 生成限流鍵（基於用戶 ID 或 IP）
            user_id = request.user.id if request.user.is_authenticated else None
            ip = request.META.get('REMOTE_ADDR')
            cache_key = f'{key_prefix}:{user_id or ip}'

            # 獲取當前請求次數
            current_requests = cache.get(cache_key, 0)

            if current_requests >= max_requests:
                return JsonResponse({
                    'status': 'error',
                    'message': '請求過於頻繁，請稍後再試'
                }, status=429)

            # 增加計數
            cache.set(cache_key, current_requests + 1, time_window)

            return func(request, *args, **kwargs)

        return wrapper
    return decorator

步驟 4：秒殺接口

import uuid
from django.utils import timezone
from django.views.decorators.http import require_POST
from django.contrib.auth.decorators import login_required

flash_cache = FlashSaleCache()

@require_POST
@login_required
@rate_limit(key_prefix='flash_sale', max_requests=3, time_window=10)
def flash_sale_purchase(request, product_id):
    """
    秒殺購買接口
    """
    user = request.user

    # 1. 檢查秒殺時間
    try:
        product = FlashSaleProduct.objects.get(id=product_id)
    except FlashSaleProduct.DoesNotExist:
        return JsonResponse({'status': 'error', 'message': '商品不存在'})

    now = timezone.now()
    if now < product.start_time:
        return JsonResponse({'status': 'error', 'message': '秒殺尚未開始'})
    if now > product.end_time:
        return JsonResponse({'status': 'error', 'message': '秒殺已結束'})

    # 2. 檢查用戶是否已經購買過（Redis）
    if flash_cache.check_user_purchased(product_id, user.id):
        return JsonResponse({'status': 'error', 'message': '您已經購買過了'})

    # 3. 扣減 Redis 庫存（原子操作）
    remaining_stock = flash_cache.deduct_stock(product_id, quantity=1)

    if remaining_stock < 0:
        return JsonResponse({'status': 'error', 'message': '庫存不足'})

    # 4. 標記用戶已購買
    flash_cache.mark_user_purchased(product_id, user.id)

    # 5. 發送消息到隊列，異步創建訂單
    order_no = f'FS{int(time.time())}{uuid.uuid4().hex[:8].upper()}'

    # 這裡可以使用 Celery 或其他消息隊列
    from .tasks import create_flash_sale_order
    create_flash_sale_order.delay(
        order_no=order_no,
        product_id=product_id,
        user_id=user.id
    )

    # 6. 立即返回成功（異步處理）
    return JsonResponse({
        'status': 'success',
        'message': '搶購成功，訂單處理中',
        'order_no': order_no
    })

步驟 5：異步訂單處理（Celery）

from celery import shared_task
from django.db import transaction
from django.db.models import F

@shared_task
def create_flash_sale_order(order_no, product_id, user_id):
    """
    異步創建秒殺訂單
    """
    try:
        with transaction.atomic():
            # 1. 扣減數據庫庫存（使用 F() 表達式）
            affected_rows = FlashSaleProduct.objects.filter(
                id=product_id,
                stock__gt=0
            ).update(stock=F('stock') - 1)

            if affected_rows == 0:
                # 數據庫庫存不足（理論上不應該發生）
                FlashSaleOrder.objects.create(
                    order_no=order_no,
                    product_id=product_id,
                    user_id=user_id,
                    status='failed'
                )
                return

            # 2. 創建訂單
            order = FlashSaleOrder.objects.create(
                order_no=order_no,
                product_id=product_id,
                user_id=user_id,
                status='success'
            )

            # 3. 其他業務邏輯（扣款、通知等）
            # ...

            return order.id

    except Exception as e:
        # 記錄錯誤日誌
        import logging
        logger = logging.getLogger(__name__)
        logger.error(f'Create flash sale order failed: {e}')

        # 標記訂單失敗
        FlashSaleOrder.objects.create(
            order_no=order_no,
            product_id=product_id,
            user_id=user_id,
            status='failed'
        )

🚀 性能優化技巧

1. 預熱緩存

在秒殺開始前，將庫存數據預熱到 Redis：

from django.core.management.base import BaseCommand

class Command(BaseCommand):
    help = '預熱秒殺商品庫存到 Redis'

    def handle(self, *args, **options):
        flash_cache = FlashSaleCache()

        products = FlashSaleProduct.objects.filter(
            start_time__gt=timezone.now()
        )

        for product in products:
            flash_cache.init_stock(product.id, product.stock)
            self.stdout.write(f'商品 {product.name} 庫存已預熱')

2. 頁面靜態化

將商品詳情頁完全靜態化，部署到 CDN：

<!-- 靜態頁面，直接放到 CDN -->
<!DOCTYPE html>
<html>
<head>
    <title>iPhone 15 Pro Max 秒殺</title>
</head>
<body>
    <h1>iPhone 15 Pro Max 256GB</h1>
    <p>原價：<del>NT$ 42,900</del></p>
    <p>秒殺價：<strong>NT$ 9,999</strong></p>

    <button id="flash-sale-btn">立即搶購</button>

    <script>
        // 秒殺按鈕只在指定時間才激活
        const startTime = new Date('2025-01-20T12:00:00+08:00');

        document.getElementById('flash-sale-btn').addEventListener('click', () => {
            if (Date.now() < startTime) {
                alert('秒殺尚未開始');
                return;
            }

            // 發送 AJAX 請求到後端秒殺接口
            fetch('/api/flash-sale/123/', {
                method: 'POST',
                headers: {
                    'X-CSRFToken': getCookie('csrftoken')
                }
            })
            .then(response => response.json())
            .then(data => {
                alert(data.message);
            });
        });
    </script>
</body>
</html>

3. Nginx 限流配置

# nginx.conf

# 限制每個 IP 每秒只能發起 10 個請求
limit_req_zone $binary_remote_addr zone=flash_sale:10m rate=10r/s;

server {
    listen 80;
    server_name example.com;

    location /api/flash-sale/ {
        limit_req zone=flash_sale burst=20 nodelay;

        proxy_pass http://django_backend;
    }
}

📊 壓力測試

使用 Locust 進行壓力測試：

# locustfile.py
from locust import HttpUser, task, between

class FlashSaleUser(HttpUser):
    wait_time = between(0.5, 1)

    def on_start(self):
        # 登入
        self.client.post('/login/', {
            'username': 'testuser',
            'password': 'password'
        })

    @task
    def purchase(self):
        self.client.post('/api/flash-sale/123/', name='秒殺購買')

運行測試：

# 模擬 1000 個用戶，每秒增加 100 個用戶
locust -f locustfile.py --host=http://localhost:8000 --users 1000 --spawn-rate 100

💡 面試要點

Q1: 秒殺系統的核心思路是什麼？

答：三個核心：

動靜分離：靜態資源走 CDN，減少後端壓力
削峰填谷：用消息隊列異步處理訂單
多層限流：從 CDN、Nginx 到應用層層限流

Q2: 為什麼要用 Redis 而不是直接操作數據庫？

答：

Redis 是內存數據庫，QPS 可達 10萬+
MySQL QPS 通常只有幾千
Redis 提供原子操作（Lua 腳本），避免超賣

Q3: 如何防止黃牛刷單？

答：

實名制：綁定身份證、手機號
限流：同一用戶/IP 限制請求頻率
驗證碼：增加自動化難度
風控系統：檢測異常行為

Q4: 如果 Redis 掛了怎麼辦？

答：

Redis 集群：主從複製 + 哨兵
熔斷降級：Redis 故障時直接返回「系統繁忙」
數據庫兜底：最壞情況下降級到數據庫（性能差但能用）

🔗 下一篇

在下一篇文章中，我們將深入學習 分布式鎖的實現，這是解決分布式環境下並發問題的關鍵技術。

閱讀時間：15 分鐘

Django 面試準備 08-2：秒殺系統設計