mysql使用Canal监听binlog日志，同步数据到MySQL、ElasticSearch、Redis、Kafka、RocketMQ和es
更新时间：2026年03月14日 11:58:34 作者：fixAllenSun
Canal是阿里巴巴开源的一款基于MySQL数据库binlog的增量数据订阅和消费组件,canal.adapter,相当于canal的客户端,会从canal-deployer中获取数据,模拟MySQL slave协议解析binlog,捕获数据变更事件,通过客户端可以同步到MySQL、Elasticsearch和HBase等存储中去
【一】安装Canal
Canal [kə'næl] 是阿里巴巴开源的一款基于MySQL数据库binlog的增量数据订阅和消费组件。
MySQL Master → binlog → Canal伪装Slave → 解析binlog → 数据分发
伪装机制：Canal模拟MySQL slave的交互协议，向MySQL master发送dump请求
解析转换：接收binlog并解析成JSON/AVRO等格式
订阅分发：客户端订阅消费解析后的增量数据
【1】配置mysql
（1）mysql开启binlog
修改 MySQL 配置文件my.cnf，重启 MySQL
cd /root/NFTurbo_DockerCompose/mysql/config vi my.cnf
开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下:
[mysqld] log-bin=mysql-bin # 开启 binlog binlog-format=ROW # 选择 ROW 模式（Canal只支持ROW） server_id=1 # 配置 MySQL replaction 需要定义，主库ID，不要和 canal 的 slaveId 重复
binlog的路径：/var/log/mysql/bin-log
重启 MySQL
（2）授权
授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant
CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES;
（3）检查binlog是否已开启
SHOW VARIABLES LIKE 'log_bin%'; -- Value为ON即成功 SHOW VARIABLES LIKE 'binlog_format'; -- Value为ROW即成功
重启mysql
【2】安装Canal-deployer（接收数据）
（1）Canal-deployer下载
用的Canal-deployer是1.1.7版本。
下载地址：https://github.com/alibaba/canal/releases
下载canal.deployer，可以直接监听MySQL的binlog，把自己伪装成MySQL的从库，只负责接收数据，并不做处理。
（2）解压
（3）配置修改
修改canal.properties，增加
canal.ip = 127.0.0.1
cd /home/allensun/develop/canal/canal_deployer/conf vim canal.properties
（4）修改example/instance.properties
（1）配置当前服务器的ip和端口
（2）配置监听mysql的用户和密码
cd /home/allensun/develop/canal/canal_deployer/conf/example vi instance.properties
（5）启动 Canal
到 bin 目录下，执行命令：./startup.sh
cd /home/allensun/develop/canal/canal_deployer/bin && ./stop.sh && ./startup.sh
有报错
把jdk安装上，再执行
（6）启动成功检查
启动成功，检查canal/logs/canal.log
看到the canal server is running now …表示启动成功。
tail -f /home/allensun/develop/canal/canal_deployer/logs/canal/canal.log
（7）查看主库状态的时候有报错
tail -f /home/allensun/develop/canal/canal_deployer/logs/example/example.log
SHOW MASTER STATUS;语句报错了，因为这是mysql8.0的语法，用来查看主库状态信息，mysql8.4应该使用
SHOW BINARY LOG STATUS;
【3】安装Canal-adapter（处理数据）
（1）下载canal.adapter
同样是使用1.1.7版本，和deployer 保持一致。
canal.adapter，相当于canal的客户端，会从canal-deployer中获取数据，然后对数据进行同步，可以同步到MySQL、Elasticsearch和HBase等存储中去。
到https://github.com/alibaba/canal/releases 下载
（2）解压
Canal-Adapter 目录结构（默认）：
canal-adapter/ ├── conf/ │ ├── application.yml # 全局核心配置 │ └── rdb/ # MySQL同步专用目录（放YAML映射文件） │ └── mysql_sync.yml # 自定义表映射YAML（名字随意） └── bin/startup.sh # 启动脚本
（3）全局核心配置：application.yml
全局配置：application.yml 负责连 Canal-Server + 目标 MySQL 从库
server:
port: 8081 # REST API 端口
tomcat:
max-threads: 1 # 最大线程数
min-spare-threads: 1 # 最小空闲线程数
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
default-property-inclusion: non_null
canal.conf:
mode: tcp # 客户端模式: tcp/kafka/rocketMQ
canalServerHost: 127.0.0.1:11111 # Canal-Server地址
# Canal-Server连接参数
batchSize: 500 # 每次获取数据量
syncBatchSize: 1000 # 同步批次提交量
retries: 3 # 重试次数
timeout: 1000 # 超时时间(ms)
vhost: / # 虚拟主机
filter: .*\\..* # 正则表达式过滤表
# 源数据库配置
srcDataSources:
defaultDS: # 数据源key，用于映射文件引用
url: jdbc:mysql://127.0.0.1:3306/source_db?useUnicode=true&characterEncoding=UTF-8&useSSL=false
username: root
password: 123456
lazy: false
canalAdapters: # 适配器配置
- instance: example # Canal实例名，与Canal-Server配置一致
groups:
- groupId: g1 # 分组ID
outerAdapters: # 目标数据源适配器
- name: es # Elasticsearch适配器
hosts: 127.0.0.1:9200 # ES地址
properties:
mode: rest # 连接模式: rest/transport
cluster.name: my-es # 集群名
- name: redis # Redis适配器
hosts: 127.0.0.1:6379
properties:
mode: single # 模式: single/sentinel/cluster
database: 0
- name: rdb # 关系型数据库适配器(MySQL)
key: mysql1 # 适配器key
properties:
jdbc.driverClassName: com.mysql.cj.jdbc.Driver
jdbc.url: jdbc:mysql://127.0.0.1:3306/target_db?useUnicode=true
jdbc.username: root
jdbc.password: 123456
- name: kafka # Kafka适配器
properties:
bootstrap.servers: 127.0.0.1:9092
acks: 1
- name: rocketMQ # RocketMQ适配器
properties:
namesrv.addr: 127.0.0.1:9876
producer.group: canal-producer
# 监控配置
monitor:
enable: true
port: 11112
user: admin
passwd: admin
（4）表映射配置：conf/xxx/xxx.yml
表映射：conf/rdb/*.yml 负责主表→从表的字段映射
# -------------------------- 固定全局配置 --------------------------
dataSourceKey: adapter # 必须 = application.yml 中的 spring.datasource.adapter.key
destination: example # 必须 = Canal 实例名
group: g1 # 自定义分组，随便写
outerAdapterKey: rdb # 固定：rdb
# -------------------------- 表映射规则（核心） --------------------------
tables:
# 表 1：user 表同步
- tableName: user # 自定义标识，随便写
sourceDb: source_db # 主库库名
sourceTable: user # 主表名
targetDb: target_db # 从库库名
targetTable: user # 从表名
mapAll: true # 自动映射所有字段（主从结构一致必须开）
pk: id # 主键（不写自动识别）
# 表 2：product 表同步
- tableName: product
sourceDb: source_db
sourceTable: product
targetDb: target_db
targetTable: product
mapAll: true
# 表 3：order 表同步
- tableName: order
sourceDb: source_db
sourceTable: order
targetDb: target_db
targetTable: order
mapAll: true
两个 YAML 必须严格对应（错一个就失败）
（5）启动
# 进入Canal-Adapter根目录 cd /your/path/canal-adapter # 启动（Linux/Mac） sh bin/startup.sh # 停止 sh bin/stop.sh
验证启动成功
# 查看日志（出现以下关键字即成功） tail -f logs/adapter/adapter.log
✅ 成功日志：
Start to subscribe canal destination: example Rdb adapter started successfully All tables have been loaded
【4】工作原理
（1）MySQL主备复制原理
MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据
（2）canal 工作原理
canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)
【二】同步数据到MySQL、ElasticSearch、Redis、Kafka、RocketMQ
【1】前置准备
Canal-Server 已正常启动，且监听 MySQL 主库 binlog
MySQL从库已创建好和主库结构一致的目标表
Canal-Adapter 目录下：删除原application.properties，新建application.yml（避免配置冲突）
表映射 YAML 必须放在 conf/rdb/ 目录下（RDB 适配器固定目录）
使用 Canal-Adapter 实现实时数据同步，核心是2 个配置文件：
（1）全局配置：conf/application.properties（也可以是conf/application.yml，连接 Canal-Server、配置同步目标、全局参数）
（2）表映射配置：conf/xxx/xxx.yml（YAML 文件，指定哪张主表同步到哪张从表、字段映射规则）
关键：Canal-Adapter 同步到 MySQL 属于RDB（关系型数据库）适配器，YAML 文件必须放在 conf/rdb/ 目录下！
MySQL Binlog ↓ Canal-Server ↓ Canal-Adapter（一套配置同步 5 个目的地） ├─ → MySQL 从库（rdb） ├─ → ElasticSearch（es） ├─ → Redis（redis） ├─ → Kafka（kafka） └─ → RocketMQ（rocketmq）
【2】全局核心配置 conf/application.yml
作用：连接 Canal-Server + 配置目标 MySQL 从库数据源 + 全局同步参数
server:
port: 8081 # REST API 端口
tomcat:
max-threads: 1 # 最大线程数
min-spare-threads: 1 # 最小空闲线程数
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
default-property-inclusion: non_null
canal.conf:
mode: tcp # 客户端模式: tcp/kafka/rocketMQ
canalServerHost: 127.0.0.1:11111 # Canal-Server地址
# Canal-Server连接参数
batchSize: 500 # 每次获取数据量
syncBatchSize: 1000 # 同步批次提交量
retries: 3 # 重试次数
timeout: 1000 # 超时时间(ms)
vhost: / # 虚拟主机
filter: .*\\..* # 正则表达式过滤表
# 源数据库配置
srcDataSources:
defaultDS: # 数据源key，用于映射文件引用
url: jdbc:mysql://127.0.0.1:3306/source_db?useUnicode=true&characterEncoding=UTF-8&useSSL=false
username: root
password: 123456
lazy: false
canalAdapters: # 适配器配置
- instance: example # Canal实例名，与Canal-Server配置一致
groups:
- groupId: g1 # 分组ID
outerAdapters: # 目标数据源适配器
- name: es # Elasticsearch适配器
hosts: 127.0.0.1:9200 # ES地址
properties:
mode: rest # 连接模式: rest/transport
cluster.name: my-es # 集群名
- name: redis # Redis适配器
hosts: 127.0.0.1:6379
properties:
mode: single # 模式: single/sentinel/cluster
database: 0
- name: rdb # 关系型数据库适配器(MySQL)
key: mysql1 # 适配器key
properties:
jdbc.driverClassName: com.mysql.cj.jdbc.Driver
jdbc.url: jdbc:mysql://127.0.0.1:3306/target_db?useUnicode=true
jdbc.username: root
jdbc.password: 123456
- name: kafka # Kafka适配器
properties:
bootstrap.servers: 127.0.0.1:9092
acks: 1
- name: rocketMQ # RocketMQ适配器
properties:
namesrv.addr: 127.0.0.1:9876
producer.group: canal-producer
# 监控配置
monitor:
enable: true
port: 11112
user: admin
passwd: admin
【3】表映射配置 conf/rdb/mysql_sync.yml
作用：定义「主库哪张表 → 目标数据源」的映射规则（支持多表同步）
⚠️ YAML 严格缩进 2 空格，不能用 Tab！
5 套映射文件
（1）同步到 MySQL 从库：conf/rdb/mysql.yml
# ====================== 表映射核心配置（必改：库名、表名） ======================
# 1. 固定：和application.yml的spring.datasource.adapter.key一致
dataSourceKey: defaultDS
# 2. 固定：和application.yml的canal.conf.destinations一致
destination: example
# 3. 分组（自定义，默认即可）
group: g1
# 4. 适配器类型（固定rdb）
outerAdapterKey: mysql1
# 5. 表映射规则（核心：主表→从表，支持多表）
tables:
# 表1：主库user表 → 从库user表
- tableName: user # 自定义标识（随意）
sourceDb: your_source_db # 【需修改】主库库名
sourceTable: user # 【需修改】主库表名
targetDb: your_target_db # 【需修改】从库库名
targetTable: user # 【需修改】从库表名
mapAll: true # 全字段自动映射（主从表结构一致必开）
# 表2：主库product表 → 从库product表
- tableName: product
sourceDb: your_source_db
sourceTable: product
targetDb: your_target_db
targetTable: product
mapAll: true
# 表3：主库order表 → 从库order表
- tableName: order
sourceDb: your_source_db
sourceTable: order
targetDb: your_target_db
targetTable: order
mapAll: true
（2）同步到 ElasticSearch：conf/es/user.yml
dataSourceKey: defaultDS # 对应application.yml中的源数据源key
destination: example # Canal实例名
groupId: g1 # 分组ID
esMapping:
_index: user_index # ES索引名
_type: _doc # ES类型（7.x后固定为_doc）
_id: 'id' # 文档ID字段，支持EL表达式: {id}
upsert: true # 启用upsert（存在则更新，不存在则插入）
sql: "SELECT # 查询SQL
id,
name,
age,
email,
DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%s') as create_time,
CASE status
WHEN 1 THEN 'active'
WHEN 0 THEN 'inactive'
ELSE 'unknown'
END as status_desc
FROM user
WHERE id > 0"
etlCondition: "where create_time >= '{}'" # ETL条件，{}会被参数替换
commitBatch: 1000 # 批量提交大小
pk: id # 主键字段，用于增量判断
skips: # 跳过某些操作类型
- INSERT
- UPDATE
- DELETE
（3）同步到 Redis：conf/redis/user.yml
dataSourceKey: defaultDS
destination: example
groupId: g1
key: 'user:{id}' # Redis key模板，支持{field}占位符
redisMapping:
database: 0 # Redis数据库编号
table: user # 源表名
expireTime: 86400 # 过期时间(秒)，0表示永不过期
pk: # 主键映射
id: id
type: string # Redis数据类型: string/hash/list/set/zset
columns: # 字段映射
id: id
name: name
age: age
email: email
sql: "SELECT id, name, age, email, status FROM user WHERE status = 1"
operationType: ALL # 操作类型: ALL/INSERT/UPDATE/DELETE
（4）同步到 Kafka：conf/kafka/user.yml
（1）application.yml Kafka专用配置
canal.conf:
mode: kafka # 必须设置为kafka模式
mqServers: 127.0.0.1:9092 # Kafka地址
flatMessage: true # 是否使用扁平化消息格式
parallelThreadSize: 8 # 并行处理线程数
kafka:
enable: true
bootstrap.servers: 127.0.0.1:9092
acks: 1
compression.type: snappy
batch.size: 16384
linger.ms: 1
（2）映射文件：conf/kafka/user.yml
dataSourceKey: defaultDS
destination: example
groupId: g1
topic: canal.user.topic # Kafka topic
partition: 0 # 分区号，-1表示自动分配
partitionHash: # 分区哈希规则
- id
tableMapping: # 表映射规则
database: source_db # 数据库名
table: user # 表名
pk: # 主键字段
id: id
sql: "SELECT id, name, age, email FROM user"
# 消息格式配置
messageFormat: json # 消息格式: json/protobuf
includeFields: # 包含的字段
- id
- name
- age
excludeFields: # 排除的字段
- password
（5）同步到 RocketMQ：conf/rocketmq/user.yml
（1）application.yml RocketMQ专用配置
canal.conf:
mode: rocketMQ # 必须设置为rocketMQ模式
mqServers: 127.0.0.1:9876 # NameServer地址
flatMessage: true
parallelThreadSize: 8
rocketMQ:
enable: true
namesrv.addr: 127.0.0.1:9876
producer.group: canal-producer
enable.message.trace: false
tag: canal # 默认tag
（2）映射文件：conf/rocketmq/user.yml
dataSourceKey: defaultDS
destination: example
groupId: g1
topic: CANAL_USER_TOPIC # RocketMQ topic
tag: user_tag # 消息tag
key: '{id}' # 消息key
tableMapping:
database: source_db
table: user
index: 1 # 顺序索引
commitBatch: 100 # 批量提交大小
pk:
id: id
sql: "SELECT
id,
name,
age,
email,
create_time,
update_time
FROM user"
# 消息过滤条件
filterCondition: "status = 1" # 只同步status=1的记录
# 字段转换
fieldMappings:
id: userId
name: userName
# 消息属性
properties:
retryTimes: 3
timeout: 3000
【三】同步数据到es
【1】全局核心配置 conf/application.yml
修改 application.properties：
cd /root/canal_adapter/conf vi application.yml
server:
port: 8081
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
default-property-inclusion: non_null
canal.conf:
mode: tcp #tcp kafka rocketMQ rabbitMQ
flatMessage: true
zookeeperHosts:
syncBatchSize: 1000
retries: -1
timeout:
accessKey:
secretKey:
consumerProperties:
# canal tcp consumer
canal.tcp.server.host: 127.0.0.1:11111
canal.tcp.zookeeper.hosts:
canal.tcp.batch.size: 500
canal.tcp.username:
canal.tcp.password:
# kafka consumer
kafka.bootstrap.servers: 127.0.0.1:9092
kafka.enable.auto.commit: false
kafka.auto.commit.interval.ms: 1000
kafka.auto.offset.reset: latest
kafka.request.timeout.ms: 40000
kafka.session.timeout.ms: 30000
kafka.isolation.level: read_committed
kafka.max.poll.records: 1000
# rocketMQ consumer
rocketmq.namespace:
rocketmq.namesrv.addr: 127.0.0.1:9876
rocketmq.batch.size: 1000
rocketmq.enable.message.trace: false
rocketmq.customized.trace.topic:
rocketmq.access.channel:
rocketmq.subscribe.filter:
# rabbitMQ consumer
rabbitmq.host:
rabbitmq.virtual.host:
rabbitmq.username:
rabbitmq.password:
rabbitmq.resource.ownerId:
srcDataSources:
defaultDS:
url: jdbc:mysql://rm-xxx.com:3306/nfturbo?useUnicode=true
username: xxx
password: xxx
canalAdapters:
- instance: example # canal instance Name or mq topic name
groups:
- groupId: g1
outerAdapters:
- name: logger
- name: es8
hosts: localhost:9200 # 127.0.0.1:9200 for rest mode
properties:
mode: transport # or rest
# security.auth: test:123456 # only used for rest mode
cluster.name: nfturbo-cluster
（1）srcDataSources配置的是监听Mysql数据库的数据源信息
（2）canalAdapters配置的是es的连接信息，其中es8指向yaml配置文件，配置的是es的索引、主键id、查询sql、批量提交数量等
【2】表映射配置 conf/rdb/mysql_sync.yml
因为在outerAdapters下面，我们配置的 name 是 es8（如果你自己安装的是 es7，记得修改成对应的版本。），所以adapter将会自动加载 conf/es8 下的所有.yml结尾的配置文件，适配器表映射文件，创建并修改 conf/es8/mytest_user.yml文件：
cd /root/canal_adapter/conf/es8 vi mytest_user.yml
dataSourceKey: defaultDS
destination: example
groupId: g1
esMapping:
_index: nfturbo_users
_id: _id
# upsert: true
# pk: id
sql: "select t.id as _id, t.nick_name as nick_name, t.state as state,t.telephone as telephone from users as t"
# objFields:
# _labels: array:;
#etlCondition: "where a.c_time>={}"
commitBatch: 3000
（4）启动canal.adapter
同样到 bin 目录下，执行命令：./startup.sh
cd /root/canal_adapter/bin && ./stop.sh && ./startup.sh tail -f /root/canal_adapter/logs/adapter/adapter.log
【3】测试同步
（1）到 es 上创建 ES 索引
访问 es（http://ip:5601/app/home#/ ），然后进入开发工具，在控制台创建如下索引：
http://192.168.220.128:5601/app/dev_tools#/console
PUT nfturbo_users
{
"mappings": {
"properties": {
"nickname": {
"type": "text"
},
"telephone": {
"type": "text"
},
"state": {
"type": "text"
}
}
}
}
（2）数据库执行INSERT
INSERT INTO `nfturbo`.`users` (`id`,`gmt_create`,`gmt_modified`,`nick_name`,`password_hash`,`state`,`telephone`,`user_role`) VALUES (14,'2024-04-18 17:47:40','2024-04-18 17:47:42','test11111','c2975f0faec10adca0ecd729c8cbc0aa','INIT','13000000000','CUSTOMER')
（3）从 ES 查询：
GET nfturbo_users/_search
{"_source": ["nick_name","telephone","state"],
"query": {
"match": {
"nick_name": "test11111"
}
}
}
到此这篇关于mysql使用Canal监听binlog日志，同步数据到MySQL、ElasticSearch、Redis、Kafka、RocketMQ和es的文章就介绍到这了,更多相关mysql使用Canal监听binlog日志来同步数据内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！
相关文章
现在php+mysql组合越来越多，这里脚本之家小编就为大家分享一下mysql的安装设置的几个小技巧2016-04-04
这篇文章主要介绍了MySQL 修改密码实例详解的相关资料,需要的朋友可以参考下2017-07-07
百万级、千万级数据处理，核心关键在于数据存储方案设计，存储方案设计的是否合理，直接影响到数据CRUD操作。总体设计可以考虑一下几个方面进行设计考虑：数据存储结构设计；索引设计；数据主键设计；查询方案设计。2021-05-05
这篇文章主要介绍了如何干涉MySQL优化器使用hash join,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下2022-09-09
win2003服务器下配置 MySQL 群集(Cluster)的方法
MySQL 群集是 MySQL 适合于分布式计算环境的高可用、高冗余版本。它采用了 NDB Cluster 存储引擎，允许在 1 个群集中运行多个 MySQL 服务器。2010-12-12
MySQL的执行流程包括查询缓存、解析器、优化器和执行器,首先,查询缓存检查是否存在查询结果,如果存在则直接返回；如果不存在,则进入解析器进行语法和语义分析,解析器将SQL语句转换为语法树,并进行词法和语法分析,接着,优化器确定最佳执行路径2024-12-12
本文将介绍mysql 触发器实现两个表的数据同步，需要学习MySQL的童鞋可以参考。2016-10-10
这篇文章主要介绍了mysql5.7使用变量进行分组排名并实现筛选方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教2024-05-05
这篇文章主要介绍了SQL语句之如何用JOIN连接多个表，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教2022-12-12
这篇文章主要介绍了Mysql 获取表的comment 字段操作，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧2021-01-01
最新评论