企业官网建设流程全解析

接到一个数据同步需求：业务库里有几十张表，每天变更量很大，下游实时数仓要求按表订阅数据。第一反应就是拿Canal订阅MySQL binlog，丢到Kafka里。但真正动手之后才发现，问题根本不在“能不能接”，而在“表这么多，topic怎么办”。每个表都手动建一个Kafka topic根本不现实，新表上线还容易漏建，于是“从binlog接入kafka，动态创建topic写入数据”就成了这次折腾的核心。

这篇文章就是从这个真实场景出发，完整记录Canal接入Kafka的动态topic方案：涉及架构选型、Canal配置、Kafka端动态建topic、新表自动扩展以及一路踩过的坑。适合准备做MySQL实时同步、实时数仓入仓、缓存更新等场景的读者参考，也适合第一次接触Canal和Kafka联动的人照着做。

1. 项目概述与链路解析

1.1 为什么会被卡在“动态创建topic”

很多人的第一版方案其实是：Canal监听MySQL，然后写死一个Kafka topic，所有表的变更消息都往这个topic里塞。这种方案在小规模验证时没问题，但一旦到了生产环境就暴露三个问题。

首先是下游消费隔离问题。多张业务表混在同一个topic里，消费端必须先用database和table字段做过滤，再按表分发数据，逻辑绕，出现问题也不好追。其次是数据量和消费并行度的问题。多张表共享一个topic，topic的分区数是固定的，某一两张表流量特别大时，其他表也会跟着被拖慢。第三是新表上线的问题。业务每加一张表，如果用固定topic，下游逻辑几乎都要跟着改一遍；如果按表建topic，又需要有人提前把topic建好，漏一步数据就进不来。

所以真正要解决的是两个层面的问题：让Canal根据表名自动决定消息写到哪个topic，以及让Kafka在表第一次出现时自动把这个topic建出来。这两个动作串联起来，就是标题里说的“动态创建topic写入数据”。

1.2 数据链路是怎么走的

整套链路并不复杂，核心角色就三个：MySQL、Canal、Kafka。

MySQL binlog ↓ Canal Server（伪装成MySQL slave，拉取并解析binlog） ↓ Kafka Producer（按库表动态路由topic） ↓ Kafka Broker（topic不存在时自动创建，或由管理端提前创建） ↓ 下游消费者（按topic订阅各自关心的表数据）

Canal在这里扮演的角色是“搬运工”。它把自己伪装成MySQL的一个从库，通过binlog协议向主库请求增量日志，解析出每条数据变更后，再以Producer身份写入Kafka。写入Kafka时，topic不再是固定的，而是根据消息里的database和table字段动态计算出来的。

这条链路里最容易被忽略的一点是：Canal只管“路由”到哪个topic，但topic本身是否存在，其实是Kafka侧负责的。如果Kafka没开自动创建topic，或者没有提前在集群里预建好，那么Canal发送时会直接报错，数据就丢了。所以在设计方案时，必须把“动态路由”和“动态创建”当成两件事来分别处理。

1.3 动态topic方案的收益与代价

先讲收益，不然不值得折腾。

最直接的好处是表与topic一一对应，下游可以按需订阅。订单表的topic只包含订单变更，商品表的topic只包含商品变更，结构清晰，排查问题也方便。其次是新表接入成本大幅降低。只要在Canal的监听正则里覆盖到新表，新表数据进来时就能自动路由到对应topic，不再需要人工手工建topic。再就是消费并行度可以按表独立设置。流量大的表分区数给多一点，流量小的表分区数给少一点，互不干扰。

当然，这种方案也有代价。最大的代价就是topic数量会随着表数量线性增长，几十张表就有几十个topic，几百张表就有几百个topic，对Kafka集群的元数据管理和运维监控都会带来额外压力。另外，不同topic之间的数据量天然是不均匀的，有些表可能一天都没几条变化，有些表每秒几百条，如果全部用同一个分区策略，很容易出现某些分区热点。

这些代价不是不可接受，但需要在设计阶段就想清楚。我的建议是：动态topic适合表数量在几百张以内、但表与表之间业务隔离要求高的场景。如果表数量已经上千，或者下游根本不关心表隔离，那不如用中心topic + 分发层的方案，这也是后面会讲到的一种替代路径。

2. 动态创建topic的三种实现路径

2.1 路径一：纯靠Kafka自动创建topic

Kafka本身提供了一个开关：auto.create.topics.enable。生产者在发送消息时，如果发现topic不存在，broker会自动把topic创建出来。

配置方式是在Kafka的服务端配置里打开这个开关，并调整两个默认参数：

auto.create.topics.enable=true num.partitions=3 default.replication.factor=1

Canal把数据发到不存在的topic时，broker收到消息后会自动创建topic，分区数就是num.partitions指定的大小，副本数则是default.replication.factor。

这条路最明显的优点是省事，几乎不需要额外写代码。适合测试环境、临时同步任务、以及业务表结构经常变化的场景。

但生产环境我一般不推荐裸用。原因很实际：Kafka自动创建topic是“一刀切”的，所有表的topic分区数完全一样，副本数也无法单独指定。数据量大的表想多分几个分区，做不到；数据量小的表不想占用太多分区，也做不到。而且一旦新业务表上线，topic被自动创建后，后续很难再改分区数，因为分区数的增加会导致同一主键的数据被散落到不同分区，顺序性会被打破。

如果确实想省事，至少要把num.partitions调整到一个相对合理的值，并在后续配合一个巡检脚本，专门扫描那些自动创建的分区数不达标的topic。

2.2 路径二：Canal动态路由 + 启动预建topic

这条路径是目前生产环境用下来最稳的组合。

Canal的动态路由能力是通过canal.mq.dynamicTopic配置实现的。在1.1.4之后的版本中，这个配置可以接收正则表达式，命中正则的数据库表会按照规则自动生成对应的topic名称。最常用的配置是：

canal.mq.dynamicTopic=.*\\..*

这表示所有库的所有表都按“库名.表名”的方式动态路由到独立topic。订单库的order_info表对应topic名称就是shop.order_info，用户库的user表对应topic名称就是user.user。

配合dynamicTopic的是Kafka侧的topic预建。由于生产环境一般不希望Kafka自动创建一套不可控的topic，所以需要在Canal启动之前，把当前所有需要同步的表的topic用AdminClient批量建好。建topic的时候可以根据每张表的预估数据量，分别指定分区数和副本数。

这样组合之后的效果是：启动时已经有完整topic清单，Canal启动后所有历史表都能正常写入；未来新增的表则靠后面的新表监听机制或自动创建兜底。这条路径可控性高，也是我会在这篇里重点展开的方案。

2.3 路径三：绕开Canal路由，自研统一分发

第三种方案思路不太一样，它不让Canal直接按表路由，而是让Canal所有消息写入一个中心topic，比如canal.events，再在消费端写一个分发程序，解析消息里的database和table字段，用AdminClient动态创建真正的业务topic，然后把消息转发过去。

这种方案的优缺点都很明显。优点是所有消息先归一，Canal侧的配置极其简单，不需要依赖动态topic这种较新的特性；新表上线时，分发层能完全掌握建topic的时机和参数；还能在分发层做统一的数据过滤、格式清洗、限流和权限控制。缺点也清楚，多了一个分发环节，链路多一跳，延迟略有增加，中心topic的流量会非常大，分发程序必须保证高可用和顺序性。

我对这路方案的定位是：数据量不大但表特别多、表结构变化频繁、团队有足够精力维护中间分发服务的场景。数据量特别大时，中心topic会成为瓶颈，反而不如让Canal直接按表路由到各个topic。

2.4 三种路径对比

从我个人经验来看，大部分业务场景用第二种就够了，第三种更多是作为表数量极大时的治理方案。文章后续的内容也都围绕第二种方案展开。

3. 配置与实操：从binlog接入Kafka

3.1 MySQL侧的准备

Canal读取MySQL的binlog，前提是MySQL已经开启了binlog，并且binlog格式必须是ROW模式。

先登录MySQL检查当前状态：

SHOW VARIABLES LIKE 'log_bin'; SHOW VARIABLES LIKE 'binlog_format'; SHOW VARIABLES LIKE 'binlog_row_image';

如果log_bin是OFF，或者binlog_format不是ROW，需要修改MySQL配置文件并重启：

[mysqld] server-id=100 log-bin=mysql-bin binlog_format=ROW binlog_row_image=FULL

binlog_row_image设置为FULL，是为了让binlog里记录完整的行数据。如果设置成MINIMAL，binlog里只记录变更字段，Canal解析出的数据就不完整，下游拿到会缺字段。

接着创建Canal专用的数据库账号。这个账号只需要复制权限，不需要给超级权限：

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal_password'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;

其中REPLICATION SLAVE权限是让Canal可以冒充从库拉取binlog，REPLICATION CLIENT权限是让Canal查询master状态。如果漏了这两个权限，Canal启动日志里会报权限相关的错误。

还有一个容易踩的坑是server-id冲突。Canal在某个数据库实例上配置的canal.instance.mysql.slaveId不能和MySQL主库本身的server-id相同，也不能和该实例上其他从库的server-id相同，否则MySQL会报错或者出现复制中断。

3.2 Canal Server全局配置

Canal的部署方式这里不展开太多，直接说关键的配置文件。解压Canal发行包后，主要改两个文件：conf/canal.properties和conf/example/instance.properties。前者是Server级别的全局配置，后者是某个实例级别的配置，决定连接哪个MySQL、监听哪张表。

canal.properties里需要重点修改的部分：

canal.serverMode=kafka canal.mq.servers=192.168.1.10:9092,192.168.1.11:9092 canal.mq.retries=0 canal.mq.batchSize=2048 canal.mq.maxAliasSize=100 canal.mq.messageModel=batch canal.mq.dynamicTopic=.*\\..* canal.mq.flatMessage=true canal.mq.partitionHash=.*\\..*:pk

逐项说明一下：

• canal.serverMode=kafka：让Canal以Kafka作为输出目标，而不是默认的TCP。
• canal.mq.servers：Kafka broker的地址列表，多个用逗号分隔。
• canal.mq.batchSize：一次批次最多携带多少条变更记录。这个值直接影响Kafka单条消息的体积，后面会专门讲。
• canal.mq.dynamicTopic=.*\\..*：开启动态topic路由，所有库表都按“库名.表名”生成topic。
• canal.mq.flatMessage=true：消息以JSON格式输出。如果设成false，Canal会以protobuf二进制格式输出，消费端解析麻烦，不建议首次使用。
• canal.mq.partitionHash=.*\\..*:pk：按主键hash决定写入哪个分区，保证同一主键的数据变更进入同一个分区，维持单行数据的顺序性。

canal.mq.flatMessage=true还有一个附带好处，消息体里直接带上database和table字段，Kafka的消费端拿到JSON就能判断是哪个库哪张表的变更，调试和二次路由都非常方便。

3.3 instance实例配置

conf/example/instance.properties决定Canal连接哪一个MySQL实例：

canal.instance.mysql.slaveId=20 canal.instance.master.address=192.168.1.30:3306 canal.instance.dbUsername=canal canal.instance.dbPassword=canal_password canal.instance.connectionCharset=UTF-8 canal.instance.filter.regex=shop\\..* canal.instance.gtidon=false

这几个参数里，canal.instance.mysql.slaveId前面说过了，不能和MySQL的server-id重复。canal.instance.master.address是MySQL的主库地址。canal.instance.filter.regex表示监听哪些库表，shop\\..*表示监听shop库下所有表。

这里特别提醒：canal.instance.filter.regex非常重要。默认情况下Canal会监听所有库所有表，这包括MySQL系统库和中间表，生产环境很容易产生大量无意义的topic和消息。建议一定按实际业务库去收敛监听范围。

instance里一般不要再配置canal.mq.topic。因为canal.mq.topic是固定topic模式，如果instance层配了固定topic，会把全局dynamicTopic的配置覆盖掉，动态路由就不生效了。这个点非常隐蔽，我在实际项目里踩过一次，排查了半天才发现是两处配置冲突。

3.4 启动Canal并验证数据链路

配置文件改好后，启动Canal：

bin/startup.sh

然后观察两个日志文件：

tail -f logs/canal/canal.log tail -f logs/example/example.log

canal.log是Server的启动日志，能看到Canal与Kafka的连接状态。example.log是instance的日志，能看到Canal与MySQL的binlog连接状态。如果两个日志都没有异常，说明Canal已经在运行了。

接着做一次最简单的验证。在MySQL里执行一条INSERT：

INSERT INTO shop.order_info (id, user_id, amount) VALUES (1, 1001, 99.00);

然后用Kafka自带的消费命令看对应topic有没有消息：

bin/kafka-console-consumer.sh --bootstrap-server 192.168.1.10:9092 --topic "shop.order_info" --from-beginning

注意topic名称里带点号，命令行里一定要加引号，否则会被解析成非法topic名称。

正常的话，会看到类似这样的JSON消息：

{ "data": [ { "id": 1, "user_id": 1001, "amount": 99.00 } ], "old": null, "type": "INSERT", "database": "shop", "table": "order_info", "pkNames": ["id"], "isDdl": false, "es": 1700000000000, "ts": 1700000000000, "sql": "", "id": 123456 }

看到type":"INSERT"以及database、table字段都对得上，说明这一条binlog已经成功走通了MySQL -> Canal -> Kafka的完整链路，并且topic名称是按“库名.表名”动态路由的。

4. 动态创建topic的工程化落地

4.1 用AdminClient批量预建topic

上一节验证的是Canal端动态路由能力，但前面已经说过，topic本身是否存在，取决于Kafka那边怎么处理。生产环境下，我更推荐在Canal启动前先用AdminClient把现有表的topic预建好，而不是依赖Kafka的自动创建。

实现不复杂，核心工具类也就几十行：

import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.NewTopic; import java.util.Collections; import java.util.List; import java.util.Properties; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; public class KafkaTopicManager { private final AdminClient admin; public KafkaTopicManager(String bootstrapServers) { Properties props = new Properties(); props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers); props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 5000); this.admin = AdminClient.create(props); } public void createTopics(List<TopicMeta> topicMetaList) throws Exception { Set<String> existing = admin.listTopics().names().get(5, TimeUnit.SECONDS); List<NewTopic> needCreate = topicMetaList.stream() .filter(meta -> !existing.contains(meta.getName())) .map(meta -> new NewTopic(meta.getName(), meta.getPartitions(), meta.getReplicationFactor())) .collect(Collectors.toList()); if (!needCreate.isEmpty()) { admin.createTopics(needCreate).all().get(30, TimeUnit.SECONDS); } } }

细节上有个经验：判断topic是否存在，用listTopics比describeTopics更稳。describeTopics在topic不存在时会抛ExecutionException，不同Kafka版本异常类型还不一致，处理起来很别扭。listTopics返回的就是当前所有topic名称，一个contains判断就够了。

在Canal启动脚本里，先调用createTopics把初始表清单处理完，再启动Canal，这样绝大概率能避免“Canal启动后找不到topic导致同步失败”的第一波问题。

4.2 从MySQL表清单自动生成topic

topic的初始清单从哪来？最可靠的数据源是MySQL的information_schema.tables。写一个脚本，连上MySQL，拉出所有需要同步的表，去掉已经被Canal过滤掉的系统表，再按“库名.表名”拼成topic名称，调用Kafka AdminClient批量创建。

import pymysql from kafka.admin import KafkaAdminClient, NewTopic def load_sync_tables(mysql_conf, sync_schemas): conn = pymysql.connect( host=mysql_conf["host"], port=mysql_conf["port"], user=mysql_conf["user"], password=mysql_conf["password"], database="information_schema", charset="utf8mb4" ) placeholders = ",".join(["%s"] * len(sync_schemas)) sql = f""" SELECT table_schema, table_name FROM tables WHERE table_type = 'BASE TABLE' AND table_schema IN ({placeholders}) """ cursor = conn.cursor() cursor.execute(sql, tuple(sync_schemas)) return [f"{schema}.{table}" for schema, table in cursor.fetchall()] def create_topics(bootstrap_servers, topics, partitions=6, replication_factor=2): admin = KafkaAdminClient(bootstrap_servers=bootstrap_servers) existing = set(admin.list_topics()) need_create = [ NewTopic(topic, num_partitions=partitions, replication_factor=replication_factor) for topic in topics if topic not in existing ] if need_create: admin.create_topics(need_create, validate_only=False) if __name__ == "__main__": sync_tables = load_sync_tables( {"host": "192.168.1.30", "port": 3306, "user": "canal", "password": "canal_password"}, ["shop", "user"] ) create_topics(["192.168.1.10:9092"], sync_tables)

这个脚本是SQL和Kafka操作连在一起的，需要注意SQL里表名和schema名可能包含特殊字符，最好再处理一下，但核心逻辑就是这个。把脚本做成一个初始化动作，每次新接入一个业务库都执行一次，比手敲topic清单可靠得多。

这里还有个取舍问题：脚本里partitions=6、replication_factor=2是写死的，实际应根据表的预估数据量来调整。最简单的经验是：按主键hash分配分区时，分区数最好大于等于未来2-3年消费并行度的预期值，因为后面扩分区会改变主键hash到分区的映射关系，容易乱序。数据量小的表给3-6个分区就够了，数据量大的核心表可以给12个以上。

4.3 新表上线怎么自动建topic

预建脚本能覆盖存量表，但解决不了新表上线的问题。业务加了一张新表，如果它没在预建清单里，Canal的动态路由会把消息发到一个不存在的topic，这时候有两个处理方式。

最简单的兜底方案是打开Kafka的自动创建topic开关，并且把默认分区数调大一点。这样Canal发送新表消息时，broker会自动建topic，消息不会丢。但代价是自动建出来的topic分区数可能不理想。

更工程化一点的方案，是在消费端做一个DDL监听。Canal的flatMessage里，DDL消息长这样：

{ "type": "CREATE", "database": "shop", "table": "product_new", "isDdl": true, "sql": "CREATE TABLE `product_new` (...) " }

消费端判断如果是isDdl=true且type=CREATE，就调用前面写的AdminClient方法，按规范创建这个表对应的topic。

if (message.isDdl()) { if ("CREATE".equals(message.getType())) { String topic = message.getDatabase() + "." + message.getTable(); KafkaTopicManager.createTopicIfAbsent(topic, 6, (short) 2); } }

不过这方案有个先有鸡还是先有蛋的问题：新表的DDL消息本身要进哪个topic？如果Canal已经按“库名.表名”动态路由，新表还没有topic，DDL消息也发送不过去。所以在实际落地时，我见过的主流做法是下面这样的折中：

• 存量表启动时预建topic。
• 新表上线靠Kafka自动创建topic兜底。
• 再写一个定时巡检脚本，检测自动创建的topic，发现分区数不符合规范就自动扩容或重新规划。

用定时任务去“修正”自动创建的topic，虽然不够优雅，但胜在稳定。自动创建的topic至少保证了消息不丢，巡检脚本则负责逐步把topic收敛到标准状态，两者配合，新表接入基本不需要人工干预。

4.4 顺序性和分区数之间的关系

动态topic方案里，顺序性问题很容易被忽视。

Canal读取binlog时，同一张表的数据变更是有序的。写入Kafka后，如果这张表的topic只有一个分区，那消费者在分区内读到的数据依然有序。但topic有多个分区时，Kafka只在分区内部保证顺序，跨分区是没有顺序保证的。

所以必须配置好canal.mq.partitionHash。配置成.*\\..*:pk之后，Canal会按照消息的主键值做hash，同一个主键的数据变更永远落到同一个分区。这样即使整张表有多个分区，对同一行数据的多次更新，消费端看到的顺序一定是对的。

这里有一个必须记住的限制：分区数一旦确定，扩容之后，同一个主键的hash结果和原来分区映射大概率会变，顺序性就会受到影响。因此建topic时，分区数尽量一步到位，不要轻易扩容。如果确实需要扩容，要接受“扩容后同一行的历史数据与新数据可能落在不同分区”的现实，消费端就得设计成能容忍乱序或者做合并处理。

5. 常见问题与排查经验

5.1 动态topic不生效，消息全部进同一个topic

遇到这种问题，先检查两处。

第一，确认Canal版本。canal.mq.dynamicTopic是在1.1.4之后的版本里支持的，如果用的是1.1.3或更老的版本，这个配置根本不生效。直接把Canal升级到1.1.7或更新版本即可。

第二，检查instance层是否配置了固定topic。instance.properties里的canal.mq.topic优先级很高，如果管理员为了测试，在instance层写死了一个topic，那么即使全局配置了dynamicTopic，消息也只会往这个固定topic里发。解决办法是删掉instance层的canal.mq.topic配置，重启Canal。

还有一个容易看走眼的地方：正则没写对。canal.mq.dynamicTopic=.*\\..*在properties文件里必须保留双反斜杠，如果只写一个，正则解析时点号变成通配，匹配结果容易超出预期，动态topic的分隔符也可能不对。

5.2 数据到了Kafka但消费者查不到

消息确实进来了，但消费者查不到，最常见的原因是topic名称带点号，命令行工具或者消费代码里没有正确处理。

看一个典型操作：

kafka-console-consumer.sh --bootstrap-server 192.168.1.10:9092 --topic shop.order_info --from-beginning

这个命令在大多数情况下会报错或者一直空等，因为topic名里的点号在Kafka工具的参数解析里是合法的，但有时候会跟你敲命令时用了空格或引号不一致有关。我在生产环境排查过，最后发现消费者那边topic名称拼错了，少了个库名前缀。这类问题别看简单，排查起来真的会耗掉一两个小时。

另外，确认消费不到消息时，用下面这个命令看下topic的offset情况：

kafka-get-offsets.sh --bootstrap-server 192.168.1.10:9092 --topic "shop.order_info"

如果offset有值，说明消息确实写进去了，问题在消费者自己那边。如果offset是0，说明消息根本没到topic里，这时候回Kafka服务端看Canal发送日志。

5.3 同一张表的数据乱序

如果已经配置了canal.mq.partitionHash=.*\\..*:pk，但消费端还是出现乱序，那么先看一下topic实际的分区数是多少。

分区数大于1，且同一主键的消息被分到不同分区，才会乱序。造成这一情况常见原因有两个：

一个是Canal版本较老，对partitionHash的支持不完整，建议升级版本。另一个是topic被Kafka自动创建了，分区数默认是num.partitions的值，比如3，而你的消费者只有一个分区在消费，理论上不会乱序，但如果消费者并发线程自己处理乱了，那是消费端的问题。真正要确认的是，消息的key是否按照主键生成。可以用Kafka客户端工具把partition信息打出来，看同一主键的消息是否都在同一分区。

如果不在同一分区，排查Canal侧的分区器配置有没有真正生效。确认canal.mq.partitionHash和canal.mq.flatMessage=true都配置后，重启Canal再测一次。

5.4 topic数量暴涨和分区热点

动态topic方案跑起来后，topic数量会一直增加，这是必然的。如果不管好，最后Kafka集群里可能有大量垃圾topic。

我吃过的教训是：Canal的filter.regex一定要收敛。默认.*\\..*会把MySQL里所有库所有表全部监听，包括那些临时表、中间表、归档表。这些表一变，自动建的topic也越来越多，Kafka的元数据压力变大，zk或KRaft里的topic列表也变长，整个集群的扇出都受影响。

控制手段有几个：

• 明确Canal监听范围，只监听真正的业务库。
• 给topic命名前加统一前缀，比如canal.shop.order_info，方便区分和清理。
• 对长时间没有消息写入的topic，建立一套下线流程，超过一定时间自动删除。
• 核心表和非核心表分开集群，或者分开Canal实例，避免互相干扰。

动态topic不是“建了就不管”，它需要一套配套的生命周期管理机制。否则到最后，topic数量会变成新的运维灾难。

从我个人实际使用感受来说，Canal动态路由+预建topic这套组合，已经能覆盖90%的常规业务同步场景。真正在工程上花时间的，是对新表的平滑接入和topic生命周期的管理。如果让我再优化一次，我会把重点放在巡检脚本和DDL监听上，把新表自动建topic做成一个更规范、更自动化的流程，而不是每次都靠人工去判断和处理。这套链路后续还可以按照业务类型拆分Canal实例，比如订单库一个实例、用户库一个实例，每个实例的topic路由策略独立，运维边界也更清晰。