Canal + Kafka 数据管道

概述

Canal 监听 MySQL Binlog，将实时数据变更投递到 Kafka/RocketMQ，下游消费者可以按需消费。这是构建实时数据管道的经典架构。

架构

MySQL Master → Binlog
      ↓
Canal Server
  ├── 解析 Binlog 为 Entry
  ├── 过滤：只关注特定库/表
  └── 投递：序列化后写入 MQ
       ↓
Kafka Topic：canal-order
  ├── ES 消费者：构建搜索引擎
  ├── Redis 消费者：更新缓存
  ├── 数仓消费者：同步数据湖
  └── 业务消费者：触发通知

一、Canal + Kafka 配置

1.1 Canal Server 配置

properties

# conf/canal.properties
# 模式：tcp 直连 → kafka 投递
canal.serverMode = kafka

# Kafka 配置
kafka.bootstrap.servers = 192.168.1.100:9092,192.168.1.101:9092
kafka.acks = all
kafka.compression.type = snappy
kafka.retries = 3
kafka.batch.size = 16384

# 投递分区规则：按表名 hash
canal.mq.dynamicTopic = canal-${dbName}-${tableName}
canal.mq.partitionsNum = 3

# FlatMessage：平铺 JSON 格式（下游消费更方便）
canal.mq.flatMessage = true

properties

# conf/example/instance.properties
# 监控的数据库
canal.instance.master.address = 192.168.1.10:3306
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal

# Binlog 解析位置
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =

# 过滤规则
canal.instance.filter.regex = order\\..*        # 只监控 order 库
canal.instance.filter.black.regex = order\\.t_log # 排除日志表

# 解析配置
canal.instance.parser.parallel = true
canal.instance.parser.parallelThreadSize = 4

1.2 数据格式

json

// FlatMessage 格式
{
  "type": "INSERT",               // INSERT / UPDATE / DELETE
  "database": "order",
  "table": "t_order",
  "es": 1712345678000,            // 执行时间
  "ts": 1712345678123,            // 投递时间
  "sql": "",                      // 原始 SQL（不开启则不填充）
  "pkNames": ["id"],
  "isDdl": false,
  "data": [                       // 变更后数据
    {
      "id": "1001",
      "order_no": "ORD20260724001",
      "user_id": "12345",
      "status": "PAID",
      "total_amount": "299.00",
      "create_time": "2026-07-24 10:00:00"
    }
  ],
  "old": [                        // 变更前数据（UPDATE 时）
    {
      "status": "PENDING"
    }
  ]
}

二、多消费者与并行消费

2.1 消费者配置

java

@Component
public class KafkaCanalConsumer {

    // 按不同的 group-id 实现多消费者独立消费
    // ES 索引更新
    @KafkaListener(topics = "canal-order-t_order",
                   groupId = "canal-es-consumer",
                   containerFactory = "kafkaListenerContainerFactory")
    public void consumeForES(String message, Acknowledgment ack) {
        CanalMessage msg = JSON.parseObject(message, CanalMessage.class);
        if ("INSERT".equals(msg.getType()) || "UPDATE".equals(msg.getType())) {
            // 写入 ES
            esTemplate.save(msg.getData().get(0), IndexCoordinates.of("order"));
        }
        ack.acknowledge();
    }

    // Redis 缓存更新
    @KafkaListener(topics = "canal-order-t_order",
                   groupId = "canal-redis-consumer")
    public void consumeForRedis(String message, Acknowledgment ack) {
        CanalMessage msg = JSON.parseObject(message, CanalMessage.class);
        String orderId = msg.getData().get(0).get("id").toString();
        // 删除缓存（延迟双删）
        redisTemplate.delete("order:" + orderId);
        // 延迟 500ms 再删一次
        scheduledExecutor.schedule(() ->
            redisTemplate.delete("order:" + orderId), 500, TimeUnit.MILLISECONDS);
        ack.acknowledge();
    }
}

2.2 手动 ACK + 并发消费

yaml

spring:
  kafka:
    consumer:
      enable-auto-commit: false                # 手动 ack
      auto-offset-reset: latest
      max-poll-records: 500                    # 每次拉取最大条数
      fetch-min-bytes: 1024
      fetch-max-wait: 100

    listener:
      ack-mode: manual                         # 手动确认
      concurrency: 3                           # 并发消费者数（= 分区数）
      missing-topics-fatal: false

三、顺序保证

3.1 分区策略

properties

# Canal 投递时保证同一行数据的顺序性

# 方案 1：按主键 hash 到同一分区
canal.mq.partitionsNum = 3
# 相同主键的数据 → 同一分区 → 顺序消费

# 方案 2：按业务字段分区
canal.mq.partitionHash = id
# 支持复合：orderNo#userId

# 消费端：1 个线程消费 1 个分区，天然保序
kafka.listener.concurrency = 3  # 分区数

3.2 幂等去重

java

@Component
public class IdempotentConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    public boolean consume(String key, String message) {
        // 基于 Binlog 位点去重
        // key: canal:offset:{table}:{id}:{binlog_file}:{binlog_position}
        String dedupKey = "canal:dedup:" + key;

        // SETNX：存在则不处理（已消费）
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(dedupKey, "1", Duration.ofMinutes(5));

        if (Boolean.TRUE.equals(success)) {
            // 首次消费，执行业务逻辑
            processMessage(message);
            return true;
        }
        // 已消费过，跳过
        log.debug("Duplicate message, skip: {}", key);
        return false;
    }
}

四、全量 + 增量同步

4.1 架构设计

全量同步                     增量同步
┌──────────────┐           ┌──────────────┐
│  DataX        │           │  Canal        │
│  (批量导出)    │           │  (实时 Binlog) │
└──────┬───────┘           └──────┬───────┘
       │                          │
       ▼                          ▼
┌─────────────────────────────────────┐
│              Kafka                   │
│  Topic: canal-order-full             │
│  Topic: canal-order-incremental      │
└─────────────────────────────────────┘
       │
       ▼
┌──────────────────────┐
│  全量+增量合并消费者     │
│  1. 先消费全量数据       │
│  2. 切换增量消费         │
│  3. 双跑验证一致性       │
└──────────────────────┘
       │
       ▼
┌──────────────┐
│  Elasticsearch  │
└─────────────────┘

4.2 全量同步实现

java

@Component
public class FullSyncJob {

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    private static final String TOPIC = "canal-order-full";

    // 分页扫描全量数据
    public void fullSync() {
        int pageSize = 1000;
        long maxId = 0;
        boolean hasMore = true;

        while (hasMore) {
            List<Order> orders = orderMapper.selectPage(maxId, pageSize);
            if (orders.isEmpty()) {
                hasMore = false;
                break;
            }

            for (Order order : orders) {
                // 装成 Canal FlatMessage 格式
                Map<String, Object> msg = new HashMap<>();
                msg.put("type", "INSERT");
                msg.put("database", "order");
                msg.put("table", "t_order");
                msg.put("data", List.of(order));

                kafkaTemplate.send(TOPIC, String.valueOf(order.getId()),
                    JSON.toJSONString(msg));
            }

            maxId = orders.get(orders.size() - 1).getId();
        }

        // 全量完成标记
        kafkaTemplate.send(TOPIC, "__COMPLETE__", "{\"type\":\"FULL_SYNC_COMPLETE\"}");
    }
}

4.3 增量切换

java

@Component
public class DataSyncOrchestrator {

    @Value("${sync.mode:incremental}")  // 启动模式：full / incremental
    private String syncMode;

    private volatile boolean fullSyncDone = false;

    @PostConstruct
    public void init() {
        if ("full".equals(syncMode)) {
            // 先执行全量
            fullSyncJob.fullSync();
            fullSyncDone = true;
            log.info("Full sync completed, switching to incremental");
        } else {
            fullSyncDone = true;  // 已有全量数据，直接增量
        }
    }

    // 增量消费者
    @KafkaListener(topics = "canal-order-t_order",
                   groupId = "canal-es-sync")
    public void onIncrementalMessage(String message) {
        if (!fullSyncDone) {
            log.warn("Full sync not done, skip incremental message");
            return;
        }
        // 处理增量数据
        processIncremental(message);
    }

    // 全量校验：全量 vs 增量数据核对
    @Scheduled(fixedDelay = 3600000)  // 每小时核对
    public void verifyConsistency() {
        // 1. 从 ES 查询总数
        long esCount = esTemplate.count(Query.findAll(), OrderIndex.class);
        // 2. 从 MySQL 查询总数
        long dbCount = orderMapper.countAll();
        // 3. 对比
        if (esCount != dbCount) {
            log.error("Data inconsistency: ES={}, DB={}", esCount, dbCount);
            alertService.sendAlert("数据不一致告警");
        }
    }
}

五、数据一致性核对

5.1 核对方案

java

@Component
public class DataConsistencyChecker {

    @Autowired
    private OrderMapper orderMapper;

    @Autowired
    private ElasticsearchRestTemplate esTemplate;

    // 逐行对比
    public void checkByBatch() {
        int pageSize = 1000;
        long offset = 0;
        int inconsistencyCount = 0;

        while (true) {
            List<Order> orders = orderMapper.selectPage(offset, pageSize);
            if (orders.isEmpty()) break;

            for (Order order : orders) {
                // 从 ES 查询
                OrderIndex esOrder = esTemplate.get(
                    String.valueOf(order.getId()), OrderIndex.class);

                if (esOrder == null) {
                    log.warn("Missing in ES: orderId={}", order.getId());
                    inconsistencyCount++;
                    continue;
                }

                // 对比关键字段
                if (!order.getStatus().equals(esOrder.getStatus())
                    || !order.getTotalAmount().equals(esOrder.getTotalAmount())) {
                    log.warn("Data mismatch: orderId={}, DB={}, ES={}",
                        order.getId(), order.getStatus(), esOrder.getStatus());
                    inconsistencyCount++;
                }
            }

            offset += pageSize;
        }

        // 汇总
        if (inconsistencyCount > 0) {
            alertService.sendAlert("数据核对完成，不一致数：" + inconsistencyCount);
        }
    }
}

六、总结

知识点	说明
Canal→Kafka	canal.serverMode=kafka，配置 dynamicTopic
FlatMessage	平铺 JSON，消费端解序列化
多消费者	不同 group-id 独立消费
并发消费	listener.concurrency = 分区数
顺序保证	主键 hash 分区 + 单分区顺序消费
幂等去重	Binlog 位点 SETNX 去重
全量+增量	DataX 全量 + Canal 增量双管道
一致性核对	定时逐行对比 MySQL vs ES

参考链接：
Canal Kafka 配置
Canal FlatMessage
Kafka 消费者配置

Canal + Kafka 数据管道 ​

概述 ​

架构 ​

一、Canal + Kafka 配置 ​

1.1 Canal Server 配置 ​

1.2 数据格式 ​

二、多消费者与并行消费 ​

2.1 消费者配置 ​

2.2 手动 ACK + 并发消费 ​

三、顺序保证 ​

3.1 分区策略 ​

3.2 幂等去重 ​

四、全量 + 增量同步 ​

4.1 架构设计 ​

4.2 全量同步实现 ​

4.3 增量切换 ​

五、数据一致性核对 ​

5.1 核对方案 ​

六、总结 ​

Canal + Kafka 数据管道

概述

架构

一、Canal + Kafka 配置

1.1 Canal Server 配置

1.2 数据格式

二、多消费者与并行消费

2.1 消费者配置

2.2 手动 ACK + 并发消费

三、顺序保证

3.1 分区策略

3.2 幂等去重

四、全量 + 增量同步

4.1 架构设计

4.2 全量同步实现

4.3 增量切换

五、数据一致性核对

5.1 核对方案

六、总结