devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

canal--canal+MQ

作者：我家自动化 | 2024-04-18 09:57:05

赞

踩

canal+mq

原文网址：canal--canal+MQ_IT利刃出鞘的博客-CSDN博客_canal mq

部署与配置

下载与安装

环境版本

操作系统：CentOS release 6.6 (Final)
java版本: jdk1.8
canal 版本: 请下载最新的安装包，本文以当前v1.1.1 的canal.deployer-1.1.1.tar.gz为例
MySQL版本 ：5.7.18（这里有一个坑，就是说mysql使用时请注意你的版本，不然报错一直连接不上）
注意 ： 关闭所有机器的防火墙，同时注意启动可以相互telnet ip 端口


下载压缩包

到官网地址(release)下载最新压缩包,请下载 canal.deployer-latest.tar.gz

将canal.deployer 复制到固定目录并解压

  1. mkdir -p /usr/local/canal
  2. cp   canal.deployer-1.1.1.tar.gz   /usr/local/canal
  3. tar -zxvf canal.deployer-1.1.1.tar.gz 

MySQL配置修改参数

开启mysql的binlog写入功能

canal的原理是基于mysql binlog技术，所以这里一定需要开启mysql的binlog写入功能，并且配置binlog模式为row.

  1. [mysqld]
  2. log-bin=mysql-bin #启用binlog
  3. binlog-format=ROW #选择row模式
  4. server_id=1 #配置mysql replaction，不能和canal的slaveId重复

创建canal用户

 canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限。

  1. CREATE USER canal IDENTIFIED BY 'canal';    
  2. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';  
  3. -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;  
  4. FLUSH PRIVILEGES;

 针对已有的账户可通过grants查询权限

show grants for 'canal'

canal配置修改参数

a. 修改instance 配置文件 vi conf/example/instance.properties（要是你配置困难，建议你直接复制）

  1. #################################################
  2. ## mysql serverId , v1.0.26+ will autoGen
  3. canal.instance.mysql.slaveId=1234
  4.  
  5. # enable gtid use true/false
  6. canal.instance.gtidon=false
  7.  
  8. # position info
  9. canal.instance.master.address=127.0.0.1:3306
  10. canal.instance.master.journal.name=
  11. canal.instance.master.position=
  12. canal.instance.master.timestamp=
  13. canal.instance.master.gtid=
  14.  
  15. # rds oss binlog
  16. canal.instance.rds.accesskey=
  17. canal.instance.rds.secretkey=
  18. canal.instance.rds.instanceId=
  19.  
  20. # table meta tsdb info
  21. canal.instance.tsdb.enable=true
  22. #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
  23. #canal.instance.tsdb.dbUsername=canal
  24. #canal.instance.tsdb.dbPassword=canal
  25.  
  26. #canal.instance.standby.address =
  27. #canal.instance.standby.journal.name =
  28. #canal.instance.standby.position =
  29. #canal.instance.standby.timestamp =
  30. #canal.instance.standby.gtid=
  31.  
  32. # username/password
  33. canal.instance.dbUsername=canal
  34. canal.instance.dbPassword=canal
  35. canal.instance.connectionCharset = UTF-8
  36. canal.instance.defaultDatabaseName =
  37. # enable druid Decrypt database password
  38. canal.instance.enableDruid=false
  39. #canal.instance.pwdPublicKey=MFwwDQYJltRRE5/zXpVX7y7GCMo2/JM6fircSAwEAAQ==
  40.  
  41. # table regex
  42. canal.instance.filter.regex=.*\\..*
  43. # table black regex
  44. canal.instance.filter.black.regex=
  45.  
  46. # mq config
  47. canal.mq.topic=test
  48. # dynamic topic route by table regex
  49. #canal.mq.dynamicTopic=.*,mytest\\..*,mytest2.user
  50. canal.mq.partition=0
  51. # hash partition config
  52. #canal.mq.partitionsNum=3
  53. #canal.mq.partitionHash=test.table:id^name,.*\\..*
  54. #################################################

对应ip 地址的MySQL 数据库需进行相关初始化与设置, 可参考 Canal QuickStart

a1.多数据源配置通道：https://www.2cto.com/database/201609/547661.html
a2.常见的异常通道：https://blog.csdn.net/my201110lc/article/details/77885720

b. 修改canal 配置文件/usr/local/canal/conf/canal.properties（一般使用默认即可，可不修改。）

  1. #################################################
  2. ######### 		common argument		############# 
  3. #################################################
  4. #canal.manager.jdbc.url=jdbc:mysql://127.0.0.1:3306/canal_manager?useUnicode=true&characterEncoding=UTF-8
  5. #canal.manager.jdbc.username=root
  6. #canal.manager.jdbc.password=121212
  7. canal.id = 1
  8. canal.ip =
  9. canal.port = 11111
  10. canal.metrics.pull.port = 11112
  11. canal.zkServers =
  12. # flush data to zk
  13. canal.zookeeper.flush.period = 1000
  14. canal.withoutNetty = false
  15. # tcp, kafka, RocketMQ
  16. canal.serverMode = kafka
  17. # flush meta cursor/parse position to file
  18. canal.file.data.dir = ${canal.conf.dir}
  19. canal.file.flush.period = 1000
  20. ## memory store RingBuffer size, should be Math.pow(2,n)
  21. canal.instance.memory.buffer.size = 16384
  22. ## memory store RingBuffer used memory unit size , default 1kb
  23. canal.instance.memory.buffer.memunit = 1024 
  24. ## meory store gets mode used MEMSIZE or ITEMSIZE
  25. canal.instance.memory.batch.mode = MEMSIZE
  26. canal.instance.memory.rawEntry = true
  27.  
  28. ## detecing config
  29. canal.instance.detecting.enable = false
  30. #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
  31. canal.instance.detecting.sql = select 1
  32. canal.instance.detecting.interval.time = 3
  33. canal.instance.detecting.retry.threshold = 3
  34. canal.instance.detecting.heartbeatHaEnable = false
  35.  
  36. # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
  37. canal.instance.transaction.size =  1024
  38. # mysql fallback connected to new master should fallback times
  39. canal.instance.fallbackIntervalInSeconds = 60
  40.  
  41. # network config
  42. canal.instance.network.receiveBufferSize = 16384
  43. canal.instance.network.sendBufferSize = 16384
  44. canal.instance.network.soTimeout = 30
  45.  
  46. # binlog filter config
  47. canal.instance.filter.druid.ddl = true
  48. canal.instance.filter.query.dcl = false
  49. canal.instance.filter.query.dml = false
  50. canal.instance.filter.query.ddl = false
  51. canal.instance.filter.table.error = false
  52. canal.instance.filter.rows = false
  53. canal.instance.filter.transaction.entry = false
  54.  
  55. # binlog format/image check
  56. canal.instance.binlog.format = ROW,STATEMENT,MIXED 
  57. canal.instance.binlog.image = FULL,MINIMAL,NOBLOB
  58.  
  59. # binlog ddl isolation
  60. canal.instance.get.ddl.isolation = false
  61.  
  62. # parallel parser config
  63. canal.instance.parser.parallel = true
  64. ## concurrent thread number, default 60% available processors, suggest not to exceed Runtime.getRuntime().availableProcessors()
  65. #canal.instance.parser.parallelThreadSize = 16
  66. ## disruptor ringbuffer size, must be power of 2
  67. canal.instance.parser.parallelBufferSize = 256
  68.  
  69. # table meta tsdb info
  70. canal.instance.tsdb.enable = true
  71. canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
  72. canal.instance.tsdb.url = jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
  73. canal.instance.tsdb.dbUsername = canal
  74. canal.instance.tsdb.dbPassword = canal
  75. # dump snapshot interval, default 24 hour
  76. canal.instance.tsdb.snapshot.interval = 24
  77. # purge snapshot expire , default 360 hour(15 days)
  78. canal.instance.tsdb.snapshot.expire = 360
  79.  
  80. # aliyun ak/sk , support rds/mq
  81. canal.aliyun.accesskey =
  82. canal.aliyun.secretkey =
  83.  
  84. #################################################
  85. ######### 		destinations		############# 
  86. #################################################
  87. canal.destinations = example
  88. # conf root dir
  89. canal.conf.dir = ../conf
  90. # auto scan instance dir add/remove and start/stop instance
  91. canal.auto.scan = true
  92. canal.auto.scan.interval = 5
  93.  
  94. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
  95. #canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
  96.  
  97. canal.instance.global.mode = spring
  98. canal.instance.global.lazy = false
  99. #canal.instance.global.manager.address = 127.0.0.1:1099
  100. #canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
  101. canal.instance.global.spring.xml = classpath:spring/file-instance.xml
  102. #canal.instance.global.spring.xml = classpath:spring/default-instance.xml
  103.  
  104. ##################################################
  105. ######### 		     MQ 		     #############
  106. ##################################################
  107. canal.mq.servers = 127.0.0.1:9092
  108. canal.mq.retries = 0
  109. canal.mq.batchSize = 16384
  110. canal.mq.maxRequestSize = 1048576
  111. canal.mq.lingerMs = 1
  112. canal.mq.bufferMemory = 33554432
  113. canal.mq.canalBatchSize = 50
  114. canal.mq.canalGetTimeout = 100
  115. canal.mq.flatMessage = true
  116. canal.mq.compressionType = none
  117. canal.mq.acks = all

启动/看日志/关闭

启动

  1. cd /usr/local/canal/
  2. sh bin/startup.sh   //关闭是sh bin/stop.sh

inux带debug方式启动：(默认使用suspend=y，阻塞等待你remote debug链接成功)

/bin/startup.sh debug 9099

  1. linux启动完成后，会在bin目录下生成canal.pid，stop.sh会读取canal.pid进行进程关闭
  2. startup.sh默认读取系统环境变量中的which java获得JAVA执行路径，需要设置PATH=$JAVA_HOME/bin环境变量

windows启动：(windows支持相对比较弱）

startup.bat 

 windows停止：直接关闭终端即可

查看日志

可以查看启动是否成功等

canal日志：canal/logs/canal/canal.log
实例日志：canal/logs/example/example.log

效果

mq相关参数说明

参数名

参数说明

默认值

canal.mq.servers

kafka为bootstrap.servers 
rocketMQ中为nameserver列表

127.0.0.1:6667

canal.mq.retries

发送失败重试次数

0

canal.mq.batchSize

kafka为ProducerConfig.BATCH_SIZE_CONFIG 
rocketMQ无意义

16384

canal.mq.maxRequestSize

kafka为ProducerConfig.MAX_REQUEST_SIZE_CONFIG 
rocketMQ无意义

1048576

canal.mq.lingerMs

kafka为ProducerConfig.LINGER_MS_CONFIG 
rocketMQ无意义

1

canal.mq.bufferMemory

kafka为ProducerConfig.BUFFER_MEMORY_CONFIG 
rocketMQ无意义

33554432

canal.mq.producerGroup

kafka无意义 
rocketMQ为ProducerGroup名

Canal-Producer

canal.mq.canalBatchSize

获取canal数据的批次大小

50

canal.mq.canalGetTimeout

获取canal数据的超时时间

100

canal.mq.flatMessage

是否为json格式 
如果设置为false,对应MQ收到的消息为protobuf格式
需要通过CanalMessageDeserializer进行解码

true

---

---

---

canal.mq.topic

mq里的topic名

无

canal.mq.dynamicTopic

mq里的动态topic规则, 1.1.3版本支持

无

canal.mq.partition

单队列模式的分区下标，

1

canal.mq.partitionsNum

散列模式的分区数

无

canal.mq.partitionHash

散列规则定义 
库名.表名 : 唯一主键，比如mytest.person: id 
1.1.3版本支持新语法，见下文

无

canal.mq.dynamicTopic 表达式说明

canal 1.1.3版本之后, 支持配置格式：schema.table，多个配置之间使用逗号分隔。

例：

表达式含义
test\\.test指定匹配的单表，发送到以 test\\.test为名字的topic上
.\\..*匹配所有表，每个表都会发送到各自表名的topic上
test指定匹配对应的库，一个库的所有表都会发送到库名的topic上
test\\.*指定匹配的表达式，针对匹配的表会发送到各自表名的topic上
test,test1\\.test1test1指定多个表达式，会将test库的表都发送到test的topic上，test1\\.test1的表发送到对应的test1\\.test1 topic上，其余的表发送到默认的canal.mq.topic值

大家可以结合自己的业务需求，设置匹配规则，建议MQ开启自动创建topic的能力

canal.mq.partitionHash 表达式说明

canal 1.1.3版本之后, 支持配置格式：schema.table:pk1^pk2，多个配置之间使用逗号分隔。

表达式含义
test\\.test:pk1^pk2指定匹配的单表，对应的hash字段为pk1 + pk2
.\\..*:id匹配，指定所有正则匹配的表对应的hash字段为id
.\\..*:$pk$正则匹配，指定所有正则匹配的表对应的hash字段为表主键(自动查找)
匹配规则啥都不写，则默认发到0这个partition上
.\\..*不指定pk信息的正则匹配，将所有正则匹配的表,对应的hash字段为表名
按表hash: 一张表的所有数据可以发到同一个分区，不同表之间会做散列 (会有热点表分区过大问题)
test\\.test:id,.\\..*针对test的表按照id散列,其余的表按照table散列

注意：大家可以结合自己的业务需求，设置匹配规则，多条匹配规则之间是按照顺序进行匹配(命中一条规则就返回)

其他详细参数可参考Canal AdminGuide

mq顺序性问题

binlog本身是有序的，写入到mq之后如何保障顺序是很多人会比较关注，在issue里也有非常多人咨询了类似的问题，这里做一个统一的解答

  1. canal目前选择支持的kafka/rocketmq，本质上都是基于本地文件的方式来支持了分区级的顺序消息的能力，也就是binlog写入mq是可以有一些顺序性保障，这个取决于用户的一些参数选择
  2. canal支持MQ数据的几种路由方式：单topic单分区，单topic多分区、多topic单分区、多topic多分区
  • canal.mq.dynamicTopic，主要控制是否是单topic还是多topic，针对命中条件的表可以发到表名对应的topic、库名对应的topic、默认topic name
  • canal.mq.partitionsNum、canal.mq.partitionHash，主要控制是否多分区以及分区的partition的路由计算，针对命中条件的可以做到按表级做分区、pk级做分区等

canal的消费顺序性，主要取决于描述2中的路由选择，举例说明：

  • 单topic单分区，可以严格保证和binlog一样的顺序性，缺点就是性能比较慢，单分区的性能写入大概在2~3k的TPS
  • 多topic单分区，可以保证表级别的顺序性，一张表或者一个库的所有数据都写入到一个topic的单分区中，可以保证有序性，针对热点表也存在写入分区的性能问题
  • 单topic、多topic的多分区，如果用户选择的是指定table的方式，那和第二部分一样，保障的是表级别的顺序性(存在热点表写入分区的性能问题)，如果用户选择的是指定pk hash的方式，那只能保障的是一个pk的多次binlog顺序性 ** pk hash的方式需要业务权衡，这里性能会最好，但如果业务上有pk变更或者对多pk数据有顺序性依赖，就会产生业务处理错乱的情况. 如果有pk变更，pk变更前和变更后的值会落在不同的分区里，业务消费就会有先后顺序的问题，需要注意

MQ数据消费

canal.client下有对应的MQ数据消费的样例工程,包含数据编解码的功能

kafka模式: com.alibaba.otter.canal.client.running.kafka.CanalKafkaClientExample
rocketMQ模式: com.alibaba.otter.canal.client.running.rocketmq.CanalRocketMQClientExample

其他网址

https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart
canal实现mysql实时数据binlog同步_数据库_阿拉斯加大闸蟹的博客-CSDN博客

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/445320
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号