Redis 与 MySQL 数据最终一致性方案

目录

  1. 问题背景
  2. 一致性方案概述
  3. Canal + MQ 异步同步架构
  4. 核心组件详解
  5. 代码实现分析
  6. 两种更新模式对比
  7. 幂等保证机制
  8. 常见问题与解决方案
  9. 面试要点

问题背景

为什么需要解决一致性?

┌─────────────────────────────────────────────────────────────────┐
│                        高并发读场景                               │
│                                                                 │
│   用户请求 ──▶ Redis缓存(命中) ──▶ 直接返回 ──▶ 性能极高 ✅         │
│                                                                 │
│   问题:如果 Redis 和 MySQL 数据不一致?                          │
│   结果:用户看到的是过期/错误数据                                   │
└─────────────────────────────────────────────────────────────────┘

常见的缓存一致性问题

问题 描述 场景
缓存穿透 查询不存在的数据,每次都击穿到DB 恶意攻击/不存在的数据
缓存击穿 热点key过期瞬间,大量请求击穿到DB 爆款商品/热点数据
缓存雪崩 大量key同时过期,请求击穿到DB 批量设置过期时间
双写不一致 DB和Redis同时更新,数据不一致 并发更新/异步延迟

一致性方案概述

12306 项目的解决方案

┌─────────────────────────────────────────────────────────────────┐
│                     写入操作                                      │
│                                                                 │
│   业务操作 ──▶ 写 MySQL ──▶ 不直接写 Redis                        │
│                                    │                            │
│                                    ▼                            │
│                            由 Binlog 驱动异步同步                  │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                     读取操作                                      │
│                                                                 │
│   读请求 ──▶ 查询 Redis ──▶ 命中直接返回                          │
│                          │                                      │
│                          │ 未命中                                │
│                          ▼                                      │
│                    查询 MySQL ──▶ 回填 Redis ──▶ 返回            │
└─────────────────────────────────────────────────────────────────┘

方案选择依据

方案 优点 缺点 适用场景
同步双写 实时性高 性能损耗大,可能不一致 低并发业务
延迟双删 实现简单 删除时机难把握 允许短暂不一致
Canal+MQ 解耦、可靠、可重试 引入额外组件,有延迟 高并发业务 ✅
订阅Binlog 实时性较好 复杂度高 高一致性要求

12306 选择了 Canal + RocketMQ 方案,因为:

  1. 不影响主流程性能
  2. MQ 保证消息可靠送达
  3. 消费端可重复消费
  4. 适合高并发场景

Canal + MQ 异步同步架构

整体架构图

┌─────────────────────────────────────────────────────────────────────────────┐
│                              MySQL                                           │
│                                                                               │
│   INSERT/UPDATE/DELETE ──▶ Binlog 日志                                       │
│                               │                                              │
│                               ▼                                              │
│                      ┌────────────────┐                                      │
│                      │    Canal       │  (模拟 MySQL 主从复制协议)              │
│                      │  解析 Binlog   │                                      │
│                      └───────┬────────┘                                      │
│                              │                                               │
│                              ▼                                               │
│                      ┌────────────────┐                                      │
│                      │   RocketMQ     │  (异步消息队列)                         │
│                      │  主题: canal   │                                      │
│                      └───────┬────────┘                                      │
│                              │                                               │
│          ┌───────────────────┼───────────────────┐                           │
│          ▼                   ▼                   ▼                           │
│   ┌──────────────┐   ┌──────────────┐   ┌──────────────┐                    │
│   │ 余票缓存更新  │   │订单 缓存更新  │   │ 其他缓存更新  │                    │
│   │   Handler    │   │   Handler    │   │   Handler    │                    │
│   └──────┬───────┘   └──────┬───────┘   └──────┬───────┘                    │
│          │                  │                  │                             │
│          ▼                  ▼                  ▼                             │
│       Redis             Redis              Redis                             │
│                                                                               │
└─────────────────────────────────────────────────────────────────────────────┘

消息流转流程

1. 业务代码执行 SQL: UPDATE t_seat SET seat_status = 1 WHERE id = 123

2. MySQL 写入 Binlog:
   +----------------+------------------+----------+
   | table          | type             | old_data | new_data
   +----------------+------------------+----------+
   | t_seat         | UPDATE           | {seat_status: 0} | {seat_status: 1}
   +----------------+------------------+----------+

3. Canal 模拟从库,解析 Binlog,发送到 MQ:
   Topic: index12306_canal_ticket-service_common-sync_topic
   Message: CanalBinlogEvent {
       table: "t_seat",
       type: "UPDATE",
       data: [...],
       old: [...]
   }

4. CanalCommonSyncBinlogConsumer 消费消息:
   - 根据 table 名称路由到对应 Handler
   - 每个 Handler 执行具体的缓存更新逻辑

5. Handler 执行 Redis 操作:
   - TicketAvailabilityCacheUpdateHandler: 更新余票缓存
   - OrderCloseCacheAndTokenUpdateHandler: 订单关闭后置处理

核心组件详解

1. CanalCommonSyncBinlogConsumer

职责:监听 Canal Binlog 消息,根据表名路由到对应的 Handler。

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = TicketRocketMQConstant.CANAL_COMMON_SYNC_TOPIC_KEY,
        consumerGroup = TicketRocketMQConstant.CANAL_COMMON_SYNC_CG_KEY
)
public class CanalCommonSyncBinlogConsumer implements RocketMQListener<CanalBinlogEvent> {

    private final AbstractStrategyChoose abstractStrategyChoose;

    @Value("${ticket.availability.cache-update.type:}")
    private String ticketAvailabilityCacheUpdateType;

    @Idempotent(...)  // MQ 消费幂等
    @Override
    public void onMessage(CanalBinlogEvent message) {
        // 过滤非 UPDATE 操作和未开启 binlog 模式的场景
        if (message.getIsDdl()
                || CollUtil.isEmpty(message.getOld())
                || !Objects.equals("UPDATE", message.getType())
                || !StrUtil.equals(ticketAvailabilityCacheUpdateType, "binlog")) {
            return;
        }
        // 根据表名路由到对应的 Handler
        abstractStrategyChoose.chooseAndExecute(
                message.getTable(),
                message,
                CanalExecuteStrategyMarkEnum.isPatternMatch(message.getTable())
        );
    }
}


@Slf4j @Component @RequiredArgsConstructor @RocketMQMessageListener( topic = TicketRocketMQConstant.CANAL_COMMON_SYNC_TOPIC_KEY, consumerGroup = TicketRocketMQConstant.CANAL_COMMON_SYNC_CG_KEY ) public class CanalCommonSyncBinlogConsumer implements RocketMQListener<CanalBinlogEvent> { private final AbstractStrategyChoose abstractStrategyChoose; @Value("${ticket.availability.cache-update.type:}") private String ticketAvailabilityCacheUpdateType; @Idempotent(...) // MQ 消费幂等 @Override public void onMessage(CanalBinlogEvent message) { // 过滤非 UPDATE 操作和未开启 binlog 模式的场景 if (message.getIsDdl() || CollUtil.isEmpty(message.getOld()) || !Objects.equals("UPDATE", message.getType()) || !StrUtil.equals(ticketAvailabilityCacheUpdateType, "binlog")) { return; } // 根据表名路由到对应的 Handler abstractStrategyChoose.chooseAndExecute( message.getTable(), message, CanalExecuteStrategyMarkEnum.isPatternMatch(message.getTable()) ); } }

路由机制

// 使用策略模式,根据表名选择对应的 Handler
abstractStrategyChoose.chooseAndExecute(tableName, message, matchMark);

// 例如:
// t_seat      ──▶ TicketAvailabilityCacheUpdateHandler
// t_order     ──▶ OrderCloseCacheAndTokenUpdateHandler

// 使用策略模式,根据表名选择对应的 Handler abstractStrategyChoose.chooseAndExecute(tableName, message, matchMark); // 例如: // t_seat ──▶ TicketAvailabilityCacheUpdateHandler // t_order ──▶ OrderCloseCacheAndTokenUpdateHandler

2. TicketAvailabilityCacheUpdateHandler

职责:处理 t_seat 表变更,更新 Redis 中的余票缓存。

@Component
@RequiredArgsConstructor
public class TicketAvailabilityCacheUpdateHandler implements AbstractExecuteStrategy<CanalBinlogEvent, Void> {

    private final DistributedCache distributedCache;

    @Override
    public void execute(CanalBinlogEvent message) {
        List<Map<String, Object>> messageDataList = new ArrayList<>();
        List<Map<String, Object>> actualOldDataList = new ArrayList<>();

        // 遍历 Binlog 数据,筛选状态变更的记录
        for (int i = 0; i < message.getOld().size(); i++) {
            Map<String, Object> oldDataMap = message.getOld().get(i);
            if (oldDataMap.get("seat_status") != null) {
                Map<String, Object> currentDataMap = message.getData().get(i);
                // 只处理: 可用↔不可用 之间的转换
                if (StrUtil.equalsAny(currentDataMap.get("seat_status").toString(),
                        String.valueOf(SeatStatusEnum.AVAILABLE.getCode()),
                        String.valueOf(SeatStatusEnum.LOCKED.getCode()))) {
                    actualOldDataList.add(oldDataMap);
                    messageDataList.add(currentDataMap);
                }
            }
        }

        if (CollUtil.isEmpty(messageDataList)) {
            return;
        }

        // 聚合相同 key 的变更
        Map<String, Map<Integer, Integer>> cacheChangeKeyMap = new HashMap<>();
        for (int i = 0; i < messageDataList.size(); i++) {
            Map<String, Object> each = messageDataList.get(i);
            Map<String, Object> actualOldData = actualOldDataList.get(i);
            String seatStatus = actualOldData.get("seat_status").toString();

            // seat_status: 0=已购, 1=可用, 2=锁定
            // 状态从"已购"变为"可用/锁定" → 余票 +1
            // 状态从"可用/锁定"变为"已购" → 余票 -1
            int increment = Objects.equals(seatStatus, "0") ? -1 : 1;

            String trainId = each.get("train_id").toString();
            String hashCacheKey = TRAIN_STATION_REMAINING_TICKET + trainId + "_"
                    + each.get("start_station") + "_" + each.get("end_station");

            Integer seatType = Integer.parseInt(each.get("seat_type").toString());
            Integer num = seatTypeMap.get(seatType);
            seatTypeMap.put(seatType, num == null ? increment : num + increment);
            cacheChangeKeyMap.put(hashCacheKey, seatTypeMap);
        }

        // Redis HINCRBY 原子递增/递减
        StringRedisTemplate instance = (StringRedisTemplate) distributedCache.getInstance();
        cacheChangeKeyMap.forEach((cacheKey, cacheVal) ->
            cacheVal.forEach((seatType, num) ->
                instance.opsForHash().increment(cacheKey, String.valueOf(seatType), num)
            )
        );
    }
}

3. OrderCloseCacheAndTokenUpdateHandler

职责:处理 t_order 表变更,当订单关闭/取消时,回滚余票和令牌桶。

@Component
@RequiredArgsConstructor
public class OrderCloseCacheAndTokenUpdateHandler implements AbstractExecuteStrategy<CanalBinlogEvent, Void> {

    private final TicketOrderRemoteService ticketOrderRemoteService;
    private final SeatService seatService;
    private final TicketAvailabilityTokenBucket ticketAvailabilityTokenBucket;

    @Override
    public void execute(CanalBinlogEvent message) {
        // 只处理 status=30 (订单关闭) 的记录
        List<Map<String, Object>> messageDataList = message.getData().stream()
                .filter(each -> each.get("status") != null)
                .filter(each -> Objects.equals(each.get("status"), "30"))
                .toList();

        if (CollUtil.isEmpty(messageDataList)) {
            return;
        }

        for (Map<String, Object> each : messageDataList) {
            String orderSn = each.get("order_sn").toString();
            Result<TicketOrderDetailRespDTO> orderDetailResult =
                    ticketOrderRemoteService.queryTicketOrderByOrderSn(orderSn);

            if (orderDetailResult.isSuccess() && orderDetailResult.getData() != null) {
                TicketOrderDetailRespDTO orderDetail = orderDetailResult.getData();

                // 1. 回滚座位状态 (锁定 → 可用)
                seatService.unlock(
                        String.valueOf(orderDetail.getTrainId()),
                        orderDetail.getDeparture(),
                        orderDetail.getArrival(),
                        BeanUtil.convert(orderDetail.getPassengerDetails(),
                                TrainPurchaseTicketRespDTO.class)
                );

                // 2. 回滚令牌桶
                ticketAvailabilityTokenBucket.rollbackInBucket(orderDetail);
            }
        }
    }
}

职责:消费延时消息,处理订单超时未支付的关闭逻辑。

@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = TicketRocketMQConstant.ORDER_DELAY_CLOSE_TOPIC_KEY,
        selectorExpression = TicketRocketMQConstant.ORDER_DELAY_CLOSE_TAG_KEY,
        consumerGroup = TicketRocketMQConstant.TICKET_DELAY_CLOSE_CG_KEY
)
public class DelayCloseOrderConsumer implements RocketMQListener<MessageWrapper<DelayCloseOrderEvent>> {

    @Idempotent(...)  // MQ 消费幂等
    @Override
    public void onMessage(MessageWrapper<DelayCloseOrderEvent> delayCloseOrderEventMessageWrapper) {
        DelayCloseOrderEvent delayCloseOrderEvent = delayCloseOrderEventMessageWrapper.getMessage();
        String orderSn = delayCloseOrderEvent.getOrderSn();

        // 1. 关闭订单
        Result<Boolean> closedTickOrder = ticketOrderRemoteService.closeTickOrder(
                new CancelTicketOrderReqDTO(orderSn));

        if (closedTickOrder.isSuccess() && !StrUtil.equals(ticketAvailabilityCacheUpdateType, "binlog")) {
            if (!closedTickOrder.getData()) {
                log.info("[延迟关闭订单] 订单号:{} 用户已支付订单", orderSn);
                return;  // 用户已支付,无需处理
            }

            // 2. 回滚座位状态
            seatService.unlock(trainId, departure, arrival, trainPurchaseTicketResults);

            // 3. 回滚 Redis 余票缓存
            StringRedisTemplate stringRedisTemplate = (StringRedisTemplate) distributedCache.getInstance();
            Map<Integer, List<TrainPurchaseTicketRespDTO>> seatTypeMap = ...
            routeDTOList.forEach(each -> {
                seatTypeMap.forEach((seatType, list) -> {
                    stringRedisTemplate.opsForHash()
                            .increment(TRAIN_STATION_REMAINING_TICKET + keySuffix, String.valueOf(seatType), list.size());
                });
            });

            // 4. 回滚令牌桶
            ticketAvailabilityTokenBucket.rollbackInBucket(ticketOrderDetail);
        }
    }
}


代码实现分析

Binlog 消息结构

ublic class CanalBinlogEvent {
    private String table;           // 表名: t_seat, t_order
    private String type;            // 操作类型: INSERT, UPDATE, DELETE
    private List<Map<String, Object>> data;    // 变更后的数据
    private List<Map<String, Object>> old;     // 变更前的数据
    private Boolean isDdl;          // 是否是 DDL 操作
}

Redis Hash 结构设计

# Key 格式: train_station_remaining_ticket_{trainId}_{出发站}_{到达站}
# Field: 座位类型 (seat_type)
# Value: 余票数量

Key: train_station_remaining_ticket_G202_北京_上海
     │
     ├─ Field: "1" (一等座)  →  Value: "100"
     ├─ Field: "2" (二等座)  →  Value: "200"
     └─ Field: "3" (商务座)  →  Value: "50"

# Key 格式: train_station_remaining_ticket_{trainId}_{出发站}_{到达站} # Field: 座位类型 (seat_type) # Value: 余票数量 Key: train_station_remaining_ticket_G202_北京_上海 │ ├─ Field: "1" (一等座) → Value: "100" ├─ Field: "2" (二等座) → Value: "200" └─ Field: "3" (商务座) → Value: "50"

原子操作: HINCRBY

// 使用 Redis HINCRBY 保证原子性
// 座位从"已购"变为"可用": HINCRBY key seat_type +1
// 座位从"可用"变为"已购": HINCRBY key seat_type -1

instance.opsForHash().increment(cacheKey, String.valueOf(seatType), increment);


// 使用 Redis HINCRBY 保证原子性 // 座位从"已购"变为"可用": HINCRBY key seat_type +1 // 座位从"可用"变为"已购": HINCRBY key seat_type -1 instance.opsForHash().increment(cacheKey, String.valueOf(seatType), increment);


两种更新模式对比

配置项

# application.yml
ticket:
  availability:
    cache-update:
      type: binlog  # 空=sync同步模式, binlog=异步模式

Sync 同步模式

业务流程:
1. 更新 MySQL
2. 同步更新 Redis
3. 返回成功

问题:
- 如果 Redis 更新失败,整个事务回滚?
- 还是忽略 Redis 失败继续提交?
- 选择忽略的话,又变成不一致了

// 同步模式代码路径: DelayCloseOrderConsumer 第96行
if (closedTickOrder.isSuccess() && !StrUtil.equals(ticketAvailabilityCacheUpdateType, "binlog")) {
    // 只有非 binlog 模式时才在这里更新缓存
    // binlog 模式下由 CanalCommonSyncBinlogConsumer 异步处理
}

Binlog 异步模式 (推荐)

业务流程:

1. 更新 MySQL
2. 提交事务
3. Canal 解析 Binlog
4. 发送 MQ 消息
5. 消费者更新 Redis

优点:
- 不影响主流程性能
- 由 MQ 保证消息可靠送达
- 消费者可重试,可幂等
- 即使 Redis 暂时不可用,消息积压后恢复继续处理

缺点:
- 有短暂延迟 (秒级)
- 引入额外组件 (Canal, RocketMQ)

性能对比

指标 Sync 同步模式 Binlog 异步模式
写入延迟 高 (等待 Redis) 低 (只等 MySQL)
Redis 失败影响 可能导致业务失败 无影响,异步重试
数据一致性 强一致 最终一致
复杂度
适用场景 低并发 高并发 ✅

幂等保证机制

为什么需要幂等?

问题场景: 消费者处理成功后,回应 MQ 时网络中断

结果: MQ 以为消费失败,会重新投递消息

如果没有幂等: 消息会被重复消费,导致数据重复更新

12306 的幂等实现


// DelayCloseOrderConsumer @Idempotent( uniqueKeyPrefix = "index12306-ticket:delay_close_order:", key = "#delayCloseOrderEventMessageWrapper.getKeys()" + "+'_'" + "#delayCloseOrderEventMessageWrapper.hashCode()", type = IdempotentTypeEnum.SPEL, scene = IdempotentSceneEnum.MQ, keyTimeout = 7200L // 2小时超时 ) @Override public void onMessage(MessageWrapper<DelayCloseOrderEvent> delayCloseOrderEventMessageWrapper) { // ... }

幂等 Key 生成逻辑


Key = "index12306-ticket:delay_close_order:" + orderSn + "_" + hashCode 例如: "index12306-ticket:delay_close_order:ORDER123456_12345678" 处理流程: 1. 第一次消费: SETNX key成功 → 执行业务 → SET key=result 2. 第二次消费: SETNX key失败 → 直接返回上次结果

CanalCommonSyncBinlogConsumer 幂等

// DelayCloseOrderConsumer
@Idempotent(
        uniqueKeyPrefix = "index12306-ticket:delay_close_order:",
        key = "#delayCloseOrderEventMessageWrapper.getKeys()"
                + "+'_'"
                + "#delayCloseOrderEventMessageWrapper.hashCode()",
        type = IdempotentTypeEnum.SPEL,
        scene = IdempotentSceneEnum.MQ,
        keyTimeout = 7200L  // 2小时超时
)
@Override
public void onMessage(MessageWrapper<DelayCloseOrderEvent> delayCloseOrderEventMessageWrapper) {
    // ...
}

常见问题与解决方案

问题1: Binlog 延迟导致短暂不一致

现象: 用户下单成功后,立即查询余票,发现还有票(实际已售完)

原因: Binlog 解析和 MQ 传递有延迟

解决方案:

  1. 前端展示"余票紧张"提示,不显示具体数量
  2. 提交订单时再次校验余票(双保险)
  3. 使用令牌桶预扣减机制

问题2: Redis 宕机期间的消息积压

现象: Redis 重启后,大概率 MQ 消息积压

解决方案:

  1. RocketMQ 本身支持消息堆积
  2. 消费者幂等设计,重复消息不会重复处理
  3. 逐步消费积压消息,最终达到一致

问题3: Canal 解析 Binlog 顺序问题

现象: 同一订单的多次变更是乱序到达

解决方案:

  1. 消费者根据订单号做处理,不依赖顺序
  2. 使用订单状态的最新值覆盖旧值
  3. 最终状态以 MySQL 为准

问题4: 热点数据更新竞争

现象: 同一 key 的并发更新导致数据覆盖

解决方案:


// 使用 Redis HINCRBY 原子递增/递减 // 而不是 GET → +1 → SET instance.opsForHash().increment(cacheKey, field, 1);


面试要点

Q1: 如何保证 Redis 和 MySQL 的数据一致性?

参考答案: 方案有多种,根据场景选择:

  1. Cache Aside (最常用): 读时懒加载,写时只写 DB,通过 Binlog 异步更新缓存
  2. Read Through: 读请求自动从缓存读取,缓存未命中则从 DB 加载并写入缓存
  3. Write Through: 写请求同时写缓存和 DB
  4. Write Behind: 写请求只写缓存,异步批量写 DB

12306 使用的是 Cache Aside + Canal 异步同步 方案。

Q2: 为什么不用同步双写?

参考答案:

  1. 同步双写会增加接口响应时间(需要等 Redis 操作完成)
  2. 如果 Redis 写入失败,需要处理回滚,增加复杂度
  3. Canal 异步方案解耦主流程,MQ 保证可靠送达
  4. 高并发场景下,同步双写可能成为性能瓶颈

Q3: 如何解决缓存穿透问题?

参考答案:

  1. 布隆过滤器: 在缓存层前加一层 BloomFilter,存在则查缓存,不存在则直接返回
  2. 缓存空值: 查询结果为空时,在缓存中设置空值 (带较短过期时间)
  3. 参数校验: 提前拦截无效请求

12306 使用的是 布隆过滤器 + 分布式锁 + 双检锁 的完整方案。

Q4: 如何解决缓存击穿问题?

参考答案:

  1. 分布式锁: 只允许一个线程查 DB,其他线程等待
  2. 热点数据永不过期: 定期异步更新
  3. 本地缓存: 使用 Caffeine 作为本地缓存热点数据

12306 使用的是 分布式锁 (Redisson) 方案。

Q5: 如何解决缓存雪崩问题?

参考答案:

  1. 过期时间随机化: 给缓存过期时间加随机值
  2. 多级缓存: 本地缓存 + 分布式缓存
  3. 过期前预热: 缓存快过期时异步更新
  4. 服务熔断降级: 缓存不可用时降级到 DB

Q6: Canal 工作原理?

参考答案:

  1. Canal 模拟 MySQL 主从复制协议,伪装成 MySQL 的从库
  2. MySQL 开启 Binlog,Canal 连接后请求获取 Binlog 事件
  3. Canal 解析 Binlog 事件,转换为 JSON 消息
  4. 发送到 Kafka/RocketMQ 等消息队列

Q7: 如果 MQ 消息丢失怎么办?

参考答案:

  1. RocketMQ 支持消息持久化
  2. 生产者确认机制 (事务消息)
  3. 消费者手动 ACK
  4. 开启幂等消费,即使重复消息也不会有问题

Q8: 如何保证消息消费的顺序性?

参考答案:

  1. RocketMQ 顺序消息: 指定 MessageQueue 和 ConsumerQueue
  2. 业务层面处理: 不依赖 MQ 顺序,在消费端根据 timestamp 或 version 处理
  3. 12306 的处理: 按订单号聚合,同一订单的消息路由到同一队列

总结

12306 的数据一致性架构


┌─────────────────────────────────────────────────────────────────┐
│                           写入流程                                │
│                                                                 │
│   业务操作 ──▶ 写 MySQL ──▶ 事务提交 ──▶ 返回成功                  │
│                                  │                              │
│                                  ▼                              │
│                           Binlog 解析                            │
│                                  │                              │
│                                  ▼                              │
│                           RocketMQ                              │
│                                  │                              │
│                                  ▼                              │
│                    多个 Handler 消费 ──▶ 更新 Redis               │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                           读取流程                                │
│                                                                 │
│   读请求 ──▶ Redis (命中) ──▶ 直接返回                           │
│                     │                                           │
│                     │ 未命中                                     │
│                     ▼                                           │
│               查 MySQL ──▶ 回填 Redis ──▶ 返回                   │
└─────────────────────────────────────────────────────────────────┘

核心技术点

技术 作用
Canal 解析 MySQL Binlog,模拟从库协议
RocketMQ 异步解耦,可靠消息传递
策略模式 根据表名路由到不同 Handler
幂等注解 防重复消费
HINCRBY 原子递增/递减
分布式锁 防缓存击穿

注意事项

  1. 不要同步双写: 影响性能,复杂易错
  2. 必须幂等消费: MQ 消息可能重复投递
  3. 使用原子操作: Redis 更新使用 HINCRBY,避免并发覆盖
  4. 监控延迟: Binlog 到 Redis 的延迟需要监控告警
Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐