在订单、商品等读多写少场景中，Redis缓存与MySQL数据库的一致性一直是一个比较棘手的问题。传统做法是在业务代码里先更新数据库，再删除或更新Redis，但无论是先删缓存还是后删缓存，都很难完全避开并发读写造成的脏数据。更稳妥的思路是把缓存更新动作从业务事务中剥离出来，让数据变更通过MySQL的binlog被独立链路捕获，再异步同步到Redis。Canal负责伪装成MySQL从库监听binlog，Kafka负责消息缓冲与解耦，Redis则作为最终同步目标，这条链路在稳定性、扩展性和业务零侵入方面都有明显优势。
与直接在业务代码中维护缓存相比，基于binlog的同步方案有一个关键区别：它感知的是数据库真正落盘的变更，而不是业务调用时的意图。即使某些更新来自后台脚本、数据订正任务或其他服务，也都能被Canal捕获，从而避免只覆盖主流程导致缓存漏更。Kafka的加入则进一步解决了单点消费能力有限的问题，当缓存同步任务积压时，可以通过增加消费者实例来提升吞吐量，同时也能利用Kafka的持久化能力在消费者宕机后继续消费。
一、为什么选择Canal+Kafka+Redis同步链路
Canal是阿里巴巴开源的一款MySQL binlog增量订阅与消费组件，它通过模拟MySQL从库的交互协议，向主库请求binlog事件。MySQL主库会在每次提交事务后把变更写入binlog，Canal拿到这些事件后解析出INSERT、UPDATE、DELETE操作以及变更前后的行数据，并封装成容易处理的结构。相比业务层手动发送消息，这种方式不需要修改业务代码，也不需要担心某些更新路径被遗漏。
引入Kafka的主要原因有三个。第一是缓冲能力，当Redis写入瞬时变慢或批量任务产生大量变更时，Kafka可以把消息先落盘，避免直接压垮消费者。第二是消费解耦，Canal只负责把解析结果投递到Kafka，至于缓存如何更新、是否需要额外加工，都由下游消费者决定，未来如果还要同步到Elasticsearch、搜索引擎或其他存储，也可以复用同一份Kafka数据。第三是扩展性，缓存同步任务通常可以按表或主键分区，Kafka天然支持多分区并行消费，为提升吞吐量提供了基础。
Redis在这个链路中扮演最终缓存角色。消费者收到Canal解析后的变更事件后，会判断操作类型并更新对应key。对于INSERT和UPDATE，通常可以重建缓存或更新缓存字段；对于DELETE，则需要删除缓存避免脏数据。由于链路是异步的，实际同步会存在几十毫秒到数百毫秒的延迟，因此这种方案更适合允许短暂不一致的读多写少业务。
二、Canal接入MySQL与Kafka配置详解
要让Canal把binlog变更投递到Kafka，需要同时调整Canal服务端配置和实例配置。首先在Canal的canal.properties中把服务模式从默认的TCP改为Kafka，并指定Kafka集群地址。下面是典型配置片段：
canal.serverMode = kafka canal.mq.servers = 127.0.0.1:9092 canal.mq.retries = 3 canal.mq.batchSize = 16384 canal.mq.maxRequestSize = 1048576 canal.mq.lingerMs = 5 canal.mq.bufferMemory = 33554432 canal.mq.canalBatchSize = 50 canal.mq.canalGetTimeout = 100 canal.mq.flatMessage = true canal.mq.compressionType = none canal.mq.acks = all
其中canal.serverMode = kafka表示开启Kafka投递模式，canal.mq.servers配置Kafka集群地址，canal.mq.acks = all表示等待所有副本确认，能提升消息可靠性。将canal.mq.flatMessage设置为true后，Canal会输出扁平的JSON消息，比原生protobuf结构更容易在消费者中解析。
实例配置通常位于conf/example/instance.properties或自定义实例目录中。这里需要填写MySQL主库地址、账号、需要监听的库表，以及投递到Kafka的topic和分区信息。下面是一个常用示例：
canal.instance.master.address = 127.0.0.1:3306 canal.instance.dbUsername = canal canal.instance.dbPassword = canal canal.instance.connectionCharset = UTF-8 canal.instance.filter.regex = .*\\..* canal.instance.filter.black.regex = mysql\..* canal.mq.topic = redis-cache-sync canal.mq.partition = 0 canal.mq.partitionsNum = 3 canal.mq.partitionHash = .*\\..*
需要特别留意canal.instance.filter.regex，它决定了哪些库表变更会被监听。生产环境中建议不要直接使用.*\\..*监听全部表，而是根据实际缓存范围精确配置，例如只监听订单库下的订单表和库存表，避免无关变更占用Kafka吞吐量。canal.mq.partitionHash用于按表名或主键进行哈希分区，如果希望保证同一行数据的变更顺序，可以配置主键级别的分区规则。
三、消费者解析Canal消息并更新Redis
开启flatMessage后，Canal投递到Kafka的JSON消息包含database、table、type、data和old等字段。type取值通常为INSERT、UPDATE、DELETE，data是变更后的行数据数组，old只在UPDATE中表示变更前的数据。消费者需要先解析这些字段，再根据表名和主键拼出Redis的key。
下面是一个使用Spring Kafka和fastjson解析Canal消息的Java消费示例。示例中根据操作类型执行不同的Redis操作：INSERT和UPDATE直接写入缓存，DELETE则删除缓存。为了简化，这里假设表结构固定，缓存key使用表名加主键拼接。
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.Map;
@Component
public class CanalRedisSyncConsumer {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@KafkaListener(topics = "redis-cache-sync", groupId = "redis-sync-group")
public void onMessage(String message) {
JSONObject json = JSON.parseObject(message);
String type = json.getString("type");
String database = json.getString("database");
String table = json.getString("table");
List<Map<String, Object>> dataList = json.getJSONArray("data").toJavaList(Map.class);
if ("INSERT".equals(type) || "UPDATE".equals(type)) {
for (Map<String, Object> row : dataList) {
String cacheKey = buildCacheKey(database, table, row);
stringRedisTemplate.opsForValue().set(cacheKey, JSON.toJSONString(row));
}
} else if ("DELETE".equals(type)) {
for (Map<String, Object> row : dataList) {
String cacheKey = buildCacheKey(database, table, row);
stringRedisTemplate.delete(cacheKey);
}
}
}
private String buildCacheKey(String database, String table, Map<String, Object> row) {
Object id = row.get("id");
return database + ":" + table + ":" + id;
}
}
实际项目中不建议把整行JSON直接写入Redis，因为Canal的data字段里可能包含不需要缓存的大字段，也会增加内存占用。更合理的方式是根据业务需要挑选字段，组装成精简的缓存对象。如果缓存结构不是简单字符串而是Hash，也可以使用opsForHash()只更新发生变化的字段，减少网络传输和Redis写入压力。对于UPDATE事件，还可以结合old字段判断哪些列真正发生了变化，只处理有效变更。
此外，消费者需要保证消息幂等。Kafka在重平衡或异常恢复时可能出现重复消费，而Redis的写入和删除本身就是幂等操作，因此上述逻辑天然具备幂等性。但如果业务中还有缓存续期、计数或缓存预热逻辑，就要额外设计幂等键或使用Redis原子操作，避免重复消费造成数据偏差。
四、一致性保障与失败补偿
异步链路无法做到数据库与缓存的强一致，只能追求最终一致。MySQL提交事务后，Canal需要读取binlog、投递Kafka、消费者再更新Redis，中间每一步都会引入延迟。要想把一致性问题控制在一定范围内，第一层保障是让Canal消费的binlog位置尽量靠近主库。Canal会定时记录位点，宕机恢复后从上次位置继续解析，避免丢失或重复大量事件。第二层保障是Kafka的持久化和ACK机制，配置acks=all可以确保消息不会因为Broker故障而丢失。
真正容易出问题的是消费者处理失败。比如Redis暂时不可用、网络抖动或数据格式异常，都可能导致某条消息处理失败。Kafka消费者可以配置手动提交偏移量，在处理成功后再提交，否则不提交并触发重试。Spring Kafka中可以把ackMode设置为MANUAL，然后调用acknowledgment.acknowledge()提交偏移。对于重试仍然失败的消息，可以将其发送到死信topic或写入本地补偿表，由定时任务扫描后重新同步缓存。
除了消费者失败，还要考虑业务上需要延迟处理的情况。例如数据库更新后马上有大量读请求，但Redis缓存刚刚被删除，此时请求会直接打到数据库。如果读压力很大，可以在删除缓存后再延迟双删一次，或者由消费者写入一个较短的临时缓存兜底。另一种常见策略是在更新数据库时同步更新缓存，同时保留Canal作为最终校验，一旦发现数据不一致就覆盖为正确值。这种方式虽然不能完全去除短暂不一致，但可以把影响范围压到最小。
五、顺序性、吞吐量与常见调优
顺序性是缓存同步中很容易被忽视的问题。如果同一行数据先执行UPDATE再执行DELETE，但Kafka消费时顺序颠倒，就可能出现先删缓存、后写入缓存，最终留下已经删除的数据。要避免这种问题，需要保证同一主键的变更进入同一个Kafka分区，并且消费者在每个分区内按顺序处理。Canal默认按表名进行分区，同一张表的所有变更通常进入同一个分区，也可以配置canal.mq.partitionHash按主键哈希，进一步把不同行的变更分散到不同分区，提升并行度。
吞吐量调优要从Canal、Kafka和消费者三个环节分别着手。Canal端可以适当调大canal.mq.batchSize和canal.mq.lingerMs，让消息批量投递，减少网络请求次数。Kafka端可以增加topic的分区数，但要注意分区数只能增加不能随意减少，设计时应预留一定余量。消费者端可以通过调整max.poll.records和并发度，让多个线程并行处理不同分区的消息。不过增加并发度时要确认Redis连接池大小是否足够，避免连接耗尽。
针对大事务或批量更新场景，Canal一次可能解析出非常多的行变更，导致单条Kafka消息体积过大。可以在Canal端限制单批行数，或者让消费者按行拆分处理并限制每批Redis写入数量。对于非常核心的缓存数据，还可以在消费者中增加本地内存缓冲，先聚合短时间内的多次更新，再批量写入Redis，从而降低频繁写入带来的网络和CPU开销。
综合来看，Canal+Kafka+Redis的同步方案适合对一致性要求不是极端严格、但数据变更频繁且读多写少的业务。它把缓存维护从业务代码中抽离出来，用binlog驱动同步，配合Kafka实现高吞吐和可扩展消费。只要在配置、消费逻辑和失败补偿上做好设计，就能获得一条稳定、可靠且易于维护的缓存同步链路。随着业务规模扩大，这套架构还可以平滑扩展消费者、增加topic分区，并接入更多下游系统。