「产品矩阵平台」数据与缓存架构

Local Cache -> Redis -> Database

tenant:{tenant_id}:app:{app_id}:content:{content_id}:v{version}

{
  "event": "content_viewed",
  "tenant_id": "t_001",
  "app_id": "app_001",
  "user_id": "u_001",
  "anonymous_id": "anon_001",
  "timestamp": "2025-10-16T16:28:36+08:00",
  "properties": {
    "content_id": "c_001",
    "source": "feed"
  }
}

手机号：138****5678
邮箱：le***@example.com
身份证：110***********1234

package cache

import (
    "context"
    "encoding/json"
    "fmt"
    "time"
)

type CacheAside[T any] struct {
    redis      RedisClient
    keyPrefix  string
    ttl        time.Duration
    lockTTL    time.Duration
}

func (c *CacheAside[T]) Get(ctx context.Context, key string, loader func() (T, error)) (T, error) {
    var result T
    cacheKey := fmt.Sprintf("%s:%s", c.keyPrefix, key)

    // 1. 先读缓存
    val, err := c.redis.Get(ctx, cacheKey).Result()
    if err == nil {
        _ = json.Unmarshal([]byte(val), &result)
        return result, nil
    }

    // 2. 缓存未命中，获取单飞锁
    lockKey := cacheKey + ":lock"
    lockOk, _ := c.redis.SetNX(ctx, lockKey, "1", c.lockTTL).Result()
    if !lockOk {
        // 其他 goroutine 正在回源，稍等再试
        time.Sleep(50 * time.Millisecond)
        return c.Get(ctx, key, loader)
    }
    defer c.redis.Del(ctx, lockKey)

    // 3. 双重检查
    val, err = c.redis.Get(ctx, cacheKey).Result()
    if err == nil {
        _ = json.Unmarshal([]byte(val), &result)
        return result, nil
    }

    // 4. 回源
    result, err = loader()
    if err != nil {
        return result, err
    }

    // 5. 写入缓存（空值也缓存防穿透，TTL 更短）
    data, _ := json.Marshal(result)
    ttl := c.ttl
    if isZero(result) {
        ttl = 5 * time.Minute // 空值缓存 5 分钟
    }
    _ = c.redis.Set(ctx, cacheKey, data, ttl).Err()
    return result, nil
}

func (c *CacheAside[T]) Invalidate(ctx context.Context, key string) error {
    cacheKey := fmt.Sprintf("%s:%s", c.keyPrefix, key)
    return c.redis.Del(ctx, cacheKey).Err()
}

package cache

import (
    "context"
    "time"
    "github.com/dgraph-io/ristretto"
)

type MultiLevelCache[T any] struct {
    local      *ristretto.Cache
    redis      RedisClient
    ttlLocal   time.Duration
    ttlRedis   time.Duration
    lockTTL    time.Duration
    keyPrefix  string
}

func (c *MultiLevelCache[T]) Get(ctx context.Context, key string, loader func() (T, error)) (T, error) {
    var result T
    cacheKey := fmt.Sprintf("%s:%s", c.keyPrefix, key)

    // 1. 先查本地缓存（Ristretto，零 GC，纳秒级）
    if val, ok := c.local.Get(cacheKey); ok {
        return val.(T), nil
    }

    // 2. 再查 Redis
    val, err := c.redis.Get(ctx, cacheKey).Result()
    if err == nil {
        _ = json.Unmarshal([]byte(val), &result)
        c.local.SetWithTTL(cacheKey, result, 1, c.ttlLocal)
        return result, nil
    }

    // 3. 回源(Cache-Aside + singleflight)
    result, err = c.loadWithSingleflight(ctx, cacheKey, loader)
    if err != nil {
        return result, err
    }

    // 4. 回填两级缓存
    data, _ := json.Marshal(result)
    _ = c.redis.Set(ctx, cacheKey, data, c.ttlRedis).Err()
    c.local.SetWithTTL(cacheKey, result, 1, c.ttlLocal)
    return result, nil
}

// 订阅配置变更，通知实例失效本地缓存
func (c *MultiLevelCache[T]) SubscribeInvalidate(channel string) {
    pubsub := c.redis.Subscribe(context.Background(), channel)
    go func() {
        for msg := range pubsub.Channel() {
            c.local.Del(msg.Payload)
        }
    }()
}

package lock

import (
    "context"
    "fmt"
    "time"
    "github.com/go-redsync/redsync/v4"
    "github.com/go-redsync/redsync/v4/redis/goredis/v9"
)

type DistributedLock struct {
    rs    *redsync.Redsync
    mutex *redsync.Mutex
}

func NewDistributedLock(pool *redis.Pool, resource string, ttl time.Duration) *DistributedLock {
    rs := redsync.New(goredis.NewPool(pool))
    return &DistributedLock{
        rs:    rs,
        mutex: rs.NewMutex(resource, redsync.WithExpiry(ttl)),
    }
}

func (l *DistributedLock) Do(ctx context.Context, fn func() error) error {
    if err := l.mutex.LockContext(ctx); err != nil {
        return fmt.Errorf("lock failed: %w", err)
    }
    defer l.mutex.UnlockContext(ctx)
    return fn()
}

// 使用示例：库存扣减
func (svc *InventoryService) Deduct(ctx context.Context, productID string, qty int) error {
    lock := NewDistributedLock(svc.redisPool, "lock:inventory:"+productID, 5*time.Second)
    return lock.Do(ctx, func() error {
        current, err := svc.GetStock(ctx, productID)
        if err != nil {
            return err
        }
        if current < qty {
            return fmt.Errorf("insufficient stock: %d < %d", current, qty)
        }
        return svc.UpdateStock(ctx, productID, current-qty)
    })
}

package outbox

import (
    "context"
    "encoding/json"
    "gorm.io/gorm"
    "time"
)

type OutboxEvent struct {
    ID        string    `gorm:"primaryKey"`
    EventType string    `gorm:"index"`
    TenantID  string    `gorm:"index"`
    AppID     string    `gorm:"index"`
    Payload   []byte
    CreatedAt time.Time
    SentAt    *time.Time `gorm:"index"`
}

// PublishInTransaction 在业务事务内写 Outbox
type Publisher struct {
    db *gorm.DB
}

func (p *Publisher) PublishInTransaction(ctx context.Context, eventType, tenantID, appID string, payload interface{}) error {
    data, _ := json.Marshal(payload)
    return p.db.WithContext(ctx).Create(&OutboxEvent{
        ID:        uuid.New().String(),
        EventType: eventType,
        TenantID:  tenantID,
        AppID:     appID,
        Payload:   data,
        CreatedAt: time.Now(),
    }).Error
}

// RelayWorker 扫描未投递事件并发送到消息队列
func (p *Publisher) RelayWorker(ctx context.Context, sender func(OutboxEvent) error) {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            var events []OutboxEvent
            p.db.Where("sent_at IS NULL").Order("created_at").Limit(100).Find(&events)
            for _, e := range events {
                if err := sender(e); err != nil {
                    continue // 重试下一批
                }
                now := time.Now()
                p.db.Model(&e).Update("sent_at", now)
            }
        }
    }
}

package desensitization

import (
    "regexp"
    "strings"
)

func Phone(phone string) string {
    if len(phone) != 11 {
        return phone
    }
    return phone[:3] + "****" + phone[7:]
}

func Email(email string) string {
    parts := strings.Split(email, "@")
    if len(parts) != 2 {
        return email
    }
    local := parts[0]
    if len(local) <= 1 {
        return email
    }
    return local[:1] + strings.Repeat("*", len(local)-1) + "@" + parts[1]
}

func IDCard(id string) string {
    if len(id) != 18 {
        return id
    }
    return id[:6] + "********" + id[14:]
}

// 在 Repository 中封装脱敏
func DesensitizeUser(user *User) {
    user.Phone = Phone(user.Phone)
    user.Email = Email(user.Email)
    user.IDCard = IDCard(user.IDCard)
}