1 配置Canal+MQ数据同步环境
1.1 配置Mysql主从同步（使用下发虚拟机无需配置）
根据Canal的工作原理，首先需要开启MySQL主从同步。
1.在MySQL中需要创建一个用户，并授权
进入mysql容器：
docker exec -it mysql /bin/bash
-- 使用命令登录：
mysql -u root -p
-- 创建用户 用户名：canal 密码：canal
create user 'canal'@'%' identified WITH mysql_native_password by 'canal';
-- 授权 *.*表示所有库
GRANT SELECT,REPLICATION SLAVE,REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
- SELECT： 允许用户查询（读取）数据库中的数据。
- REPLICATION SLAVE： 允许用户作为 MySQL 复制从库，用于同步主库的数据。
- REPLICATION CLIENT： 允许用户连接到主库并获取关于主库状态的信息。
在MySQL配置文件my.cnf中设置如下信息，开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式
ROW 模式表示以行为单位记录每个被修改的行的变更
修改如下：
vi /usr/mysql/conf/my.cnf
[mysqld] #打开binlog log-bin=mysql-bin #选择ROW(行)模式 binlog-format=ROW #配置MySQL replaction需要定义，不要和canal的slaveId重复 server_id=1 expire_logs_days=3 max_binlog_size = 100m max_binlog_cache_size = 512m
说明：在学习阶段为了保证足够的服务器存储空间，binlog日志最大保存100m，mysql会定时清理binlog
2、重启MySQL，查看配置信息
- 使用命令查看是否打开binlog模式：
SHOW VARIABLES LIKE 'log_bin';
ON表示开启binlog模式。
show variables like 'binlog_format';
当 binlog_format 的值为 row 时，表示 MySQL 服务器当前配置为使用行级别的二进制日志记录，这对于数据库复制和数据同步来说更为安全，因为它记录了对数据行的确切更改。
- 查看binlog日志文件列表：
SHOW BINARY LOGS;
- 查看当前正在写入的binlog文件：
SHOW MASTER STATUS;
1.2 安装Canal（使用下发虚拟机无需安装）
获取canal镜像
docker pull canal/canal-server:latest
创建/data/soft/canal目录:
mkdir -p /data/soft/canal
在/data/soft/canal下创建 canal.properties，内容如下，注意修改mq的配置信息：
################################################# ######### common argument ############# ################################################# # tcp bind ip canal.ip = # register ip to zookeeper canal.register.ip = canal.port = 11111 canal.metrics.pull.port = 11112 # canal instance user/passwd # canal.user = canal # canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458 # canal admin config #canal.admin.manager = 127.0.0.1:8089 canal.admin.port = 11110 canal.admin.user = admin canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441 # admin auto register #canal.admin.register.auto = true #canal.admin.register.cluster = #canal.admin.register.name = canal.zkServers = # flush data to zk canal.zookeeper.flush.period = 1000 canal.withoutNetty = false # tcp, kafka, rocketMQ, rabbitMQ canal.serverMode = rabbitMQ # flush meta cursor/parse position to file canal.file.data.dir = ${canal.conf.dir} canal.file.flush.period = 1000 ## memory store RingBuffer size, should be Math.pow(2,n) canal.instance.memory.buffer.size = 16384 ## memory store RingBuffer used memory unit size , default 1kb canal.instance.memory.buffer.memunit = 1024 ## meory store gets mode used MEMSIZE or ITEMSIZE canal.instance.memory.batch.mode = MEMSIZE canal.instance.memory.rawEntry = true ## detecing config canal.instance.detecting.enable = false #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now() canal.instance.detecting.sql = select 1 canal.instance.detecting.interval.time = 3 canal.instance.detecting.retry.threshold = 3 canal.instance.detecting.heartbeatHaEnable = false # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery canal.instance.transaction.size = 1024 # mysql fallback connected to new master should fallback times canal.instance.fallbackIntervalInSeconds = 60 # network config canal.instance.network.receiveBufferSize = 16384 canal.instance.network.sendBufferSize = 16384 canal.instance.network.soTimeout = 30 # binlog filter config canal.instance.filter.druid.ddl = true canal.instance.filter.query.dcl = false # 这个配置一定要修改 canal.instance.filter.query.dml = true canal.instance.filter.query.ddl = false canal.instance.filter.table.error = false canal.instance.filter.rows = false canal.instance.filter.transaction.entry = false canal.instance.filter.dml.insert = false canal.instance.filter.dml.update = false canal.instance.filter.dml.delete = false # binlog format/image check canal.instance.binlog.format = ROW,STATEMENT,MIXED canal.instance.binlog.image = FULL,MINIMAL,NOBLOB # binlog ddl isolation canal.instance.get.ddl.isolation = false # parallel parser config canal.instance.parser.parallel = true ## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors() #canal.instance.parser.parallelThreadSize = 16 ## disruptor ringbuffer size, must be power of 2 canal.instance.parser.parallelBufferSize = 256 # table meta tsdb info canal.instance.tsdb.enable = true canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:} canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; canal.instance.tsdb.dbUsername = canal canal.instance.tsdb.dbPassword = canal # dump snapshot interval, default 24 hour canal.instance.tsdb.snapshot.interval = 24 # purge snapshot expire , default 360 hour(15 days) canal.instance.tsdb.snapshot.expire = 360 ################################################# ######### destinations ############# ################################################# canal.destinations = xzb-canal # conf root dir canal.conf.dir = ../conf # auto scan instance dir add/remove and start/stop instance canal.auto.scan = true canal.auto.scan.interval = 5 # set this value to 'true' means that when binlog pos not found, skip to latest. # WARN: pls keep 'false' in production env, or if you know what you want. canal.auto.reset.latest.pos.mode = false canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml canal.instance.global.mode = spring canal.instance.global.lazy = false canal.instance.global.manager.address = ${canal.admin.manager} #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml canal.instance.global.spring.xml = classpath:spring/file-instance.xml #canal.instance.global.spring.xml = classpath:spring/default-instance.xml ################################################## ######### MQ Properties ############# ################################################## # aliyun ak/sk , support rds/mq canal.aliyun.accessKey = canal.aliyun.secretKey = canal.aliyun.uid= canal.mq.flatMessage = true canal.mq.canalBatchSize = 50 canal.mq.canalGetTimeout = 100 # Set this value to "cloud", if you want open message trace feature in aliyun. canal.mq.accessChannel = local canal.mq.database.hash = true canal.mq.send.thread.size = 30 canal.mq.build.thread.size = 8 ################################################## ######### Kafka ############# ################################################## kafka.bootstrap.servers = 127.0.0.1:9092 kafka.acks = all kafka.compression.type = none kafka.batch.size = 16384 kafka.linger.ms = 1 kafka.max.request.size = 1048576 kafka.buffer.memory = 33554432 kafka.max.in.flight.requests.per.connection = 1 kafka.retries = 0 kafka.kerberos.enable = false kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf" kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf" ################################################## ######### RocketMQ ############# ################################################## rocketmq.producer.group = test rocketmq.enable.message.trace = false rocketmq.customized.trace.topic = rocketmq.namespace = rocketmq.namesrv.addr = 127.0.0.1:9876 rocketmq.retry.times.when.send.failed = 0 rocketmq.vip.channel.enabled = false rocketmq.tag = ################################################## ######### RabbitMQ ############# ################################################## rabbitmq.host = 192.168.101.68 rabbitmq.virtual.host = /hmall rabbitmq.exchange = exchange.canal-hmall rabbitmq.username = hmall rabbitmq.password = 123 rabbitmq.deliveryMode = 2
上边配置文件中主要修改RabbitMQ 区域中MQ的相关配置。
################################################## ######### RabbitMQ ############# ################################################## rabbitmq.host = 192.168.101.68 rabbitmq.virtual.host = /hmall rabbitmq.exchange = exchange.canal-hmall rabbitmq.username = hmall rabbitmq.password = 123 rabbitmq.deliveryMode = 2
创建instance.properties，内容如下：
canal.instance.master.journal.name 用于指定主库正在写入的 binlog 文件的名称。
如果不配置 canal.instance.master.journal.name，Canal 会尝试自动检测 MySQL 主库的 binlog 文件，并从最新位置开始进行复制。
################################################# ## mysql serverId , v1.0.26+ will autoGen canal.instance.mysql.slaveId=1000 # enable gtid use true/false canal.instance.gtidon=false # position info canal.instance.master.address=192.168.101.68:3306 canal.instance.master.journal.name=mysql-bin.000001 canal.instance.master.position=0 canal.instance.master.timestamp= canal.instance.master.gtid= # rds oss binlog canal.instance.rds.accesskey= canal.instance.rds.secretkey= canal.instance.rds.instanceId= # table meta tsdb info canal.instance.tsdb.enable=true #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb #canal.instance.tsdb.dbUsername=canal #canal.instance.tsdb.dbPassword=canal #canal.instance.standby.address = #canal.instance.standby.journal.name = #canal.instance.standby.position = #canal.instance.standby.timestamp = #canal.instance.standby.gtid= # username/password canal.instance.dbUsername=canal canal.instance.dbPassword=canal canal.instance.connectionCharset = UTF-8 # enable druid Decrypt database password canal.instance.enableDruid=false #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ== # table regex # canal.instance.filter.regex=test01\\..*,test02\\..* #canal.instance.filter.regex=test01\\..*,test02\\.t1 #canal.instance.filter.regex=jzo2o-foundations\\.serve_sync,jzo2o-orders-0\\.orders_seize,jzo2o-orders-0\\.orders_dispatch,jzo2o-orders-0\\.serve_provider_sync,jzo2o-customer\\.serve_provider_sync #canal.instance.filter.regex=jzo2o-orders-1\\.orders_dispatch,jzo2o-orders-1\\.orders_seize,jzo2o-foundations\\.serve_sync,jzo2o-customer\\.serve_provider_sync,jzo2o-orders-1\\.serve_provider_sync,jzo2o-orders-1\\.history_orders_sync,jzo2o-orders-1\\.history_orders_serve_sync,jzo2o-market\\.activity canal.instance.filter.regex=hm-item\\.item_sync # table black regex canal.instance.filter.black.regex=mysql\\.slave_.* # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2) #canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch # mq config #canal.mq.topic=topic_test01 # dynamic topic route by schema or table regex #canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..* #canal.mq.dynamicTopic=topic_test01:test01\\..*,topic_test02:test02\\..* #canal.mq.dynamicTopic=canal-mq-jzo2o-orders-dispatch:jzo2o-orders-0\\.orders_dispatch,canal-mq-jzo2o-orders-seize:jzo2o-orders-0\\.orders_seize,canal-mq-jzo2o-foundations:jzo2o-foundations\\.serve_sync,canal-mq-jzo2o-customer-provider:jzo2o-customer\\.serve_provider_sync,canal-mq-jzo2o-orders-provider:jzo2o-orders-0\\.serve_provider_sync #canal.mq.dynamicTopic=canal-mq-jzo2o-orders-dispatch:jzo2o-orders-1\\.orders_dispatch,canal-mq-jzo2o-orders-seize:jzo2o-orders-1\\.orders_seize,canal-mq-jzo2o-foundations:jzo2o-foundations\\.serve_sync,canal-mq-jzo2o-customer-provider:jzo2o-customer\\.serve_provider_sync,canal-mq-jzo2o-orders-provider:jzo2o-orders-1\\.serve_provider_sync,canal-mq-jzo2o-orders-serve-history:jzo2o-orders-1\\.history_orders_serve_sync,canal-mq-jzo2o-orders-history:jzo2o-orders-1\\.history_orders_sync,canal-mq-jzo2o-market-resource:jzo2o-market\\.activity canal.mq.dynamicTopic=canal-mq-hmall-item:hm-item\\.item_sync canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=3 #canal.mq.partitionHash=test.table:id^name,.*\\..* #canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6 #################################################
canal.instance.filter.regex和canal.mq.dynamicTopic的配置稍后解释。
创建日志目录：
mkdir -p /data/soft/canal/logs /data/soft/canal/conf
启动容器：
docker run --name canal -p 11111:11111 -d -v /data/soft/canal/instance.properties:/home/admin/canal-server/conf/xzb-canal/instance.properties -v /data/soft/canal/canal.properties:/home/admin/canal-server/conf/canal.properties -v /data/soft/canal/logs:/home/admin/canal-server/logs/xzb-canal -v /data/soft/canal/conf:/home/admin/canal-server/conf/xzb-canal canal/canal-server:latest
1.3 安装RabbitMQ（使用下发虚拟机无需安装）
安装RabbitMQ参考 微服务课程内容。
配置虚拟主机：/hmall
账号：hmall
密码：123
1.4 配置Canal+RabbitMQ
下边通过配置Canal与RabbitMQ，保证Canal收到binlog消息将数据发送至MQ。
最终我们要实现的是：
修改hm-item数据库下的item_sync表的数据后通过canal将修改信息发送到MQ。
1、在Canal中配置RabbitMQ的连接信息[已完成]
修改/data/soft/canal/canal.properties
# tcp, kafka, rocketMQ, rabbitMQ canal.serverMode = rabbitMQ ################################################## ######### RabbitMQ ############# ################################################## rabbitmq.host = 192.168.101.68 rabbitmq.virtual.host = /hmall rabbitmq.exchange = exchange.canal-hmall rabbitmq.username = hmall rabbitmq.password = 123 rabbitmq.deliveryMode = 2
本项目用于数据同步的MQ交换机：exchange.canal-hmall
虚拟主机地址：/hmall
账号和密码：hmall/ 123
rabbitmq.deliveryMode = 2 设置消息持久化
2、设置需要读取binlog的mysql库和表[已完成]
修改/data/soft/canal/instance.properties
canal.instance.filter.regex需要监听的mysql库和表
- 全库：
.*\\..* - 指定库下的所有表：
canal\\..* - 指定库下的指定表：
canal\\.canal,test\\.test
库名\\.表名：转义需要用\\，使用逗号分隔多个库
这里配置监听 hm-item数据库下item_sync表，如下：
canal.instance.filter.regex=hm-item\\.item_sync
3、在Canal配置MQ的topic[已完成]
这里使用动态topic，格式为：topic:schema.table,topic:schema.table,topic:schema.table
配置如下：
canal.mq.dynamicTopic=canal-mq-hmall-item:hm-item\\.item_sync
上边的配置表示：对hm-item数据库的item_sync表的修改消息发到exchange.canal-hmall交换机，topic为canal-mq-hmall-item关联的队列中。
4、进入rabbitMQ配置交换机和队列
创建exchange.canal-hmall交换机：
创建队列：canal-mq-hmall-item
绑定交换机：
绑定成功：
1.5 测试数据同步
复制item表到item_sync表。
重启canal
修改hmall-item数据库的item_sync表的数据，稍等片刻查看canal-mq-hmall-item队列，如果队列中有的消息说明同步成功，如下图：
如果没有同步到 MQ参考常见问题中“数据不同步”进行解决。
我们可以查询队列中的消息内容发现它一条type为"UPDATE"的消息，如下所示：
{ "data" : [ { "brand" : "RIMOWA", "category" : "拉杆箱", "comment_count" : "0", "create_time" : "2019-05-01 00:00:00", "creater" : null, "id" : "317578", "image" : "https://m.360buyimg.com/mobilecms/s720x720_jfs/t6934/364/1195375010/84676/e9f2c55f/597ece38N0ddcbc77.jpg!q70.jpg.webp", "isAD" : "0", "name" : "RIMOWA 21寸托运箱拉杆箱 SALSA AIR系列果绿色 820.70.36.41", "price" : "28900", "sold" : "0", "spec" : "{\"颜色\": \"红色\", \"尺码\": \"21寸\"}", "status" : "1", "stock" : "10000", "update_time" : "2023-05-06 11:06:17", "updater" : "100" } ], "database" : "hm-item", "es" : 1724319800000.0, "id" : 1, "isDdl" : false, "mysqlType" : { "brand" : "varchar(100)", "category" : "varchar(200)", "comment_count" : "int", "create_time" : "datetime", "creater" : "bigint", "id" : "bigint", "image" : "varchar(200)", "isAD" : "tinyint", "name" : "varchar(200)", "price" : "int", "sold" : "int", "spec" : "varchar(200)", "status" : "int", "stock" : "int", "update_time" : "datetime", "updater" : "bigint" }, "old" : [ { "name" : "RIMOWA 21寸托运箱拉杆箱 SALSA AIR系列果绿色 820.70.36.4" } ], "pkNames" : null, "sql" : "", "sqlType" : { "brand" : 12, "category" : 12, "comment_count" : 4, "create_time" : 93, "creater" : -5, "id" : -5, "image" : 12, "isAD" : -6, "name" : 12, "price" : 4, "sold" : 4, "spec" : 12, "status" : 4, "stock" : 4, "update_time" : 93, "updater" : -5 }, "table" : "item_sync", "ts" : 1724319801224.0, "type" : "UPDATE" }
常见问题
数据不同步
当发现修改了数据库后修改的数据并没有发送到MQ，通过查看Canal的日志发现下边的错误。
进入Canal目录，查看日志：
cd /data/soft/canal/logs tail -f xzb-canal.log
Canal报错如下：
2023-09-22 08:34:40.802 [destination = xzb-canal , address = /192.168.101.68:3306 , EventParser] WARN c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=mysql-bin.000055,position=486221,serverId=1,gtid=,timestamp=1695341830000] cost : 13ms , the next step is binlog dump 2023-09-22 08:34:40.811 [destination = xzb-canal , address = /192.168.101.68:3306 , EventParser] ERROR c.a.o.canal.parse.inbound.mysql.dbsync.DirectLogFetcher - I/O error while reading from client socket java.io.IOException: Received error packet: errno = 1236, sqlstate = HY000 errmsg = Could not find first log file name in binary log index file at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher.fetch(DirectLogFetcher.java:102) ~[canal.parse-1.1.5.jar:na] at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.dump(MysqlConnection.java:238) [canal.parse-1.1.5.jar:na] at com.alibaba.otter.canal.parse.inbound.AbstractEventParser$1.run(AbstractEventParser.java:262) [canal.parse-1.1.5.jar:na] at java.lang.Thread.run(Thread.java:748) [na:1.8.0_181]
找到关键的位置：Could not find first log file name in binary log index file
根据日志分析是Canal找不到mysql-bin.000055 的486221位置，原因是mysql-bin.000055文件不存在，这是由于为了节省磁盘空间将binlog日志清理了。
解决方法：
把canal复位从最开始开始同步的位置。
1）首先重置mysql的bin log：
连接mysql执行：reset master
把canal复位从最开始开始同步的位置。
1）首先重置mysql的bin log：
docker exec -it mysql /bin/bash
-- 使用命令登录（密码：mysql）：
mysql -u root -p
连接mysql执行：reset master;
执行后所有的binlog删除，从000001号开始
通过show master status;查看 ，结果显示 mysql-bin.000001 
2）先停止canal
docker stop canal
3）删除meta.dat
rm -rf /data/soft/canal/conf/meta.dat
4) 再启动canal
docker start canal
MQ同步消息无法消费
这里以Es和MySQL之间的同步举例：
当出现ES和MySQL数据不同步时可能会出现MQ的同步消息无法被消费，比如：从MySQL删除一条记录通过同步程序将ES中对应的记录进行删除，此时由于ES中没有该记录导致删除ES中的记录失败。出现此问题的原因是因为测试数据混乱导致，可以手动将MQ中的消息删除。
进入MQ的管理控制台，进入要清理消息的队列，通过purge功能清理消息：
