云原生分布式架构实战：基于 Canal + 消息队列解决高并发下缓存与数据库双写一致性

[客户端] ──写请求──> [业务微服务] ──执行事务──> [云数据库 RDS/MySQL]
                                                   │
                                          写入 Binlog (ROW 模式)
                                                   │
                                                   ▼
                                         [Canal / 增量同步服务] (伪装成 Slave 拉取解析)
                                                   │
                                          结构化 JSON 消息投递
                                                   │
                                                   ▼
                                         [分布式消息队列 (Kafka/RocketMQ)]
                                                   │
                                            ACK 确认 / 失败重试
                                                   │
                                                   ▼
                                          [缓存同步消费服务] ──删除指定 Key──> [Redis 集群]

package main

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

	"github.com/redis/go-redis/v9"
)

var ctx = context.Background()

// CanalMessage 定义 Canal 投递到消息队列的标准结构
type CanalMessage struct {
	Database string            `json:"database"`
	Table    string            `json:"table"`
	Type     string            `json:"type"` // INSERT, UPDATE, DELETE
	Data     []map[string]any  `json:"data"` // 变更后的最新行数据
	Old      []map[string]any  `json:"old"`  // 变更前的旧行数据
}

// 缓存淘汰处理器
type CacheInvalidator struct {
	rdb *redis.Client
}

func (ci *CacheInvalidator) ProcessMessage(msgBytes []byte) error {
	var canalMsg CanalMessage
	if err := json.Unmarshal(msgBytes, &canalMsg); err != nil {
		return fmt.Errorf("解析 Canal 消息失败: %w", err)
	}

	// 过滤目标业务表（例如只关注商品表 product_info）
	if canalMsg.Database != "mall_prod" || canalMsg.Table != "product_info" {
		return nil
	}

	// 无论是 UPDATE 还是 DELETE，都需要从缓存中剔除失效项
	for _, row := range canalMsg.Data {
		productId, ok := row["id"]
		if !ok {
			continue
		}

		cacheKey := fmt.Sprintf("cache:product:%v", productId)

		// 执行删除并实施指数退避重试（最多重试 3 次）
		if err := ci.deleteWithRetry(cacheKey, 3); err != nil {
			log.Printf("[严重告警] 缓存删除彻底失败，需投递死信队列: Key=%s, Err=%v", cacheKey, err)
			return err // 向上抛出错误，拒绝 ACK
		}

		log.Printf("[成功淘汰] 缓存失效执行完毕: %s", cacheKey)
	}
	return nil
}

func (ci *CacheInvalidator) deleteWithRetry(key string, maxRetries int) error {
	var err error
	for i := 0; i < maxRetries; i++ {
		err = ci.rdb.Del(ctx, key).Err()
		if err == nil {
			return nil
		}
		// 指数退避间隔：100ms, 200ms, 400ms...
		time.Sleep(time.Duration(100*(1<<i)) * time.Millisecond)
	}
	return err
}

func main() {
	rdb := redis.NewClient(&redis.Options{
		Addr: "127.0.0.1:6379",
	})
	invalidator := &CacheInvalidator{rdb: rdb}

	// 模拟一条从消息队列拉取到的 Binlog 变更消息
	sampleCanalJSON := []byte(`{
		"database": "mall_prod",
		"table": "product_info",
		"type": "UPDATE",
		"data": [{"id": 10086, "name": "云原生计算实战", "price": 99.0}]
	}`)

	if err := invalidator.ProcessMessage(sampleCanalJSON); err != nil {
		log.Fatalf("处理失败: %v", err)
	}
}