一、先把场景说清楚：多源数据集成到底在解决什么问题

很多团队发展到一定阶段，都会遇到一个共同的烦恼：数据散落在好几个地方。订单在MySQL里，用户行为日志在MongoDB里，商品信息在PostgreSQL里，还有些数据在Oracle老系统里躺着。业务部门天天喊着要“全局视角”，老板要看实时报表，风控要跨库关联，搜索要近实时索引。这时候，把分散的数据统一同步到一个集中的存储（比如数据仓库、Elasticsearch、另一个MySQL实例）就成了刚需。

Canal是阿里巴巴开源的一个组件，核心能力是模拟MySQL主从复制的协议，把自己伪装成MySQL的从库，然后接收主库的binlog变更，从而拿到增删改的数据。它天生就是干这个活的。但你可能会问：一个Canal只能连一个MySQL，多源怎么办？

答案不是只靠Canal，而是靠一套架构设计。这篇文章就聊聊，当数据源头不止一个时，怎么用Canal做多源集成，以及多源同步里最让人头疼的“冲突”问题怎么解。

二、多源集成的架构设计：从“点对点”到“枢纽式”

2.1 最朴素的方案：一个源配一个Canal

刚开始接触Canal的人，最容易想到的架构是这样：MySQL1 -> Canal1 -> 目标库，MySQL2 -> Canal2 -> 目标库。每个Canal各管各的，互不干扰。这种“点对点”模式胜在简单，配置起来也快，适合数据源只有两三个、业务不复杂的场景。

但是一旦数据源多了，比如五个、八个，这种模式就会暴露出问题：你需要部署很多Canal实例，每台都要维护；目标端的连接数会暴涨；如果多个源里都有同一张业务表（比如不同分库里的订单表），你要在目标端搞多张表，业务查询时还得做union，这就很麻烦。

2.2 成熟的架构：源端多Canal，汇聚端统一处理

稍微正规一点的架构应该是这样的：每个MySQL源独立部署一个Canal实例，但Canal不会直接把数据写到最终目标，而是先写到一个中间层——消息队列（比如Kafka或RocketMQ）。然后再由一个统一的消费程序（或者直接用Canal的adapter）从消息队列里读数据，经过过滤、清洗、转换、合并之后，再写入目标存储。

这种架构的好处很明显：

  • 解耦：Canal只负责“搬运”，不关心目标端长什么样。
  • 削峰填谷：突发流量不怕，消息队列能缓冲。
  • 统一收口：所有源的数据都汇到同一个队列或者同一个consumer里，方便做全局处理。
  • 容错：某个Canal挂了，消息不会丢，拉起来继续消费就行。

下图就是这种架构的简化描述，你可以想象成：多个水龙头（数据源）都往一个蓄水池（消息队列）里放水，然后由一个水泵（消费程序）统一抽水到农田（目标库）。水龙头数量再多，水泵和农田那边只看到一股稳定的水流。

2.3 目标端表设计：多源合并的两种思路

多源数据的汇聚，最常见的是两种场景：

  • 场景A：不同业务的表，各归各。比如MySQL1有订单表，MySQL2有用户表，MySQL3有商品表。那目标端直接建三张表就行，互不影响。
  • 场景B：同一张业务表，拆在多库。比如订单按用户ID取模分库，order_db_0、order_db_1、order_db_2。这种情况下，我们希望在目标端只有一张“大订单表”，所有分库的数据都往里灌。

场景B是真正考验设计能力的地方。最简单的做法是，让每个Canal实例都把数据写到同一个Kafka topic，然后消费时根据表名和主键，用“upsert”的方式写入目标表。这时候就会遇到一个没法回避的问题：冲突。

三、多源同步冲突的本质和常见类型

说到冲突，先得明白冲突是怎么来的。多源同步里的冲突，本质上就是“两个源对同一条数据产生了不同的看法，但目标端只能保留一个结果”。

具体来说，常见的有这么几种情况：

3.1 主键冲突

最简单也最直白。两个不同的MySQL库，业务上都是订单表，订单ID都是自增的。结果库1里有一个订单ID=100，库2里也有一个订单ID=100，但它们分别代表不同的业务订单。如果直接照搬主键写入目标表，第二个肯定会撞车。

3.2 字段更新顺序冲突

同一业务主键的数据在两个库里都会更新。比如用户表，库1负责更新用户的昵称，库2负责更新用户的积分。当两个库几乎同时更新同一条记录时，目标端会收到两条消息：一条把昵称改为“小明”，另一条把积分改为“500”。如果消费程序处理这两条消息的顺序颠倒，就可能导致“旧数据覆盖新数据”的问题。

3.3 删除与更新冲突

库1里一条记录被删除了，库2里同一主键的记录还在更新。消费程序先处理删除，再处理更新，那目标端就会“死而复生”。或者反过来，先更新再删除，那倒是没问题，但你不能保证消息顺序永远正确。

3.4 结构性冲突

两个源里同一张表，字段A在库1里叫name，在库2里叫user_name；或者字段类型不一致，一个int一个string。这种冲突在集成阶段就得处理，否则消费端写数据时会报错。

四、冲突解决策略：实践中的几板斧

4.1 策略一：业务字段上的“全局唯一ID”

解决主键冲突最直接的思路是——别让两个源自己生成ID。我们在源端设计表时就预留一个全局唯一的业务ID，比如订单号用“全局发号器”（比如雪花算法）生成，而不是依赖数据库自增。这样即使两个分库的数据，其主键也不会重复。

如果源表已经用了自增ID，没法改呢？那就在目标端换一个“复合主键”：比如“source_id + original_id”。source_id用来标识数据来自哪个源，original_id是源库里的原始主键。这样至少不会因为主键撞车而写不进去。

我见过很多团队在实际中就是这么做的：目标表里加一个src_flag字段，值为“db1”或“db2”，然后主键设为(src_flag, id)。虽然看似多了一个字段，但解决了一大堆麻烦。

4.2 策略二：版本号（乐观锁）控制更新顺序

处理字段更新顺序冲突的标准做法，是引入“版本号机制”。源表增加一个version字段，每次更新时版本号+1。消费端拿到binlog里的数据后，比较消息里的版本号和目标端当前记录的版本号。只有消息版本大于目标端版本，才允许覆盖。

举个实际例子：库1更新了昵称，对应版本号是5；库2更新了积分，对应版本号是4。如果两条消息几乎同时到达，消费端先处理积分更新（版本4），发现目标端当前版本是3，那就更新；然后处理昵称更新（版本5），发现目标端版本已经是4了，5>4，继续更新。这样最终结果就是昵称为新，积分也为新。可如果反过来了，先处理的昵称（版本5），目标端从3跳到5；再处理积分（版本4），发现4<5，就丢弃。这样最终结果就是昵称新，积分旧——虽然丢了积分更新，但至少不会被旧昵称污染。比完全乱序好得多。

如果业务表里没有version字段，还可以用更新时间戳作为比较依据。但要注意，如果两个数据库主机的时钟不同步，时间戳比较就不可靠了。所以最稳妥的还是显式加版本号，由应用层或binlog里的事务ID参与决策。

4.3 策略三：按主键分桶路由，强制串行化

如果你觉得“版本号比较”逻辑很绕，还有一种更笨但更可靠的方法——让同一个主键的消息永远只被同一个消费者线程处理。在Kafka里，可以通过设置“按主键哈希分区”来实现。比如把消息的key设为目标主键（或复合主键），这样同一个key的消息会进入同一个分区，Kafka保证同一个分区内的消息是严格有序的。

这样一来，消费端逐个处理同一个主键的更新，就永远不会出现乱序。你只需要保证源端binlog发送到Kafka时，每个主键的变更消息确实是先后顺序产生的。Canal发送到Kafka时，如果你把table名加主键作为分区key，那这个顺序就能得到保证。

这种策略有个限制：消息吞吐量会被压制在单个分区的处理能力之内。但如果业务上同一个主键的更新频率不是极高，这完全是划算的。因为换来了“零冲突，无脑写”的省心体验。

4.4 策略四：删除延迟合并（墓碑机制）

删除与更新的冲突，通常用“墓碑”思路解决。做法是：当源端删除了某条数据时，Canal收到的binlog是一条DELETE语句，消费端不直接删除目标端记录，而是把目标端记录标记为“已删除”，比如增加一个deleted字段置为1，同时保留数据。

这样如果有后续的更新消息来了，消费端先看这个key对应的是不是“墓碑”，如果是，就不执行更新（因为源已经删了），或者根据业务需求把deleted字段重置回0并更新数据。但大多数情况下，删除就意味着“这条记录不该存在”，所以后续更新就直接丢弃。

等到一个确认窗口期（比如24小时）过后，如果这个主键没有再收到任何更新，消费端再真正把它从目标表里物理删除。这样可以避免“删除”和“更新”交叉带来的逻辑混乱。

4.5 策略五：结构冲突统一在ETL层转换

结构性冲突不在同步时的随机冲突，而是在规划阶段就要解决的。在消费端，可以做一个“字段映射配置”，把不同源的字段名统一到目标表的字段名。比如：

  • db1的name -> 目标表的user_name
  • db2的user_name -> 目标表的user_name

类型不一致的，比如db1的age是int，db2的age是varchar，在消费端做一次cast转换。这事儿不用搞得很复杂，写一个轻量级的转换脚本，在插入目标表前执行就好。

五、实战示例：一个完整的多源同步流程（技术栈：Java + Spring Boot + Canal + Kafka + MyBatis-Plus）

为了让你更直观地感受整个设计，下面给一个能落地的示例。场景是：两个MySQL库（order_db_1和order_db_2）都有一张t_order表，主键都是自增id，但业务上同一订单在两个库不会重复（实际上不会，但为了演示冲突解决，我们假设订单id可能撞车）。目标库是一个MongoDB的orders集合，但为了照顾大多数读者，我们用MySQL目标库，表结构加一个src_flag作为复合主键的一部分。

5.1 目标表结构

-- 目标库中的订单表，用于存储来自两个源的数据
CREATE TABLE t_order_center (
    src_flag        VARCHAR(10)     NOT NULL COMMENT '来源标识：db1或db2',
    order_id        BIGINT          NOT NULL COMMENT '源库中的原始订单ID',
    user_id         BIGINT          NOT NULL COMMENT '下单用户ID',
    amount          DECIMAL(10,2)   NOT NULL COMMENT '订单金额',
    status          VARCHAR(20)     NOT NULL COMMENT '订单状态',
    update_time     DATETIME        NOT NULL COMMENT '更新时间，用于冲突比较',
    PRIMARY KEY (src_flag, order_id)
) COMMENT='多源订单汇总表';

5.2 Canal配置（以db1为例，db2同理）

{
  "canal.instance.mysql.slaveId": 1001,
  "canal.instance.master.address": "192.168.1.101:3306",
  "canal.instance.dbUsername": "canal_reader",
  "canal.instance.dbPassword": "Canal@123",
  "canal.instance.connectionCharset": "UTF-8",
  "canal.instance.filter.regex": "order_db_1.t_order",
  "canal.mq.topic": "order_sync_topic",
  "canal.mq.partition.hash": "order_id",
  "canal.mq.partition.hash.algorithm": "murmur"
}

db2的配置，把IP换成192.168.1.102，库名换成order_db_2，slaveId改成1002，其余相同。

5.3 Kafka消费者统一处理

下面是一个Consumer的核心代码，包含主键冲突规避、时间戳冲突处理和字段映射，全部注释完整。

// 技术栈：Java + Spring Boot + Kafka + MyBatis-Plus
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Component
public class OrderSyncConsumer {

    private final OrderCenterMapper orderCenterMapper;

    public OrderSyncConsumer(OrderCenterMapper orderCenterMapper) {
        this.orderCenterMapper = orderCenterMapper;
    }

    /**
     * 监听Kafka中的binlog消息，所有源Canal发送的数据都进入同一个topic
     * @param record Kafka消息，value是Canal转发的JSON格式binlog数据
     */
    @KafkaListener(topics = "order_sync_topic", groupId = "order-sync-group")
    public void onMessage(ConsumerRecord<String, String> record) {
        JSONObject msg = JSONObject.parseObject(record.value());

        // 提取binlog中携带的原始数据
        String database = msg.getString("database");      // 源库名，如 order_db_1
        String table = msg.getString("table");            // 表名，这里固定为 t_order
        String eventType = msg.getString("type");         // INSERT / UPDATE / DELETE
        JSONObject data = msg.getJSONObject("data");      // 变更后的整行数据

        // 通过源库名生成来源标记
        String srcFlag = database.replace("order_db_", "db");

        // 原始订单ID
        Long orderId = data.getLong("id");

        // 用“srcFlag + orderId”作为目标端的业务主键
        OrderCenter order = new OrderCenter();
        order.setSrcFlag(srcFlag);
        order.setOrderId(orderId);

        if ("DELETE".equalsIgnoreCase(eventType)) {
            // 删除策略：先标记后物理删，这里用“墓碑法”简化，直接执行删除
            // 真实场景可改为标记字段延迟删除
            orderCenterMapper.deleteById(order);
            return;
        }

        // 以下是INSERT和UPDATE的处理，统一按“存在则比较时间，否则插入”来执行
        OrderCenter exist = orderCenterMapper.selectById(order);

        if (exist == null) {
            // 目标端无记录，直接插入
            OrderCenter fresh = buildNewRecord(srcFlag, data);
            orderCenterMapper.insert(fresh);
        } else {
            // 目标端已有记录，比较更新时间，只有新数据才覆盖
            LocalDateTime incomingTime = parseTime(data.getString("update_time"));
            LocalDateTime existingTime = exist.getUpdateTime();

            // 如果消息里的更新时间晚于或等于库里已有时间，才执行更新
            if (incomingTime.isAfter(existingTime) || incomingTime.equals(existingTime)) {
                OrderCenter fresh = buildNewRecord(srcFlag, data);
                fresh.setId(exist.getId()); // 保留主键ID做条件更新
                orderCenterMapper.updateById(fresh);
            }
            // 否则丢弃这条旧消息，不做任何操作
        }
    }

    /**
     * 根据Canal解析出的数据行，构造成目标表实体
     */
    private OrderCenter buildNewRecord(String srcFlag, JSONObject data) {
        OrderCenter order = new OrderCenter();
        order.setSrcFlag(srcFlag);
        order.setOrderId(data.getLong("id"));
        order.setUserId(data.getLong("user_id"));
        order.setAmount(data.getBigDecimal("amount"));
        order.setStatus(data.getString("status"));
        order.setUpdateTime(parseTime(data.getString("update_time")));
        return order;
    }

    /**
     * 将字符串时间解析为LocalDateTime，兼容各种格式
     */
    private LocalDateTime parseTime(String time) {
        if (time == null || time.isEmpty()) {
            return LocalDateTime.now();
        }
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        return LocalDateTime.parse(time, formatter);
    }
}

上面的代码里，最核心的就是“先查询是否存在，再比较更新时间”这段逻辑。这是一种典型的“在目标端做裁决”的方案，不需要源端配合，随时随地都能用。

但必须提醒你，这段代码在极端高并发下可能有问题：两个线程同时查到不存在，然后都执行insert，会导致目标端复合主键冲突。解决的办法有两个：

  • 把消费并发度调为1（同一个分区单线程消费），这样就天然串行。
  • 用数据库的“INSERT ... ON DUPLICATE KEY UPDATE”来原子化处理。

我更推荐你结合数据库特性做原子化，比如用MyBatis-Plus的自定义SQL：

-- 技术栈：MySQL原生SQL
INSERT INTO t_order_center
(src_flag, order_id, user_id, amount, status, update_time)
VALUES
(#{srcFlag}, #{orderId}, #{userId}, #{amount}, #{status}, #{updateTime})
ON DUPLICATE KEY UPDATE
user_id = IF(VALUES(update_time) >= update_time, VALUES(user_id), user_id),
amount  = IF(VALUES(update_time) >= update_time, VALUES(amount), amount),
status  = IF(VALUES(update_time) >= update_time, VALUES(status), status),
update_time = IF(VALUES(update_time) >= update_time, VALUES(update_time), update_time);

这种写法既处理了主键冲突，又通过update_time字段做了时间比较，一个语句解决两个问题，性能还高。

六、更多关联技术：你以为只有Canal？还有这些好搭档

6.1 Canal的另一个形态：Canal Adapter

如果你不想自己写消费者，Canal官方提供了Adapter组件，可以把他当成一个“现成的同步管道”。它的好处是内置了RDB（关系数据库）、ES、HBase等适配器，支持配置化同步，不需要写代码。

但是它的缺点是灵活性有限。如果你需要实现复杂的冲突解决逻辑（比如上面说的按源库优先级或版本号决策），还是自己写消费端更靠谱。所以我的建议是：场景简单且只需要单向同步，可以试试Adapter；场景复杂，还是自己做ETL吧。

6.2 Debezium：另一个多源同步选手

不少朋友会在选型时拿Debezium和Canal对比。Debezium也是干binlog同步这件事的，但它支持更多数据库（MySQL、PostgreSQL、SQL Server、MongoDB等），而且与Kafka Connect深度集成。如果你需要同时同步MySQL和PostgreSQL，那Canal就不太行了（Canal只支持MySQL），这时候可以选择Debezium。

但从国内团队使用习惯和运维成本来看，Canal依然很流行，因为不少团队的MySQL生态多，而且Canal的部署和管理相对轻量。

七、应用场景总结与注意事项

7.1 典型的应用场景

  • 从库路由：多个业务库共享给一个运营后台查询，需要做实时汇总。比如订单中心、用户中心、商品中心的数据合并。
  • 搜索索引构建：业务数据分散在多个MySQL，需要实时同步到Elasticsearch，供搜索和聚合分析使用。
  • 数据仓库近实时入仓：把多源MySQL数据同步到ClickHouse或Doris，替代T+1的传统同步。
  • 缓存重建：多源数据变化后同步到Redis或本地缓存，用于热点数据展示。

7.2 技术优缺点

先说优点：

  • 实时性好，毫秒级延迟，基本能做到秒级可见。
  • 对源库侵入性低，Canal模拟从库去拉binlog，不影响源库业务。
  • 扩展性强，多源并行部署，架构天然支持水平扩展。
  • 生态成熟，Kafka做缓冲，消费端可任意对接。

再说缺点：

  • Canal只支持MySQL和MariaDB，不能覆盖其他数据库源。
  • 部署运维有一定成本，尤其是Canal集群和Kafka集群需要监控。
  • binlog带来的数据量大，可能占带宽，如果业务更新频繁，需要评估带宽和存储。
  • 冲突解决方案需要业务方积极配合，很多情况下源表没有version字段，需要改建表或接受一定的数据不一致窗口。

7.3 实践中的注意事项

  • 不要直接依赖业务时间：如果系统时钟不同步，时间比较冲突策略会失效。尽量在binlog消息里带上Canal的executeTime，那是从MySQL事务提交时拿到的真实时间，比业务字段更可靠。
  • 监控Canal同步位点：Canal的消费位点如果长时间停滞，会导致目标端数据严重滞后。建议配置一个监控脚本，定期检查位点延迟，超阈值就告警。
  • 幂等性设计要贯穿始终：消费端宕机重启后可能重复消费Kafka消息。所有写入操作都要做到幂等，否则会导致数据重复或错乱。
  • 目标端要留审计字段：比如sync_time、src_flag、last_update_by。这样排查问题时有据可查。
  • 测试阶段务必模拟多源并发：不要光测单条同步，一定要构造两个库同时更新同一条记录、删除后更新、网络延迟导致乱序等场景，验证你的冲突策略真的有效。

八、文章总结

多源数据集成不是一个能“一把梭”的简单任务，它需要在架构层面先理清数据流，再在细节层面设计好冲突对策。

架构上，建议走“一源一Canal + 消息队列 + 统一消费端”的成熟路线，让各个环节各司其职，避免点对点的大量耦合。冲突处理上，没有银弹，通常要做组合拳：用复合主键或者全局ID解决主键冲突，用版本号或时间戳解决更新顺序冲突，用墓碑机制解决删除和更新交叉问题，再配合ETL层做字段映射和类型转换搞定结构冲突。

再提醒一遍，如果你的系统对数据一致性要求极其严格，建议从源头就设计好版本号字段和全局唯一ID，这比事后在同步层“擦屁股”要省心一百倍。多源同步的尽头不是“把数据搬过去”，而是“搬到对、不丢、不乱”。希望这篇文章能让你在搭建自己的多源同步体系时，少踩几个坑，思路更清晰。