02 - 秒杀系统设计

// 错误码定义
const (
    // 成功
    CodeSuccess = 0

    // 客户端错误 4xxxx
    CodeInvalidParam      = 40000  // 参数错误
    CodeStockNotEnough    = 40001  // 库存不足
    CodeDuplicatePurchase = 40002  // 重复购买
    CodeSeckillNotStart   = 40003  // 秒杀未开始
    CodeSeckillEnded      = 40004  // 秒杀已结束
    CodeInvalidToken      = 40005  // 令牌无效
    CodeUserLimitExceed   = 40006  // 超过限购数量

    // 限流/熔断 429xx
    CodeRateLimitExceeded = 42900  // 限流
    CodeCircuitBreakerOpen = 42901 // 熔断

    // 服务端错误 5xxxx
    CodeInternalError     = 50000  // 内部错误
    CodeServiceUnavailable = 50300 // 服务不可用
    CodeDatabaseError     = 50001  // 数据库错误
    CodeCacheError        = 50002  // 缓存错误
)

package service

import (
    "database/sql"
    "errors"
)

type SeckillService struct {
    db *sql.DB
}

// CreateOrder 创建秒杀订单（V1 版本 - 悲观锁）
func (s *SeckillService) CreateOrder(userID, seckillID int64, quantity int) (string, error) {
    // 开启事务
    tx, err := s.db.Begin()
    if err != nil {
        return "", err
    }
    defer tx.Rollback()

    // 1. 检查用户是否已购买（防重复）
    var count int
    checkSQL := "SELECT COUNT(*) FROM user_purchase WHERE user_id = ? AND seckill_id = ?"
    if err := tx.QueryRow(checkSQL, userID, seckillID).Scan(&count); err != nil {
        return "", err
    }
    if count > 0 {
        return "", errors.New("您已购买过该商品")
    }

    // 2. 锁定库存并扣减（悲观锁 FOR UPDATE）
    var stock, productID int64
    var price int
    lockSQL := `
        SELECT product_id, available_stock, seckill_price
        FROM seckill_product
        WHERE id = ? AND status = 1
        FOR UPDATE
    `
    if err := tx.QueryRow(lockSQL, seckillID).Scan(&productID, &stock, &price); err != nil {
        return "", err
    }

    if stock < quantity {
        return "", errors.New("库存不足")
    }

    // 扣减库存
    updateSQL := "UPDATE seckill_product SET available_stock = available_stock - ? WHERE id = ?"
    if _, err := tx.Exec(updateSQL, quantity, seckillID); err != nil {
        return "", err
    }

    // 3. 创建订单
    orderID := generateOrderID()
    insertOrderSQL := `
        INSERT INTO seckill_order (order_id, user_id, product_id, seckill_id, quantity, total_price, status, pay_expire_time)
        VALUES (?, ?, ?, ?, ?, ?, 0, DATE_ADD(NOW(), INTERVAL 15 MINUTE))
    `
    if _, err := tx.Exec(insertOrderSQL, orderID, userID, productID, seckillID, quantity, price*quantity); err != nil {
        return "", err
    }

    // 4. 记录用户购买
    insertPurchaseSQL := "INSERT INTO user_purchase (user_id, seckill_id, product_id, quantity) VALUES (?, ?, ?, ?)"
    if _, err := tx.Exec(insertPurchaseSQL, userID, seckillID, productID, quantity); err != nil {
        return "", err
    }

    // 5. 记录库存流水
    insertLogSQL := `
        INSERT INTO stock_log (seckill_id, order_id, user_id, quantity, stock_before, stock_after, operation_type)
        VALUES (?, ?, ?, ?, ?, ?, 1)
    `
    if _, err := tx.Exec(insertLogSQL, seckillID, orderID, userID, -quantity, stock, stock-quantity); err != nil {
        return "", err
    }

    // 提交事务
    if err := tx.Commit(); err != nil {
        return "", err
    }

    return orderID, nil
}

func generateOrderID() string {
    // 生成订单号: SK + 时间戳 + 随机数
    return fmt.Sprintf("SK%d%04d", time.Now().Unix(), rand.Intn(10000))
}

package middleware

import (
    "golang.org/x/time/rate"
    "sync"
)

// 用户维度限流器
type UserRateLimiter struct {
    limiters map[int64]*rate.Limiter
    mu       sync.RWMutex
    rate     rate.Limit  // 每秒请求数
    burst    int         // 桶容量
}

func NewUserRateLimiter(r rate.Limit, b int) *UserRateLimiter {
    return &UserRateLimiter{
        limiters: make(map[int64]*rate.Limiter),
        rate:     r,
        burst:    b,
    }
}

func (u *UserRateLimiter) GetLimiter(userID int64) *rate.Limiter {
    u.mu.Lock()
    defer u.mu.Unlock()

    limiter, exists := u.limiters[userID]
    if !exists {
        limiter = rate.NewLimiter(u.rate, u.burst)
        u.limiters[userID] = limiter
    }

    return limiter
}

func (u *UserRateLimiter) Allow(userID int64) bool {
    return u.GetLimiter(userID).Allow()
}

// HTTP 中间件
func RateLimitMiddleware(limiter *UserRateLimiter) gin.HandlerFunc {
    return func(c *gin.Context) {
        userID := getUserIDFromContext(c)

        if !limiter.Allow(userID) {
            c.JSON(429, gin.H{
                "code":    42900,
                "message": "请求过于频繁，请稍后再试",
            })
            c.Abort()
            return
        }

        c.Next()
    }
}

package service

import (
    "context"
    "github.com/go-redis/redis/v8"
)

type SeckillServiceV2 struct {
    rdb   *redis.Client
    db    *sql.DB
    kafka *kafka.Producer
}

// Lua 脚本：原子扣减库存
const decrStockScript = `
local key = KEYS[1]
local userKey = KEYS[2]
local quantity = tonumber(ARGV[1])
local userID = ARGV[2]

-- 检查用户是否已购买
if redis.call('SISMEMBER', userKey, userID) == 1 then
    return -2  -- 重复购买
end

-- 扣减库存
local stock = tonumber(redis.call('GET', key) or 0)
if stock >= quantity then
    redis.call('DECRBY', key, quantity)
    redis.call('SADD', userKey, userID)
    return stock - quantity  -- 返回剩余库存
else
    return -1  -- 库存不足
end
`

func (s *SeckillServiceV2) CreateOrder(ctx context.Context, userID, seckillID int64, quantity int) (string, error) {
    stockKey := fmt.Sprintf("seckill:stock:%d", seckillID)
    userKey := fmt.Sprintf("seckill:users:%d", seckillID)

    // 执行 Lua 脚本预扣库存
    result, err := s.rdb.Eval(ctx, decrStockScript,
        []string{stockKey, userKey},
        quantity, userID).Int()
    if err != nil {
        return "", err
    }

    if result == -2 {
        return "", errors.New("您已购买过该商品")
    }
    if result == -1 {
        return "", errors.New("库存不足")
    }

    // 生成订单 ID
    orderID := generateOrderID()

    // 异步发送到 Kafka（削峰）
    orderMsg := OrderMessage{
        OrderID:   orderID,
        UserID:    userID,
        SeckillID: seckillID,
        Quantity:  quantity,
        Timestamp: time.Now().Unix(),
    }

    if err := s.kafka.Publish("seckill-orders", orderMsg); err != nil {
        // 发送失败，回滚 Redis 库存
        s.rollbackRedisStock(ctx, stockKey, userKey, userID, quantity)
        return "", err
    }

    return orderID, nil
}

// 回滚 Redis 库存
func (s *SeckillServiceV2) rollbackRedisStock(ctx context.Context, stockKey, userKey string, userID int64, quantity int) {
    s.rdb.IncrBy(ctx, stockKey, int64(quantity))
    s.rdb.SRem(ctx, userKey, userID)
}

package consumer

import (
    "github.com/Shopify/sarama"
)

type OrderConsumer struct {
    db *sql.DB
}

func (c *OrderConsumer) Consume(msg *sarama.ConsumerMessage) error {
    var order OrderMessage
    if err := json.Unmarshal(msg.Value, &order); err != nil {
        return err
    }

    // 开启事务
    tx, err := c.db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()

    // 1. 查询商品信息
    var productID int64
    var price int
    querySQL := "SELECT product_id, seckill_price FROM seckill_product WHERE id = ?"
    if err := tx.QueryRow(querySQL, order.SeckillID).Scan(&productID, &price); err != nil {
        return err
    }

    // 2. 插入订单
    insertOrderSQL := `
        INSERT INTO seckill_order (order_id, user_id, product_id, seckill_id, quantity, total_price, status, pay_expire_time)
        VALUES (?, ?, ?, ?, ?, ?, 0, DATE_ADD(NOW(), INTERVAL 15 MINUTE))
    `
    if _, err := tx.Exec(insertOrderSQL, order.OrderID, order.UserID, productID, order.SeckillID, order.Quantity, price*order.Quantity); err != nil {
        return err
    }

    // 3. 扣减 DB 库存（最终一致性）
    updateSQL := "UPDATE seckill_product SET available_stock = available_stock - ? WHERE id = ? AND available_stock >= ?"
    result, err := tx.Exec(updateSQL, order.Quantity, order.SeckillID, order.Quantity)
    if err != nil {
        return err
    }

    affected, _ := result.RowsAffected()
    if affected == 0 {
        // DB 库存不足（Redis 和 DB 不一致），需要告警
        log.Error("DB stock insufficient", "seckill_id", order.SeckillID)
        return errors.New("DB 库存不足")
    }

    // 4. 记录用户购买
    insertPurchaseSQL := "INSERT INTO user_purchase (user_id, seckill_id, product_id, quantity) VALUES (?, ?, ?, ?)"
    if _, err := tx.Exec(insertPurchaseSQL, order.UserID, order.SeckillID, productID, order.Quantity); err != nil {
        return err
    }

    // 5. 记录库存流水
    insertLogSQL := `
        INSERT INTO stock_log (seckill_id, order_id, user_id, quantity, stock_before, stock_after, operation_type)
        VALUES (?, ?, ?, ?, 0, 0, 1)
    `
    if _, err := tx.Exec(insertLogSQL, order.SeckillID, order.OrderID, order.UserID, -order.Quantity); err != nil {
        return err
    }

    // 提交事务
    return tx.Commit()
}

package service

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/base64"
    "fmt"
    "time"
)

type TokenService struct {
    secretKey string
}

// GenerateToken 生成秒杀令牌
func (t *TokenService) GenerateToken(userID, seckillID int64) (string, error) {
    // 令牌有效期 10 秒
    expireTime := time.Now().Add(10 * time.Second).Unix()

    // 生成签名: HMAC-SHA256(userID|seckillID|expireTime, secretKey)
    data := fmt.Sprintf("%d|%d|%d", userID, seckillID, expireTime)
    h := hmac.New(sha256.New, []byte(t.secretKey))
    h.Write([]byte(data))
    signature := base64.URLEncoding.EncodeToString(h.Sum(nil))

    // 令牌格式: userID|seckillID|expireTime|signature
    token := fmt.Sprintf("%s|%s", data, signature)
    return base64.URLEncoding.EncodeToString([]byte(token)), nil
}

// VerifyToken 验证秒杀令牌
func (t *TokenService) VerifyToken(token string, userID, seckillID int64) error {
    // 解码
    decoded, err := base64.URLEncoding.DecodeString(token)
    if err != nil {
        return errors.New("令牌格式错误")
    }

    // 解析
    parts := strings.Split(string(decoded), "|")
    if len(parts) != 4 {
        return errors.New("令牌格式错误")
    }

    tokenUserID, _ := strconv.ParseInt(parts[0], 10, 64)
    tokenSeckillID, _ := strconv.ParseInt(parts[1], 10, 64)
    expireTime, _ := strconv.ParseInt(parts[2], 10, 64)
    signature := parts[3]

    // 验证用户和商品
    if tokenUserID != userID || tokenSeckillID != seckillID {
        return errors.New("令牌无效")
    }

    // 验证过期时间
    if time.Now().Unix() > expireTime {
        return errors.New("令牌已过期")
    }

    // 验证签名
    data := fmt.Sprintf("%d|%d|%d", userID, seckillID, expireTime)
    h := hmac.New(sha256.New, []byte(t.secretKey))
    h.Write([]byte(data))
    expectedSignature := base64.URLEncoding.EncodeToString(h.Sum(nil))

    if signature != expectedSignature {
        return errors.New("令牌签名错误")
    }

    return nil
}

package middleware

import (
    "github.com/sony/gobreaker"
    "time"
)

// 熔断器配置
func NewCircuitBreaker() *gobreaker.CircuitBreaker {
    settings := gobreaker.Settings{
        Name:        "seckill",
        MaxRequests: 3,                     // 半开状态最大请求数
        Interval:    10 * time.Second,      // 统计周期
        Timeout:     30 * time.Second,      // 熔断后多久尝试恢复
        ReadyToTrip: func(counts gobreaker.Counts) bool {
            // 失败率 > 50% 或 连续失败 > 5 次，触发熔断
            failureRatio := float64(counts.TotalFailures) / float64(counts.Requests)
            return counts.Requests >= 10 && (failureRatio >= 0.5 || counts.ConsecutiveFailures > 5)
        },
        OnStateChange: func(name string, from gobreaker.State, to gobreaker.State) {
            log.Infof("Circuit breaker %s state changed from %s to %s", name, from, to)
        },
    }

    return gobreaker.NewCircuitBreaker(settings)
}

// HTTP 中间件
func CircuitBreakerMiddleware(cb *gobreaker.CircuitBreaker) gin.HandlerFunc {
    return func(c *gin.Context) {
        _, err := cb.Execute(func() (interface{}, error) {
            c.Next()

            // 如果响应状态码 >= 500，视为失败
            if c.Writer.Status() >= 500 {
                return nil, errors.New("service error")
            }
            return nil, nil
        })

        if err != nil {
            // 熔断器打开，返回降级响应
            c.JSON(503, gin.H{
                "code":    50300,
                "message": "服务暂时不可用，请稍后重试",
            })
            c.Abort()
        }
    }
}

package lock

import (
    "context"
    "github.com/go-redis/redis/v8"
    "time"
)

type RedisLock struct {
    rdb *redis.Client
}

// Lock 获取分布式锁
func (l *RedisLock) Lock(ctx context.Context, key string, ttl time.Duration) (bool, error) {
    // SET key value NX EX ttl
    result, err := l.rdb.SetNX(ctx, key, "1", ttl).Result()
    return result, err
}

// Unlock 释放锁
func (l *RedisLock) Unlock(ctx context.Context, key string) error {
    return l.rdb.Del(ctx, key).Err()
}

// 使用示例：库存对账时加锁
func (s *SeckillService) ReconcileStock(ctx context.Context, seckillID int64) error {
    lockKey := fmt.Sprintf("lock:reconcile:%d", seckillID)

    // 尝试获取锁
    locked, err := s.lock.Lock(ctx, lockKey, 30*time.Second)
    if err != nil {
        return err
    }
    if !locked {
        return errors.New("对账任务正在进行中")
    }
    defer s.lock.Unlock(ctx, lockKey)

    // 执行对账逻辑
    // ...

    return nil
}

package cache

import (
    "github.com/patrickmn/go-cache"
    "time"
)

type MultiLevelCache struct {
    local *cache.Cache
    redis *redis.Client
    db    *sql.DB
}

func NewMultiLevelCache(rdb *redis.Client, db *sql.DB) *MultiLevelCache {
    return &MultiLevelCache{
        local: cache.New(5*time.Minute, 10*time.Minute),
        redis: rdb,
        db:    db,
    }
}

// GetProduct 获取商品信息（三级缓存）
func (m *MultiLevelCache) GetProduct(ctx context.Context, seckillID int64) (*Product, error) {
    cacheKey := fmt.Sprintf("product:%d", seckillID)

    // L1: 本地缓存
    if val, found := m.local.Get(cacheKey); found {
        return val.(*Product), nil
    }

    // L2: Redis
    var product Product
    val, err := m.redis.Get(ctx, cacheKey).Result()
    if err == nil {
        json.Unmarshal([]byte(val), &product)
        m.local.Set(cacheKey, &product, cache.DefaultExpiration)
        return &product, nil
    }

    // L3: MySQL
    err = m.db.QueryRow("SELECT * FROM seckill_product WHERE id = ?", seckillID).Scan(&product)
    if err != nil {
        return nil, err
    }

    // 回写缓存
    data, _ := json.Marshal(product)
    m.redis.Set(ctx, cacheKey, data, 1*time.Hour)
    m.local.Set(cacheKey, &product, cache.DefaultExpiration)

    return &product, nil
}

package job

import (
    "time"
)

type OrderTimeoutJob struct {
    db    *sql.DB
    redis *redis.Client
}

// Run 定时扫描超时订单
func (j *OrderTimeoutJob) Run() {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for range ticker.C {
        j.cancelTimeoutOrders()
    }
}

func (j *OrderTimeoutJob) cancelTimeoutOrders() {
    // 查询超时订单（支付超时时间 < 当前时间，且状态为待支付）
    query := `
        SELECT id, order_id, user_id, seckill_id, quantity
        FROM seckill_order
        WHERE status = 0 AND pay_expire_time < NOW()
        LIMIT 100
    `

    rows, err := j.db.Query(query)
    if err != nil {
        log.Error("查询超时订单失败", err)
        return
    }
    defer rows.Close()

    for rows.Next() {
        var id, userID, seckillID, quantity int64
        var orderID string
        rows.Scan(&id, &orderID, &userID, &seckillID, &quantity)

        // 取消订单
        j.cancelOrder(id, orderID, userID, seckillID, quantity)
    }
}

func (j *OrderTimeoutJob) cancelOrder(id int64, orderID string, userID, seckillID, quantity int64) {
    ctx := context.Background()

    // 开启事务
    tx, err := j.db.Begin()
    if err != nil {
        return
    }
    defer tx.Rollback()

    // 1. 更新订单状态为超时取消
    updateSQL := "UPDATE seckill_order SET status = 3 WHERE id = ? AND status = 0"
    result, err := tx.Exec(updateSQL, id)
    if err != nil {
        return
    }

    affected, _ := result.RowsAffected()
    if affected == 0 {
        return  // 订单已被处理
    }

    // 2. 回滚库存
    rollbackSQL := "UPDATE seckill_product SET available_stock = available_stock + ? WHERE id = ?"
    tx.Exec(rollbackSQL, quantity, seckillID)

    // 3. 删除用户购买记录
    deleteSQL := "DELETE FROM user_purchase WHERE user_id = ? AND seckill_id = ?"
    tx.Exec(deleteSQL, userID, seckillID)

    // 4. 记录库存流水
    insertLogSQL := `
        INSERT INTO stock_log (seckill_id, order_id, user_id, quantity, stock_before, stock_after, operation_type)
        VALUES (?, ?, ?, ?, 0, 0, 2)
    `
    tx.Exec(insertLogSQL, seckillID, orderID, userID, quantity)

    // 提交事务
    if err := tx.Commit(); err != nil {
        return
    }

    // 5. 回滚 Redis 库存
    stockKey := fmt.Sprintf("seckill:stock:%d", seckillID)
    userKey := fmt.Sprintf("seckill:users:%d", seckillID)
    j.redis.IncrBy(ctx, stockKey, quantity)
    j.redis.SRem(ctx, userKey, userID)

    log.Infof("订单 %s 超时取消，库存已回滚", orderID)
}

package filter

import (
    "github.com/bits-and-blooms/bloom/v3"
)

type PurchaseFilter struct {
    filter *bloom.BloomFilter
}

func NewPurchaseFilter(expectedItems uint, falsePositiveRate float64) *PurchaseFilter {
    return &PurchaseFilter{
        filter: bloom.NewWithEstimates(expectedItems, falsePositiveRate),
    }
}

// MayExist 检查用户是否可能已购买（快速过滤）
func (p *PurchaseFilter) MayExist(userID, seckillID int64) bool {
    key := fmt.Sprintf("%d:%d", userID, seckillID)
    return p.filter.TestString(key)
}

// Add 添加购买记录
func (p *PurchaseFilter) Add(userID, seckillID int64) {
    key := fmt.Sprintf("%d:%d", userID, seckillID)
    p.filter.AddString(key)
}

// 使用示例
func (s *SeckillService) CreateOrder(userID, seckillID int64) (string, error) {
    // 第一步：Bloom Filter 快速判断
    if s.filter.MayExist(userID, seckillID) {
        // 可能已购买，进一步检查 Redis
        userKey := fmt.Sprintf("seckill:users:%d", seckillID)
        exists, _ := s.rdb.SIsMember(ctx, userKey, userID).Result()
        if exists {
            return "", errors.New("您已购买过该商品")
        }
    }

    // 继续秒杀流程...
}

package idgen

import (
    "fmt"
    "sync"
    "time"
)

// Snowflake ID 生成器
type SnowflakeIDGenerator struct {
    mu           sync.Mutex
    timestamp    int64
    machineID    int64  // 机器 ID (10 bits)
    sequence     int64  // 序列号 (12 bits)
}

const (
    epoch           = int64(1609459200000)  // 2021-01-01 00:00:00
    machineBits     = uint(10)
    sequenceBits    = uint(12)
    machineMax      = int64(-1) ^ (int64(-1) << machineBits)
    sequenceMask    = int64(-1) ^ (int64(-1) << sequenceBits)
    machineShift    = sequenceBits
    timestampShift  = machineBits + sequenceBits
)

func NewSnowflakeIDGenerator(machineID int64) *SnowflakeIDGenerator {
    if machineID < 0 || machineID > machineMax {
        panic("invalid machine ID")
    }

    return &SnowflakeIDGenerator{
        timestamp: 0,
        machineID: machineID,
        sequence:  0,
    }
}

func (s *SnowflakeIDGenerator) NextID() (int64, error) {
    s.mu.Lock()
    defer s.mu.Unlock()

    now := time.Now().UnixNano() / 1e6

    if now < s.timestamp {
        return 0, fmt.Errorf("clock moved backwards")
    }

    if now == s.timestamp {
        // 同一毫秒内，序列号自增
        s.sequence = (s.sequence + 1) & sequenceMask
        if s.sequence == 0 {
            // 序列号溢出，等待下一毫秒
            for now <= s.timestamp {
                now = time.Now().UnixNano() / 1e6
            }
        }
    } else {
        // 新的毫秒，序列号重置
        s.sequence = 0
    }

    s.timestamp = now

    id := ((now - epoch) << timestampShift) |
          (s.machineID << machineShift) |
          s.sequence

    return id, nil
}

// 生成订单号
func GenerateOrderID(machineID int64) string {
    gen := NewSnowflakeIDGenerator(machineID)
    id, _ := gen.NextID()
    return fmt.Sprintf("SK%d", id)
}

package ratelimit

import (
    "sync"
    "time"
)

type TokenBucket struct {
    capacity   int64        // 桶容量
    tokens     int64        // 当前令牌数
    rate       int64        // 令牌生成速率（个/秒）
    lastUpdate time.Time    // 上次更新时间
    mu         sync.Mutex
}

func NewTokenBucket(capacity, rate int64) *TokenBucket {
    return &TokenBucket{
        capacity:   capacity,
        tokens:     capacity,
        rate:       rate,
        lastUpdate: time.Now(),
    }
}

// TryAcquire 尝试获取令牌
func (tb *TokenBucket) TryAcquire(count int64) bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    // 计算应该新增的令牌数
    now := time.Now()
    elapsed := now.Sub(tb.lastUpdate).Seconds()
    newTokens := int64(elapsed * float64(tb.rate))

    // 更新令牌数（不超过容量）
    tb.tokens += newTokens
    if tb.tokens > tb.capacity {
        tb.tokens = tb.capacity
    }
    tb.lastUpdate = now

    // 检查是否有足够的令牌
    if tb.tokens >= count {
        tb.tokens -= count
        return true
    }

    return false
}

// 使用示例
var globalBucket = NewTokenBucket(100000, 10000)  // 容量 10 万，速率 1 万/秒

func HandleRequest() {
    if !globalBucket.TryAcquire(1) {
        // 限流
        return errors.New("系统繁忙，请稍后再试")
    }

    // 处理请求...
}

package job

// 秒杀开始前 1 小时预热缓存
type CacheWarmupJob struct {
    db    *sql.DB
    redis *redis.Client
}

func (j *CacheWarmupJob) Warmup() {
    // 查询即将开始的秒杀活动
    query := `
        SELECT id, product_id, title, original_price, seckill_price, stock, available_stock, start_time, end_time
        FROM seckill_product
        WHERE start_time > NOW() AND start_time < DATE_ADD(NOW(), INTERVAL 1 HOUR)
    `

    rows, err := j.db.Query(query)
    if err != nil {
        return
    }
    defer rows.Close()

    ctx := context.Background()

    for rows.Next() {
        var product SeckillProduct
        rows.Scan(&product.ID, &product.ProductID, &product.Title, ...)

        // 1. 缓存商品详情
        productKey := fmt.Sprintf("seckill:product:%d", product.ID)
        data, _ := json.Marshal(product)
        j.redis.Set(ctx, productKey, data, 2*time.Hour)

        // 2. 缓存库存
        stockKey := fmt.Sprintf("seckill:stock:%d", product.ID)
        j.redis.Set(ctx, stockKey, product.AvailableStock, 2*time.Hour)

        // 3. 初始化用户购买记录集合
        userKey := fmt.Sprintf("seckill:users:%d", product.ID)
        j.redis.Del(ctx, userKey)  // 清空旧数据

        log.Infof("预热秒杀商品 %d 成功", product.ID)
    }
}

package db

import (
    "database/sql"
    _ "github.com/go-sql-driver/mysql"
)

type DBCluster struct {
    master *sql.DB
    slaves []*sql.DB
    slaveIndex int
}

func NewDBCluster(masterDSN string, slaveDSNs []string) (*DBCluster, error) {
    master, err := sql.Open("mysql", masterDSN)
    if err != nil {
        return nil, err
    }

    slaves := make([]*sql.DB, len(slaveDSNs))
    for i, dsn := range slaveDSNs {
        slave, err := sql.Open("mysql", dsn)
        if err != nil {
            return nil, err
        }
        slaves[i] = slave
    }

    return &DBCluster{
        master: master,
        slaves: slaves,
    }, nil
}

// GetMaster 获取主库（写）
func (c *DBCluster) GetMaster() *sql.DB {
    return c.master
}

// GetSlave 获取从库（读，轮询负载均衡）
func (c *DBCluster) GetSlave() *sql.DB {
    if len(c.slaves) == 0 {
        return c.master
    }

    slave := c.slaves[c.slaveIndex%len(c.slaves)]
    c.slaveIndex++
    return slave
}

package sharding

import (
    "hash/crc32"
)

const (
    DBCount    = 8  // 8 个库
    TableCount = 8  // 每库 8 张表
)

// ShardingStrategy 分片策略
type ShardingStrategy struct{}

// GetShardDB 根据 user_id 计算库索引
func (s *ShardingStrategy) GetShardDB(userID int64) int {
    return int(userID % DBCount)
}

// GetShardTable 根据 user_id 计算表索引
func (s *ShardingStrategy) GetShardTable(userID int64) int {
    return int((userID / DBCount) % TableCount)
}

// GetTableName 获取实际表名
func (s *ShardingStrategy) GetTableName(userID int64) string {
    dbIndex := s.GetShardDB(userID)
    tableIndex := s.GetShardTable(userID)
    return fmt.Sprintf("db%d.seckill_order_%d", dbIndex, tableIndex)
}

// 使用示例
func (s *SeckillService) InsertOrder(order *Order) error {
    strategy := &ShardingStrategy{}
    tableName := strategy.GetTableName(order.UserID)

    sql := fmt.Sprintf(`
        INSERT INTO %s (order_id, user_id, product_id, ...)
        VALUES (?, ?, ?, ...)
    `, tableName)

    _, err := s.db.Exec(sql, order.OrderID, order.UserID, ...)
    return err
}