canal 环境搭建（docker）

墨蓝 2022-10-09 04:52 514阅读 0赞

canal 环境搭建（docker）

官网：https://github.com/alibaba/canal/wiki/Docker-QuickStart

docker 仓库：https://hub.docker.com/r/canal/canal-server/tags

配置文件：https://github.com/alibaba/canal/wiki/AdminGuide

*********************

架构设计

单机：canal server、canal client单节点直连

  1.               ![2021062515525815.png][]

客户端连接

  1. public class CanalConnectors {
  3.     public static CanalConnector newSingleConnector(SocketAddress address, String destination, String username, String password) {
  4.         SimpleCanalConnector canalConnector = new SimpleCanalConnector(address, username, password, destination);
  5.         canalConnector.setSoTimeout(60000);
  6.         canalConnector.setIdleTimeout(3600000);
  7.         return canalConnector;
  8.     }

canal server高可用：canal server集群部署、canal client直连canal server服务列表

  1.               ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70][]

canal server：集群部署，将节点信息存储在在zookeeper中，可实现高可用

canal client：通过canal server静态服务列表建立连接

  1. public class CanalConnectors {
  3.     public static CanalConnector newClusterConnector(List<? extends SocketAddress> addresses, String destination, String username, String password) {
  4.         ClusterCanalConnector canalConnector = new ClusterCanalConnector(username, password, destination, new SimpleNodeAccessStrategy(addresses));
  5.         canalConnector.setSoTimeout(60000);
  6.         canalConnector.setIdleTimeout(3600000);
  7.         return canalConnector;
  8.     }

canal server、canal client高可用：canal server集群部署、canal client开启多个

  1.               ![20210625160946400.png][]

canal server：集群部署，节点信息存储在zookeeper中，可实现高可用

canal client：canal client注册到zookeeper中，从zookeeper中获取canal server信息

  1. public class CanalConnectors {
  3.     public static CanalConnector newClusterConnector(String zkServers, String destination, String username, String password) {
  4.         ClusterCanalConnector canalConnector = new ClusterCanalConnector(username, password, destination, new ClusterNodeAccessStrategy(destination, ZkClientx.getZkClient(zkServers)));
  5.         canalConnector.setSoTimeout(60000);
  6.         canalConnector.setIdleTimeout(3600000);
  7.         return canalConnector;
  8.     }

*********************

canal server 配置

canal配置加载方式：ManagerCanalInstanceGenerator、SpringCanalInstanceGenerator

  1.                   ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzMTYyNQ_size_16_color_FFFFFF_t_70 1][]

ManagerCanalInstanceGenerator：可视化界面配置canal参数

SpringCanalnstanceGenerator：本地文件配置(xxx-instance.xml、canal.properties、instance.properties)

*********************

xxx-instance.xml

创建CanalIstanceWithSpring实例，可选文件如下

memory-instance.xml：元数据在内存存储

file-instane.xml：**元数据持久化到文件，**log parser position优先在内存中查找，查找不到则到文件中查找

default-instance.xml：元数据保存到zookeeper，log parser position优先在内存中查找，查找不到则到zookeeper中查找

group-instance.xml：将多个parser组合成一个parser，可用于将分库分表后的数据导入同一地方存储分析，元数据默认保存在内存中

  1. public class CanalInstanceWithSpring extends AbstractCanalInstance {
  2.     private static final Logger logger = LoggerFactory.getLogger(CanalInstanceWithSpring.class);
  4.     public CanalInstanceWithSpring() {
  5.     }
  8. **************
  9. AbstractCanalInstance
  11. public class AbstractCanalInstance extends AbstractCanalLifeCycle implements CanalInstance {
  12.     private static final Logger logger = LoggerFactory.getLogger(AbstractCanalInstance.class);
  13.     protected Long canalId;                          //canal标识
  14.     protected String destination;                    //instance实例的名称，一个canal下可有多个instance
  16.     protected CanalEventStore<Event> eventStore;     //eventStore，存储拉取的数据
  17.     protected CanalEventParser eventParser;          //解析数据源
  18.     protected CanalEventSink<List<Entry>> eventSink; //处理转换数据
  20.     protected CanalMetaManager metaManager;          //元数据管理器，parser log position、cursor position等数据
  21.     protected CanalAlarmHandler alarmHandler;        //报警处理类
  22.     protected CanalMQConfig mqConfig;                //mq配置，支持rocketmq、kafka、rabbitmq
  24.     public AbstractCanalInstance() {
  25.     }

*********************

properties 文件

canal.properties：配置canal server上instance的公共属性

instance.properties：配置instance的属性，若有同名配置，instance.properties优先级更高

canal.properties

  1. #################################################
  2. #########         common argument        #############
  3. #################################################
  4. canal.id = 1                     #canal server的唯一标识，默认为1
  6. # canal server用户名、密码，canal user、password如果不设置，则不开启密码功能
  7. canal.user = canal
  8. canal.passwd = E3619321C1A937C46A0D8BD1DAC39F93B27D4458
  10. canal.ip =                       #canal server绑定的ip地址
  11. canal.port = 11111               #canal server tcp连接端口，供客户端使用，默认为11111
  12. canal.metrics.pull.port = 11112  #canal server指标数据端口，默认为11112
  14. canal.register.ip =              #canal server注册到zookeeper中的ip信息
  15. canal.zkServers =                #canal server连接的zookeeper集群，如：10.20.144.22:2181,10.20.144.51:2181
  16. canal.zookeeper.flush.period = 1000  #数据持久化到zookeeper的周期，默认为1000毫秒
  18. # canal admin配置
  19. canal.admin.manager = 127.0.0.1:8089
  20. canal.admin.port = 11110
  21. canal.admin.user = admin
  22. canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
  24. # admin自动注册
  25. #canal.admin.register.auto = true
  26. #canal.admin.register.cluster =
  27. #canal.admin.register.name =
  29. canal.withoutNetty = false
  31. # canal服务端模式，可选值：tcp, kafka, rocketMQ, rabbitMQ
  32. canal.serverMode = tcp
  34. # flush meta cursor/parse position to file
  35. # 将元数据cursor、parse position保存到文件
  36. canal.file.data.dir = ${canal.conf.dir}
  37. canal.file.flush.period = 1000
  39. # eventStore内存空间设置
  40. canal.instance.memory.batch.mode = MEMSIZE   #ITEMSIZE：buffer.size表示记录数量
  41.                                              #MEMSIZE（默认值）：buffer.size * buffer.memunit限制存储空间大小
  42. canal.instance.memory.buffer.size = 16384    #记录数或者记录大小
  43. canal.instance.memory.buffer.memunit = 1024  #存储单位，默认为1Kb
  44. canal.instance.memory.rawEntry = true        #存储原始字符串，不做序列化处理
  46. ## 心跳检查mysql是否可用
  47. canal.instance.detecting.enable = false      #是否开启心跳检查，默认为false
  48. #canal.instance.detecting.sql = insert into retl.xdual values(1,now()) on duplicate key update x=now()
  49.                                              #心跳检查sql
  50. canal.instance.detecting.interval.time = 3   #心跳检查时间间隔，默认为3
  51. canal.instance.detecting.retry.threshold = 3 #心跳检查重试次数，默认为3
  52. canal.instance.detecting.heartbeatHaEnable = false  #心跳检查mysql不可用时，是否自动切换到内分的数据库
  53.                                                     #默认为false
  55. # support maximum transaction size, more than the size of the transaction will be cut into multiple transactions delivery
  56. # 支持的最大事务长度，超过该长度后，可能会切割存储到eventStore中，无法保证事务的完整可见性
  57. canal.instance.transaction.size =  1024
  60. # mysql fallback connected to new master should fallback times
  61. # canal server切换新的mysql后，需要往前查找binlog的事件，默认为60s
  62. # mysql 主从库同步存在延时，需要往前查找，保证数据不丢失
  63. canal.instance.fallbackIntervalInSeconds = 60
  66. # 网络配置
  67. canal.instance.network.receiveBufferSize = 16384    #canal server接受数据的最大缓存（从mysql解析的数据）
  68. canal.instance.network.sendBufferSize = 16384       #canal server发送数据的最大缓存（发送给canal cilent的数据）
  69. canal.instance.network.soTimeout = 30               #canal server读取数据超时时间，默认为30s
  72. # binlog过滤配置（binlog filter config）
  73. canal.instance.filter.druid.ddl = true              #是否使用druid解析ddl语句，来获取数据库名、表名
  74. canal.instance.filter.query.dcl = false             #是否忽略dcl语句（grant、commit、rollback）
  75. canal.instance.filter.query.dml = false             #是否忽略dml语句（insert、delete、update等）
  76. canal.instance.filter.query.ddl = false             #是否忽略ddl语句（create table、create view等）
  77. canal.instance.filter.table.error = false           #是否忽略binlog表结构获取失败的异常
  78.                                                     #主要解决回溯binlog时，对应的表已被删除，
  79.                                                     #或者表结构和binlog不一致的情况
  80. canal.instance.filter.rows = false                  #是否忽略dml导致的数据变更，默认为false
  81.                                                     #主要针对用户只订阅ddl、dcl操作
  82. canal.instance.filter.transaction.entry = false     #是否忽略事务头，事务尾，默认为false
  83. canal.instance.filter.dml.insert = false            #是否忽略dml insert操作，默认为false
  84. canal.instance.filter.dml.update = false            #是否忽略dml update操作，默认为false
  85. canal.instance.filter.dml.delete = false            #是否忽略dml delete操作，默认为false
  88. # binlog format/image检查（binlog format/image check）
  89. canal.instance.binlog.format = ROW,STATEMENT,MIXED  #默认支持ROW，STATEMENT、MIXED
  90. canal.instance.binlog.image = FULL,MINIMAL,NOBLOB   #默认支持FULL、MINIMAL、NOBLOB
  92. # binlog ddl isolation
  93. canal.instance.get.ddl.isolation = false            #ddl语句是否使用单独的batch返回，默认为false
  94.                                                     #如果和其他ddl/dml在同一batch返回，并发处理时前后顺序不能保证，可能会改变表结构
  96. # 并行配置
  97. canal.instance.parser.parallel = true               #eventParser是否并行解析binlog，默认为true
  98. canal.instance.parser.parallelThreadSize = 16       #并行处理线程数，默认为60%可用线程数
  99. canal.instance.parser.parallelBufferSize = 256      #并行解析的ringBuffer队列数，需为2的指数
  101. # table meta tsdb info
  102. # tableMetaTSDB：处理ddl语句造成的表结构变更
  103. canal.instance.tsdb.enable = true                   #是否开启tablemeta tsdb
  105. # 全局tsdb配置文件
  106. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/h2-tsdb.xml
  107. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
  109. canal.instance.tsdb.dir = ${canal.file.data.dir:../conf}/${canal.instance.destination:}
  110.                                                     #主要针对h2-tsdb.xml时对应h2文件的存放目录,默认为conf/xx/h2.mv.db
  111. canal.instance.tsdb.url =  
  112. jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
  113.                                                     #jdbc url的配置，h2的地址为默认值，如果是mysql需要自行定义
  114. canal.instance.tsdb.dbUsername = canal              #用户名
  115. canal.instance.tsdb.dbPassword = canal              #密码
  116. canal.instance.tsdb.snapshot.interval = 24          #快照间隔，默认为24小时
  117. canal.instance.tsdb.snapshot.expire = 360           #快照失效时间，默认为360小时(15天)
  120. #################################################
  121. #########         destinations        #############
  122. #################################################
  123. canal.destinations = example                        #当前服务器上部署的instance列表
  124. canal.conf.dir = ../conf                            #conf所在目录
  126. # auto scan instance dir add/remove and start/stop instance
  127. canal.auto.scan = true                              #是否开启instance自动扫描，默认为true
  128.                                                     #新增instance目录，自动加载配置，lazy=true时，自动启动
  129.                                                     #删除instance目录时，卸载对应配置，如果已经启动，则停止对应instance
  130. canal.auto.scan.interval = 5                        #自动扫描的时间间隔，默认为5s
  133. # set this value to 'true' means that when binlog pos not found, skip to latest.
  134. # WARN: pls keep 'false' in production env, or if you know what you want.
  135. # true：如果没有找到binlog position，从最新点位开启解析
  136. # 生产环境中建议设置为false
  137. canal.auto.reset.latest.pos.mode = false
  139. canal.instance.global.mode = spring                 #全局配置加载方式，默认为spring（使用本地文件加载）
  140. canal.instance.global.lazy = false                  #全局lazy模式，默认为false
  141. canal.instance.global.manager.address = ${canal.admin.manager}              #全局manager配置地址，可视化配置canal server时使用
  143. # 全局spring.xml配置文件位置，创建CanalInstanceWithSpring对象
  144. canal.instance.global.spring.xml = classpath:spring/memory-instance.xml
  145. canal.instance.global.spring.xml = classpath:spring/file-instance.xml
  146. canal.instance.global.spring.xml = classpath:spring/default-instance.xml
  149. ##################################################
  150. #########           MQ Properties      #############
  151. ##################################################
  152. # aliyun ak/sk , support rds/mq
  153. # 阿里云mq配置
  155. canal.aliyun.accessKey =
  156. canal.aliyun.secretKey =
  157. canal.aliyun.uid=
  159. canal.mq.flatMessage = true
  160. canal.mq.canalBatchSize = 50
  161. canal.mq.canalGetTimeout = 100
  162. # Set this value to "cloud", if you want open message trace feature in aliyun.
  163. canal.mq.accessChannel = local
  165. canal.mq.database.hash = true
  166. canal.mq.send.thread.size = 30
  167. canal.mq.build.thread.size = 8
  170. ##################################################
  171. #########              Kafka              #############
  172. ##################################################
  173. # kafka 配置
  175. kafka.bootstrap.servers = 127.0.0.1:9092
  176. kafka.acks = all
  177. kafka.compression.type = none
  178. kafka.batch.size = 16384
  179. kafka.linger.ms = 1
  180. kafka.max.request.size = 1048576
  181. kafka.buffer.memory = 33554432
  182. kafka.max.in.flight.requests.per.connection = 1
  183. kafka.retries = 0
  185. kafka.kerberos.enable = false
  186. kafka.kerberos.krb5.file = "../conf/kerberos/krb5.conf"
  187. kafka.kerberos.jaas.file = "../conf/kerberos/jaas.conf"
  189. ##################################################
  190. #########             RocketMQ         #############
  191. ##################################################
  192. # rocketMQ 配置
  194. rocketmq.producer.group = test
  195. rocketmq.enable.message.trace = false
  196. rocketmq.customized.trace.topic =
  197. rocketmq.namespace =
  198. rocketmq.namesrv.addr = 127.0.0.1:9876
  199. rocketmq.retry.times.when.send.failed = 0
  200. rocketmq.vip.channel.enabled = false
  201. rocketmq.tag = 
  203. ##################################################
  204. #########             RabbitMQ         #############
  205. ##################################################
  206. # rabbitMQ 配置
  208. rabbitmq.host =
  209. rabbitmq.virtual.host =
  210. rabbitmq.exchange =
  211. rabbitmq.username =
  212. rabbitmq.password =
  213. rabbitmq.deliveryMode =

instance.properties

  1. #################################################
  2. # canal.instance.mysql.slaveId=0       #canal server id，需保证当前mysql集群中该值唯一
  3. canal.instance.gtidon=false            #是否使用mysql gtid订阅模式
  5. # eventParser 解析的master binlog信息
  6. canal.instance.master.address=127.0.0.1:3306    #主机地址
  7. canal.instance.master.journal.name=             #binlog文件
  8. canal.instance.master.position=                 #binlog起始解析的偏移量
  9. canal.instance.master.timestamp=                #binlog起始解析的时间戳
  10. canal.instance.master.gtid=                     #起始解析的gtid位点
  12. # mysql备机信息
  13. canal.instance.standby.address =
  14. canal.instance.standby.journal.name =
  15. canal.instance.standby.position =
  16. canal.instance.standby.timestamp =
  17. canal.instance.standby.gtid=
  20. # rds oss binlog
  21. # 阿里云binlog在18小时候会自动清理上传到阿里云，如果不需要下载oss上的binlog，可不配置
  22. canal.instance.rds.accesskey=                   #阿里云账号的ackKey信息
  23. canal.instance.rds.secretkey=                   #阿里云账号的secretKey信息
  24. canal.instance.rds.instanceId=                  #阿里云账号的instanceId信息
  26. # table meta tsdb info，可覆盖canal properties上的配置
  27. canal.instance.tsdb.enable=true
  28. canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
  29. canal.instance.tsdb.dbUsername=canal
  30. canal.instance.tsdb.dbPassword=canal
  32. # mysql数据库用户名、密码、编码集
  33. canal.instance.dbUsername=canal
  34. canal.instance.dbPassword=canal
  35. canal.instance.connectionCharset = UTF-8
  37. # enable druid Decrypt database password
  38. canal.instance.enableDruid=false
  39. #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
  41. canal.instance.filter.regex=.*\\..*                   #需要解析的表，默认.*\\..*,解析所有表
  42. canal.instance.filter.black.regex=mysql\\.slave_.*    #不需要解析的表
  44. # 示例
  45. .*\\..*：所有表
  46. database\\..*：数据库database下的所有表
  47. database\\.test.*：数据库database下test开头的表
  48. database\\.test：数据库database小的表test
  49. database\\.test.*,database\\.t2：数据库database下test开头的表，database下的t2
  52. # 解析的字段
  53. # table field filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
  54. canal.instance.filter.field=test1.t_product:id/subject/keywords,test2.t_company:id/name/contact/ch
  55. # table field black filter(format: schema1.tableName1:field1/field2,schema2.tableName2:field1/field2)
  56. canal.instance.filter.black.field=test1.t_product:subject/product_image,test2.t_company:id/name/contact/ch
  58. # mq配置
  59. canal.mq.topic=example                                   #静态topic：数据发送到固定topic
  60. canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*  #动态topic：根据数据库名、表名设置
  61. 说明：canal.mq.dynamicTopic格式为schema、schema.tableName、
  62.      topicName:schema、topic:schema.tableName（匹配的数据库、表发送到指定的topicName上）
  64. # 示例
  65. mytest1.user：发送到mytest1_user topic上
  66. mytest2\\..*：发送到mytest2_tableName topic上
  67. test：test库中的所有数据都发送到test topic上
  68. topicName:test\\..*：test下的所有表数据都发送到topicName上
  69. test,test1\\.*：数据库test1中的表发送到test1_tableName topic上
  70.                 数据库test中的所有表发送到test topic上
  71.                 其余所有数据发送到canal.mq.topic指定的topic上
  73. # 发送分区设置
  74. canal.mq.partition=0                                     #发送到固定分区0
  76. # hash partition config
  77. canal.mq.partitionsNum=3                                #分区数
  78. canal.mq.partitionHash=test.table:id^name,.*\\..*       #根据数据库名、表名选择发送的分区
  80. # 为不同的topic动态设置partition num
  81. canal.mq.dynamicTopicPartitionNum=test.*:4,mycanal:6    #动态分区数，
  82.                                                         #test.*:4 test开头topic分区数为4，
  83.                                                          mycanal:6 topic mycanal分区数为6
  85. # 示例
  86. test.table：不设置任何字段，hash字段默认为表名
  87. test.table:id：test.table发送的分区hash字段为id
  88. test.table:id^name：test.table发送的分区使用id、name hash计算得到
  89. .*\\..*:id：所有表的hash字段都为id
  90. .*\\..*:$pk$：所有表的hash字段为表的主键（自动搜索主键）
  91. partitionHash为空：发送到0分区
  92. test.table,test.table2:id：test.table根据tableName hash
  93.                            test.table2根据id hash
  94.                            其余全部发送到对应topic的0分区

*********************

tsdb 相关配置

  1.          ![20210627161332846.png][]

sql：创建存储tsdb信息的表

  1. CREATE TABLE IF NOT EXISTS `meta_snapshot` (
  2.   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  3.   `gmt_create` datetime NOT NULL COMMENT '创建时间',
  4.   `gmt_modified` datetime NOT NULL COMMENT '修改时间',
  5.   `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
  6.   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
  7.   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
  8.   `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id',
  9.   `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳',
  10.   `data` longtext DEFAULT NULL COMMENT '表结构数据',
  11.   `extra` text DEFAULT NULL COMMENT '额外的扩展信息',
  12.   PRIMARY KEY (`id`),
  13.   UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  14.   KEY `destination` (`destination`),
  15.   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  16.   KEY `gmt_modified` (`gmt_modified`)
  17. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='表结构记录表快照表';
  19. CREATE TABLE IF NOT EXISTS `meta_history` (
  20.   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  21.   `gmt_create` datetime NOT NULL COMMENT '创建时间',
  22.   `gmt_modified` datetime NOT NULL COMMENT '修改时间',
  23.   `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
  24.   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
  25.   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
  26.   `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id',
  27.   `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳',
  28.   `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema',
  29.   `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema',
  30.   `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table',
  31.   `sql_text` longtext DEFAULT NULL COMMENT '执行的sql',
  32.   `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型',
  33.   `extra` text DEFAULT NULL COMMENT '额外的扩展信息',
  34.   PRIMARY KEY (`id`),
  35.   UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  36.   KEY `destination` (`destination`),
  37.   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  38.   KEY `gmt_modified` (`gmt_modified`)
  39. ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='表结构变化明细表';

h2-tsdb.xml：使用h2存储tsdb信息

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
  4.     xmlns:aop="http://www.springframework.org/schema/aop" xmlns:lang="http://www.springframework.org/schema/lang"
  5.     xmlns:context="http://www.springframework.org/schema/context"
  6.     xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  7.            http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
  8.            http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-2.0.xsd
  9.            http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd
  10.            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"
  11.     default-autowire="byName">
  13.     <!-- properties -->
  14.     <bean class="com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer" lazy-init="false">
  15.         <property name="ignoreResourceNotFound" value="true" />
  16.         <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/><!-- 允许system覆盖 -->
  17.         <property name="locationNames">
  18.             <list>
  19.                 <value>classpath:canal.properties</value>
  20.                 <value>classpath:${canal.instance.destination:}/instance.properties</value>
  21.             </list>
  22.         </property>
  23.     </bean>
  25.     <!-- 基于db的实现 -->
  26.     <bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta" destroy-method="destory">
  27.         <property name="metaHistoryDAO" ref="metaHistoryDAO"/>
  28.         <property name="metaSnapshotDAO" ref="metaSnapshotDAO"/>
  29.     </bean>
  31.     <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close">
  32.         <property name="driverClassName" value="org.h2.Driver" />
  33.         <property name="url" value="${canal.instance.tsdb.url:}" />
  34.         <property name="username" value="${canal.instance.tsdb.dbUsername:}" />
  35.         <property name="password" value="${canal.instance.tsdb.dbPassword:}" />
  36.           <property name="maxActive" value="30" />
  37.         <property name="initialSize" value="0" />
  38.         <property name="minIdle" value="1" />
  39.         <property name="maxWait" value="10000" />
  40.         <property name="timeBetweenEvictionRunsMillis" value="60000" />
  41.         <property name="minEvictableIdleTimeMillis" value="300000" />
  42.         <property name="testWhileIdle" value="true" />
  43.         <property name="testOnBorrow" value="false" />
  44.         <property name="testOnReturn" value="false" />
  45.         <property name="useUnfairLock" value="true" />
  46.         <property name="validationQuery" value="SELECT 1" />
  47.     </bean>
  49.     <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
  50.         <property name="dataSource" ref="dataSource"/>
  51.         <property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/>
  52.     </bean>
  54.     <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
  55.         <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
  56.     </bean>
  58.     <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
  59.         <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
  60.     </bean>
  61. </beans>

mysql-tsdb.xml：使用mysql存储tsdb信息

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans"
  3.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
  4.     xmlns:aop="http://www.springframework.org/schema/aop" xmlns:lang="http://www.springframework.org/schema/lang"
  5.     xmlns:context="http://www.springframework.org/schema/context"
  6.     xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
  7.            http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.0.xsd
  8.            http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-2.0.xsd
  9.            http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.0.xsd
  10.            http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd"
  11.     default-autowire="byName">
  13.     <!-- properties -->
  14.     <bean class="com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer" lazy-init="false">
  15.         <property name="ignoreResourceNotFound" value="true" />
  16.         <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/><!-- 允许system覆盖 -->
  17.         <property name="locationNames">
  18.             <list>
  19.                 <value>classpath:canal.properties</value>
  20.                 <value>classpath:${canal.instance.destination:}/instance.properties</value>
  21.             </list>
  22.         </property>
  23.     </bean>
  25.     <!-- 基于db的实现 -->
  26.     <bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta" destroy-method="destory">
  27.         <property name="metaHistoryDAO" ref="metaHistoryDAO"/>
  28.         <property name="metaSnapshotDAO" ref="metaSnapshotDAO"/>
  29.     </bean>
  31.     <bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close">
  32.         <property name="driverClassName" value="com.mysql.jdbc.Driver" />
  33.         <property name="url" value="${canal.instance.tsdb.url:}" />
  34.         <property name="username" value="${canal.instance.tsdb.dbUsername:}" />
  35.         <property name="password" value="${canal.instance.tsdb.dbPassword:}" />
  36.         <property name="maxActive" value="30" />
  37.         <property name="initialSize" value="0" />
  38.         <property name="minIdle" value="1" />
  39.         <property name="maxWait" value="10000" />
  40.         <property name="timeBetweenEvictionRunsMillis" value="60000" />
  41.         <property name="minEvictableIdleTimeMillis" value="300000" />
  42.         <property name="validationQuery" value="SELECT 1" />
  43.         <property name="exceptionSorterClassName" value="com.alibaba.druid.pool.vendor.MySqlExceptionSorter" />
  44.         <property name="validConnectionCheckerClassName" value="com.alibaba.druid.pool.vendor.MySqlValidConnectionChecker" />
  45.         <property name="testWhileIdle" value="true" />
  46.         <property name="testOnBorrow" value="false" />
  47.         <property name="testOnReturn" value="false" />
  48.         <property name="useUnfairLock" value="true" />
  49.     </bean>
  51.     <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
  52.         <property name="dataSource" ref="dataSource"/>
  53.         <property name="configLocation" value="classpath:spring/tsdb/sql-map/sqlmap-config.xml"/>
  54.     </bean>
  56.     <bean id="metaHistoryDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO">
  57.         <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
  58.     </bean>
  60.     <bean id="metaSnapshotDAO" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO">
  61.         <property name="sqlSessionFactory" ref="sqlSessionFactory"/>
  62.     </bean>
  63. </beans>

*********************

canal server 单机

创建mysql

  1. [root@centos ~]# docker run -it -d --net fixed --ip 172.18.0.2 \
  2. > -p 3306:3306 --privileged=true \
  3. > --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
  5. # mysql8 bin_log默认开启
  6. mysql> show variables like "log_bin";
  7. +---------------+-------+
  8. | Variable_name | Value |
  9. +---------------+-------+
  10. | log_bin       | ON    |
  11. +---------------+-------+
  12. 1 row in set (0.00 sec)
  14. # mysql8 binlog_format默认为ROW
  15. mysql> show variables like "binlog_format";
  16. +---------------+-------+
  17. | Variable_name | Value |
  18. +---------------+-------+
  19. | binlog_format | ROW   |
  20. +---------------+-------+
  21. 1 row in set (0.01 sec)

创建用户并授予权限，供canal server连接使用

  1. mysql> create user canal identified with mysql_native_password by "123456";
  2. Query OK, 0 rows affected (0.11 sec)
  4. mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  5. Query OK, 0 rows affected (0.01 sec)
  7. mysql> flush privileges;
  8. Query OK, 0 rows affected (0.00 sec)

创建canal server

  1. docker run -it -d --net fixed --ip 172.18.0.3 \
  2. -p 11111:11111 --name canal-server \
  3. -e canal.instance.master.address=172.18.0.2:3306  \
  4. -e canal.instance.dbUsername=canal \
  5. -e canal.instance.dbPassword=123456 canal/canal-server

查看canal server日志

  1. [root@centos ~]# docker logs canal-server
  2. DOCKER_DEPLOY_TYPE=VM
  3. ==> INIT /alidata/init/02init-sshd.sh
  4. ==> EXIT CODE: 0
  5. ==> INIT /alidata/init/fix-hosts.py
  6. ==> EXIT CODE: 0
  7. ==> INIT DEFAULT
  8. Generating SSH1 RSA host key:                              [  OK  ]
  9. Starting sshd:                                             [  OK  ]
  10. Starting crond:                                            [  OK  ]
  11. ==> INIT DONE
  12. ==> RUN /home/admin/app.sh
  13. ==> START ...
  14. start canal ...
  15. start canal successful

查看instance 日志

  1. [root@centos ~]# docker exec -it canal-server bash
  2. [root@55e591394ef4 admin]# cd canal-server/logs/example
  3. [root@55e591394ef4 example]# ls
  4. example.log
  5. [root@55e591394ef4 example]# cat example.log
  6. 2021-06-29 01:03:54.889 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
  7. 2021-06-29 01:03:54.954 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
  8. 2021-06-29 01:03:54.954 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
  9. 2021-06-29 01:03:55.025 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
  10. 2021-06-29 01:03:55.378 [destination = example , address = /172.18.0.2:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
  11. 2021-06-29 01:03:55.379 [destination = example , address = /172.18.0.2:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status
  12. 2021-06-29 01:03:57.678 [destination = example , address = /172.18.0.2:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=binlog.000002,position=4,serverId=1,gtid=<null>,timestamp=1624899604000] cost : 2206ms , the next step is binlog dump

说明：instance成功连接上MySQL，并开始解析数据源

*********************

canal server 单机多instance

修改 canal.properties

  1. canal.destinations = example,example2

修改 example/instance.properties

  1. canal.instance.master.address=172.18.0.11:3306
  2. canal.instance.dbUsername=canal
  3. canal.instance.dbPassword=123456

修改 example2/instance.properties

  1. canal.instance.master.address=172.18.0.12:3306
  2. canal.instance.dbUsername=canal
  3. canal.instance.dbPassword=123456

创建 mysql实例

  1. docker run -it -d --net fixed --ip 172.18.0.11 -p 3306:3306 --privileged=true \
  2. --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
  4. docker run -it -d --net fixed --ip 172.18.0.12 -p 3307:3306 --privileged=true \
  5. --name mysql2 -e  MYSQL_ROOT_PASSWORD=123456 mysql
  8. # mysql、mysql2 创建用户并授权
  9. create user canal identified with mysql_native_password by "123456";
  10. GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  11. flush privileges;

创建 canal实例

  1. docker run -it -d --net fixed --ip 172.18.0.3 -p 11111:11111 --name canal-server \
  2. -v /usr/canal/single/conf:/home/admin/canal-server/conf  canal/canal-server
  5. # 宿主机目录
  6. [root@centos conf]# pwd
  7. /usr/canal/single/conf
  8. [root@centos conf]# ls
  9. canal_local.properties  canal.properties  example  example2  logback.xml  metrics  spring

查看canal server日志

  1. [root@centos example]# docker logs canal-server
  2. DOCKER_DEPLOY_TYPE=VM
  3. ==> INIT /alidata/init/02init-sshd.sh
  4. ==> EXIT CODE: 0
  5. ==> INIT /alidata/init/fix-hosts.py
  6. ==> EXIT CODE: 0
  7. ==> INIT DEFAULT
  8. Generating SSH1 RSA host key:                              [  OK  ]
  9. Starting sshd:                                             [  OK  ]
  10. Starting crond:                                            [  OK  ]
  11. ==> INIT DONE
  12. ==> RUN /home/admin/app.sh
  13. ==> START ...
  14. start canal ...
  15. start canal successful
  16. ==> START SUCCESSFUL ...

canal server实例创建成功

查看 instance实例日志

  1. [root@centos example]# docker exec -it canal-server bash
  2. [root@ac38c13bce07 admin]# cd canal-server/logs
  3. [root@ac38c13bce07 logs]# pwd
  4. /home/admin/canal-server/logs
  5. [root@ac38c13bce07 logs]# ls
  6. canal  example  example2
  8. [root@ac38c13bce07 logs]# ls canal
  9. canal.log  canal_stdout.log
  10. [root@ac38c13bce07 logs]# ls example
  11. example.log
  12. [root@ac38c13bce07 logs]# ls example2
  13. example2.log
  15. # instance日志
  16. [root@ac38c13bce07 logs]# cat example/example.log
  17. 2021-06-29 22:53:04.947 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
  18. 2021-06-29 22:53:05.089 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
  19. 2021-06-29 22:53:05.089 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
  20. 2021-06-29 22:53:05.153 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
  21. 2021-06-29 22:53:05.958 [destination = example , address = /172.18.0.11:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
  22. 2021-06-29 22:53:05.969 [destination = example , address = /172.18.0.11:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status
  23. 2021-06-29 22:53:12.659 [destination = example , address = /172.18.0.11:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=binlog.000002,position=4,serverId=1,gtid=<null>,timestamp=1624978095000] cost : 6604ms , the next step is binlog dump
  25. # instance2 日志
  26. [root@ac38c13bce07 logs]# cat example2/example2.log
  27. 2021-06-29 22:53:06.677 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example2 
  28. 2021-06-29 22:53:06.717 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
  29. 2021-06-29 22:53:06.717 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
  30. 2021-06-29 22:53:06.735 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
  31. 2021-06-29 22:53:06.918 [destination = example2 , address = /172.18.0.12:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
  32. 2021-06-29 22:53:06.934 [destination = example2 , address = /172.18.0.12:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status
  33. 2021-06-29 22:53:12.660 [destination = example2 , address = /172.18.0.12:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=binlog.000002,position=4,serverId=1,gtid=<null>,timestamp=1624978126000] cost : 5665ms , the next step is binlog dump

instance、instance2创建成功

*********************

canal server 单机 mysql存储tsdb

创建 mysql、mysql2实例

  1. docker run -it -d --net fixed --ip 172.18.0.21 -p 3306:3306 --privileged=true \
  2. --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
  4. docker run -it -d --net fixed --ip 172.18.0.20 -p 3307:3306 --privileged=true \
  5. --name mysql2 -e  MYSQL_ROOT_PASSWORD=123456 mysql
  7. *************
  8. mysql：源数据
  10. # 创建用户，添加权限
  11. mysql> create user canal identified with mysql_native_password by "123456";
  12. Query OK, 0 rows affected (0.01 sec)
  14. mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  15. Query OK, 0 rows affected (0.00 sec)
  17. mysql> flush privileges;
  18. Query OK, 0 rows affected (0.01 sec)
  22. *************
  23. mysql2：存储tsdb
  25. # 创建用户，添加权限
  26. mysql> create user canal identified with mysql_native_password by "123456";
  27. Query OK, 0 rows affected (0.00 sec)
  29. mysql> GRANT ALL ON *.* TO 'canal'@'%';
  30. Query OK, 0 rows affected (0.01 sec)
  32. mysql> flush privileges;
  33. Query OK, 0 rows affected (0.00 sec)
  35. # 创建数据库、表
  36. mysql> create database example;
  37. Query OK, 1 row affected (0.00 sec)
  39. mysql> use example;
  40. Database changed
  41. mysql> CREATE TABLE IF NOT EXISTS `meta_snapshot` (
  42.     ->   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '',
  43.     ->   `gmt_create` datetime NOT NULL COMMENT '',
  44.     ->   `gmt_modified` datetime NOT NULL COMMENT '',
  45.     ->   `destination` varchar(128) DEFAULT NULL COMMENT '',
  46.     ->   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog',
  47.     ->   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog',
  48.     ->   `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlogid',
  49.     ->   `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog',
  50.     ->   `data` longtext DEFAULT NULL COMMENT '',
  51.     ->   `extra` text DEFAULT NULL COMMENT '',
  52.     ->   PRIMARY KEY (`id`),
  53.     ->   UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  54.     ->   KEY `destination` (`destination`),
  55.     ->   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  56.     ->   KEY `gmt_modified` (`gmt_modified`)
  57.     -> ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='';
  58. 改时间',
  59.   `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
  60.   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
  61.   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
  62.   `binlog_master_id` varchar(64) DEFAULT Query OK, 0 rows affected, 3 warnings (0.04 sec)
  64. mysql>  
  65. mysql> CREATE TABLE IF NOT EXISTS `meta_history` (
  66.     ->   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '',
  67.     ->   `gmt_create` datetime NOT NULL COMMENT '',
  68.     ->   `gmt_modified` datetime NOT NULL COMMENT '',
  69.     ->   `destination` varchar(128) DEFAULT NULL COMMENT '',
  70.     ->   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog',
  71.     ->   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog',
  72.     ->   `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlogid',
  73.     ->   `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog',
  74.     ->   `use_schema` varchar(1024) DEFAULT NULL COMMENT 'sqlschema',
  75.     ->   `sql_schema` varchar(1024) DEFAULT NULL COMMENT 'schema',
  76.     ->   `sql_table` varchar(1024) DEFAULT NULL COMMENT 'table',
  77.     ->   `sql_text` longtext DEFAULT NULL COMMENT 'sql',
  78.     ->   `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql',
  79.     ->   `extra` text DEFAULT NULL COMMENT '',
  80.     ->   PRIMARY KEY (`id`),
  81.     ->   UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  82.     ->   KEY `destination` (`destination`),
  83.     ->   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  84.     ->   KEY `gmt_modified` (`gmt_modified`)
  85.     -> ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='';
  86. Query OK, 0 rows affected, 3 warnings (0.02 sec)

修改配置文件

  1. # canal.properties
  2. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
  5. # example/instance.properties
  6. canal.instance.master.address=172.18.0.11:3306
  7. canal.instance.dbUsername=canal
  8. canal.instance.dbPassword=123456
  10. canal.instance.tsdb.enable=true
  11. canal.instance.tsdb.url=jdbc:mysql://172.18.0.20:3306/example
  12. canal.instance.tsdb.dbUsername=canal
  13. canal.instance.tsdb.dbPassword=123456

创建 canal server实例

  1. docker run -it -d --net fixed --ip 172.18.0.23 -p 11111:11111 --name canal-server \
  2. -v /usr/canal/single/tsdb/conf:/home/admin/canal-server/conf  canal/canal-server

查看canal server日志

  1. [root@centos tsdb]# docker logs canal-server
  2. DOCKER_DEPLOY_TYPE=VM
  3. ==> INIT /alidata/init/02init-sshd.sh
  4. ==> EXIT CODE: 0
  5. ==> INIT /alidata/init/fix-hosts.py
  6. ==> EXIT CODE: 0
  7. ==> INIT DEFAULT
  8. Generating SSH1 RSA host key:                              [  OK  ]
  9. Starting sshd:                                             [  OK  ]
  10. Starting crond:                                            [  OK  ]
  11. ==> INIT DONE
  12. ==> RUN /home/admin/app.sh
  13. ==> START ...
  14. start canal ...
  15. start canal successful
  16. ==> START SUCCESSFUL ...

canal server启动成功

查看instance日志

  1. [root@centos tsdb]# docker exec -it canal-server bash
  2. [root@69b22f3fc434 admin]# cd canal-server/logs/example
  3. [root@69b22f3fc434 example]# cat example.log
  4. 2021-06-30 00:43:57.172 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
  5. 2021-06-30 00:43:57.205 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
  6. 2021-06-30 00:43:57.205 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
  7. 2021-06-30 00:43:57.213 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
  8. 2021-06-30 00:43:57.410 [destination = example , address = /172.18.0.21:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
  9. 2021-06-30 00:43:57.410 [destination = example , address = /172.18.0.21:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status
  10. 2021-06-30 00:43:59.387 [destination = example , address = /172.18.0.21:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=binlog.000002,position=4,serverId=1,gtid=<null>,timestamp=1624984902000] cost : 1966ms , the next step is binlog dump

instance创建成功

*********************

canal server 集群

mysql：元数据

mysql2：存储内存表快照

zookeeper：存储集群元数据

canal-server、canal-server2：canal server集群

创建 mysql实例

  1. #mysql：源数据
  2. docker run -it -d --net fixed --ip 172.18.0.31 -p 3306:3306 --privileged=true \
  3. --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
  5. #mysql2：存储内存表快照数据
  6. docker run -it -d --net fixed --ip 172.18.0.32 -p 3307:3306 --privileged=true \
  7. --name mysql2 -e  MYSQL_ROOT_PASSWORD=123456 mysql
  10. **************
  11. mysql创建用户、授权
  13. mysql> create user canal identified with mysql_native_password by "123456";
  14. Query OK, 0 rows affected (0.01 sec)
  16. mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  17. Query OK, 0 rows affected (0.00 sec)
  19. mysql> flush privileges;
  20. Query OK, 0 rows affected (0.00 sec)
  23. **************
  24. mysql2创建用户、授权，创建数据库、表
  26. mysql> create user canal identified with mysql_native_password by "123456";
  27. Query OK, 0 rows affected (0.01 sec)
  29. mysql> GRANT ALL ON *.* TO 'canal'@'%';
  30. Query OK, 0 rows affected (0.00 sec)
  32. mysql> flush privileges;
  33. Query OK, 0 rows affected (0.01 sec)
  35. mysql> create database example;
  36. Query OK, 1 row affected (0.01 sec)
  38. mysql> use example;
  39. Database changed
  40. mysql> CREATE TABLE IF NOT EXISTS `meta_snapshot` (
  41.     ->   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '',
  42.     ->   `gmt_create` datetime NOT NULL COMMENT '',
  43.     ->   `gmt_modified` datetime NOT NULL COMMENT '',
  44.     ->   `destination` varchar(128) DEFAULT NULL COMMENT '',
  45.     ->   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog',
  46.     ->   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog',
  47.     ->   `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlogid',
  48.     ->   `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog',
  49.     ->   `data` longtext DEFAULT NULL COMMENT '',
  50.     ->   `extra` text DEFAULT NULL COMMENT '',
  51.     ->   PRIMARY KEY (`id`),
  52.     ->   UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  53.     ->   KEY `destination` (`destination`),
  54.     ->   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  55.     ->   KEY `gmt_modified` (`gmt_modified`)
  56.     -> ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='';
  57. 改时间',
  58.   `destination` varchar(128) DEFAULT NULL COMMENT '通道名称',
  59.   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名',
  60.   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量',
  61.   `binlog_master_id` varchar(64) DEFAULT Query OK, 0 rows affected, 3 warnings (0.02 sec)
  63. mysql>  
  64. mysql> CREATE TABLE IF NOT EXISTS `meta_history` (
  65.     ->   `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '',
  66.     ->   `gmt_create` datetime NOT NULL COMMENT '',
  67.     ->   `gmt_modified` datetime NOT NULL COMMENT '',
  68.     ->   `destination` varchar(128) DEFAULT NULL COMMENT '',
  69.     ->   `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog',
  70.     ->   `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog',
  71.     ->   `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlogid',
  72.     ->   `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog',
  73.     ->   `use_schema` varchar(1024) DEFAULT NULL COMMENT 'sqlschema',
  74.     ->   `sql_schema` varchar(1024) DEFAULT NULL COMMENT 'schema',
  75.     ->   `sql_table` varchar(1024) DEFAULT NULL COMMENT 'table',
  76.     ->   `sql_text` longtext DEFAULT NULL COMMENT 'sql',
  77.     ->   `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql',
  78.     ->   `extra` text DEFAULT NULL COMMENT '',
  79.     ->   PRIMARY KEY (`id`),
  80.     ->   UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`),
  81.     ->   KEY `destination` (`destination`),
  82.     ->   KEY `destination_timestamp` (`destination`,`binlog_timestamp`),
  83.     ->   KEY `gmt_modified` (`gmt_modified`)
  84.     -> ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='';
  85. Query OK, 0 rows affected, 3 warnings (0.02 sec)

创建 zookeeper实例

  1. docker run -it -d --net fixed --ip 172.18.0.33 -p 2181:2181 --name zoo zookeeper

修改 canal配置文件

  1. # canal.properties
  2. canal.zkServers=172.18.0.33:2181
  3. canal.instance.tsdb.spring.xml = classpath:spring/tsdb/mysql-tsdb.xml
  4. canal.instance.global.spring.xml = classpath:spring/default-instance.xml
  7. # instance.properties
  8. canal.instance.master.address=172.18.0.31:3306
  9. canal.instance.dbUsername=canal
  10. canal.instance.dbPassword=123456
  12. canal.instance.tsdb.enable=true
  13. canal.instance.tsdb.url=jdbc:mysql://172.18.0.32:3306/example
  14. canal.instance.tsdb.dbUsername=canal
  15. canal.instance.tsdb.dbPassword=123456

创建 canal server集群

  1. docker run -it -d --net fixed --ip 172.18.0.34 -p 11111:11111 --name canal-server \
  2. -v /usr/canal/cluster/server/conf:/home/admin/canal-server/conf  canal/canal-server
  4. docker run -it -d --net fixed --ip 172.18.0.35 -p 11112:11111 --name canal-server2 \
  5. -v /usr/canal/cluster/server2/conf:/home/admin/canal-server/conf  canal/canal-server

查看 canal server日志

  1. # canal-server
  2. [root@centos ~]# docker logs canal-server
  3. DOCKER_DEPLOY_TYPE=VM
  4. ==> INIT /alidata/init/02init-sshd.sh
  5. ==> EXIT CODE: 0
  6. ==> INIT /alidata/init/fix-hosts.py
  7. ==> EXIT CODE: 0
  8. ==> INIT DEFAULT
  9. Generating SSH1 RSA host key:                              [  OK  ]
  10. Starting sshd:                                             [  OK  ]
  11. Starting crond:                                            [  OK  ]
  12. ==> INIT DONE
  13. ==> RUN /home/admin/app.sh
  14. ==> START ...
  15. start canal ...
  16. start canal successful
  17. ==> START SUCCESSFUL ...
  20. # canal-server2
  21. [root@centos ~]# docker logs canal-server2
  22. DOCKER_DEPLOY_TYPE=VM
  23. ==> INIT /alidata/init/02init-sshd.sh
  24. ==> EXIT CODE: 0
  25. ==> INIT /alidata/init/fix-hosts.py
  26. ==> EXIT CODE: 0
  27. ==> INIT DEFAULT
  28. Generating SSH1 RSA host key:                              [  OK  ]
  29. Starting sshd:                                             [  OK  ]
  30. Starting crond:                                            [  OK  ]
  31. ==> INIT DONE
  32. ==> RUN /home/admin/app.sh
  33. ==> START ...
  34. start canal ...
  35. start canal successful
  36. ==> START SUCCESSFUL ...

canal-server、canal-server2 启动成功

查看 instance日志

  1. # canal-server
  2. [root@centos ~]# docker exec -it canal-server bash
  3. [root@ddcb9672751c admin]# cd canal-server/logs
  4. [root@ddcb9672751c logs]# ls
  5. canal
  8. # canal-server2
  9. [root@centos cluster]# docker exec -it canal-server2 bash
  10. [root@e54128f28507 admin]# cd canal-server/logs
  11. [root@e54128f28507 logs]# ls
  12. canal  example
  13. [root@e54128f28507 logs]# cd example
  14. [root@e54128f28507 example]# ls
  15. example.log
  16. [root@e54128f28507 example]# cat example.log
  17. 2021-06-30 22:42:37.377 [ZkClient-EventThread-10-172.18.0.33:2181] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
  18. 2021-06-30 22:42:37.497 [ZkClient-EventThread-10-172.18.0.33:2181] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
  19. 2021-06-30 22:42:37.498 [ZkClient-EventThread-10-172.18.0.33:2181] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
  20. 2021-06-30 22:42:37.597 [ZkClient-EventThread-10-172.18.0.33:2181] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
  21. 2021-06-30 22:42:38.351 [destination = example , address = /172.18.0.31:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
  22. 2021-06-30 22:42:38.359 [destination = example , address = /172.18.0.31:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status
  23. 2021-06-30 22:42:40.622 [destination = example , address = /172.18.0.31:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=binlog.000002,position=4,serverId=1,gtid=<null>,timestamp=1625063533000] cost : 2206ms , the next step is binlog dump

canal-server没有启动，处于Standby状态

canal-server2启动成功，可以看到启动日志

zookeeper中存储的元数据

  1. [root@centos cluster]# docker exec -it zoo bash
  2. root@bfa8dada58dc:/apache-zookeeper-3.6.2-bin# bin/zkCli.sh
  3. Connecting to localhost:2181
  4. 2021-06-30 15:11:46,453 [myid:] - INFO  [main:Environment@98] - Client environment:zookeeper.version=3.6.2--803c7f1a12f85978cb049af5e4ef23bd8b688715, built on 09/04/2020 12:44 GMT
  5. 2021-06-30 15:11:46,465 [myid:] - INFO  [main:Environment@98] - Client environment:host.name=bfa8dada58dc
  6. 2021-06-30 15:11:46,466 [myid:] - INFO  [main:Environment@98] - Client environment:java.version=11.0.8
  7. 2021-06-30 15:11:46,472 [myid:] - INFO  [main:Environment@98] - Client environment:java.vendor=N/A
  8. 2021-06-30 15:11:46,472 [myid:] - INFO  [main:Environment@98] - Client environment:java.home=/usr/local/openjdk-11
  9. 2021-06-30 15:11:46,473 [myid:] - INFO  [main:Environment@98] - Client environment:java.class.path=/apache-zookeeper-3.6.2-bin/bin/../zookeeper-server/target/classes:/apache-zookeeper-3.6.2-bin/bin/../build/classes:/apache-zookeeper-3.6.2-bin/bin/../zookeeper-server/target/lib/*.jar:/apache-zookeeper-3.6.2-bin/bin/../build/lib/*.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/zookeeper-prometheus-metrics-3.6.2.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/zookeeper-jute-3.6.2.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/zookeeper-3.6.2.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/snappy-java-1.1.7.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/slf4j-log4j12-1.7.25.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/slf4j-api-1.7.25.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/simpleclient_servlet-0.6.0.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/simpleclient_hotspot-0.6.0.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/simpleclient_common-0.6.0.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/simpleclient-0.6.0.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-transport-native-unix-common-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-transport-native-epoll-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-transport-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-resolver-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-handler-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-common-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-codec-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/netty-buffer-4.1.50.Final.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/metrics-core-3.2.5.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/log4j-1.2.17.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/json-simple-1.1.1.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jline-2.14.6.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jetty-util-9.4.24.v20191120.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jetty-servlet-9.4.24.v20191120.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jetty-server-9.4.24.v20191120.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jetty-security-9.4.24.v20191120.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jetty-io-9.4.24.v20191120.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jetty-http-9.4.24.v20191120.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/javax.servlet-api-3.1.0.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jackson-databind-2.10.3.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jackson-core-2.10.3.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/jackson-annotations-2.10.3.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/commons-lang-2.6.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/commons-cli-1.2.jar:/apache-zookeeper-3.6.2-bin/bin/../lib/audience-annotations-0.5.0.jar:/apache-zookeeper-3.6.2-bin/bin/../zookeeper-*.jar:/apache-zookeeper-3.6.2-bin/bin/../zookeeper-server/src/main/resources/lib/*.jar:/conf:
  10. 2021-06-30 15:11:46,479 [myid:] - INFO  [main:Environment@98] - Client environment:java.library.path=/usr/java/packages/lib:/usr/lib64:/lib64:/lib:/usr/lib
  11. 2021-06-30 15:11:46,480 [myid:] - INFO  [main:Environment@98] - Client environment:java.io.tmpdir=/tmp
  12. 2021-06-30 15:11:46,480 [myid:] - INFO  [main:Environment@98] - Client environment:java.compiler=<NA>
  13. 2021-06-30 15:11:46,480 [myid:] - INFO  [main:Environment@98] - Client environment:os.name=Linux
  14. 2021-06-30 15:11:46,480 [myid:] - INFO  [main:Environment@98] - Client environment:os.arch=amd64
  15. 2021-06-30 15:11:46,481 [myid:] - INFO  [main:Environment@98] - Client environment:os.version=3.10.0-957.el7.x86_64
  16. 2021-06-30 15:11:46,481 [myid:] - INFO  [main:Environment@98] - Client environment:user.name=root
  17. 2021-06-30 15:11:46,481 [myid:] - INFO  [main:Environment@98] - Client environment:user.home=/root
  18. 2021-06-30 15:11:46,485 [myid:] - INFO  [main:Environment@98] - Client environment:user.dir=/apache-zookeeper-3.6.2-bin
  19. 2021-06-30 15:11:46,486 [myid:] - INFO  [main:Environment@98] - Client environment:os.memory.free=21MB
  20. 2021-06-30 15:11:46,492 [myid:] - INFO  [main:Environment@98] - Client environment:os.memory.max=247MB
  21. 2021-06-30 15:11:46,493 [myid:] - INFO  [main:Environment@98] - Client environment:os.memory.total=29MB
  22. 2021-06-30 15:11:46,514 [myid:] - INFO  [main:ZooKeeper@1006] - Initiating client connection, connectString=localhost:2181 sessionTimeout=30000 watcher=org.apache.zookeeper.ZooKeeperMain$MyWatcher@6166e06f
  23. 2021-06-30 15:11:46,524 [myid:] - INFO  [main:X509Util@77] - Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation
  24. 2021-06-30 15:11:46,547 [myid:] - INFO  [main:ClientCnxnSocket@239] - jute.maxbuffer value is 1048575 Bytes
  25. 2021-06-30 15:11:46,589 [myid:] - INFO  [main:ClientCnxn@1716] - zookeeper.request.timeout value is 0. feature enabled=false
  26. Welcome to ZooKeeper!
  27. 2021-06-30 15:11:46,687 [myid:localhost:2181] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1167] - Opening socket connection to server localhost/127.0.0.1:2181.
  28. 2021-06-30 15:11:46,688 [myid:localhost:2181] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1169] - SASL config status: Will not attempt to authenticate using SASL (unknown error)
  29. JLine support is enabled
  30. 2021-06-30 15:11:46,761 [myid:localhost:2181] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@999] - Socket connection established, initiating session, client: /127.0.0.1:43956, server: localhost/127.0.0.1:2181
  31. 2021-06-30 15:11:46,827 [myid:localhost:2181] - INFO  [main-SendThread(localhost:2181):ClientCnxn$SendThread@1433] - Session establishment complete on server localhost/127.0.0.1:2181, session id = 0x100002977550009, negotiated timeout = 30000
  33. WATCHER::
  35. WatchedEvent state:SyncConnected type:None path:null
  36. [zk: localhost:2181(CONNECTED) 0] get /otter/canal/destinations/example/running
  37. {"active":true,"address":"172.18.0.35:11111"}

当前运行的节点信息：{“active”:true,”address”:”172.18.0.35:11111”}

发表评论

表情：
评论列表 （有 0 条评论，514人围观）

还没有评论，来说两句吧...

相关阅读

    相关 docker环境搭建

    docker有免费版本和收费的企业版本，企业版本无非是增加了一些安全性方面的集成，一般企业会选择docker CE免费版本,安全性方面，绝大部分企业都是自己实现内部的安全集成。

    柔情只为你懂柔情只为你懂/ 0 赞/ 498 阅读

    相关 Docker：环境搭建

    一，虚拟机环境搭建     \ Linux镜像要求CentOS版本为7，且内核大于3.10     \ 具体安装步骤不做演示，只对几个遇到的问题进行说明        

    忘是亡心i忘是亡心i/ 0 赞/ 634 阅读