Canal 安装

用户1483438

发布于 2022-07-26 21:42:05

5670

发布于 2022-07-26 21:42:05

文章被收录于专栏：大数据共享大数据共享

简单介绍

下载

官网最新版本我本地版本canal-1.1.2

解压安装

在解压后的目录中创建一个canal目录原因下面说

[admin@hadoop102 software]$ mkdir /opt/module/canal

 tar -zxvf canal.deployer-1.1.2.tar.gz -C /opt/module/canal

查看/opt/module/canal目录之所以需要创建一个目录，就是因为canal解压之后，就是四个独立的文件(bin，conf，lib，logs)，并没有被包装起来。

[admin@hadoop102 software]$ ll /opt/module/canal
总用量 4
drwxrwxr-x. 2 admin admin 76 7月  31 19:18 bin
drwxrwxr-x. 5 admin admin 93 7月  31 19:18 conf
drwxrwxr-x. 2 admin admin 4096 7月  31 19:18 lib
drwxrwxr-x. 2 admin admin 6 11月 26 2018 logs

conf

-rwxrwxr-x. 1 admin admin 4262 11月 26 2018 canal.properties
drwxrwxr-x. 2 admin admin 33 7月  31 19:18 example
-rwxrwxr-x. 1 admin admin 3109 11月 26 2018 logback.xml
drwxrwxr-x. 2 admin admin 39 7月  31 19:18 metrics
drwxrwxr-x. 3 admin admin 149 7月  31 19:18 spring

有两个很重要的文件第一个就是：canal.properties 第二个就是：example目录下的instance.properties 文件

-rwxrwxr-x. 1 admin admin 1572 11月 26 2018 instance.properties

了解上面两个配置文件之前先看一张图

Canal Server：就是我们现在安装的Canal服务。目前Canal是可以将数据写入到一些中间中，比如Kafka、RabbitMQ，ActiveMQ该版本应该是不支持的。

canal.properties：就是用于做全局配置
instance.properties：是可以有多个的，你需要订阅几个mysql的实例，就配置几个instance.properties即可。比如你有两台服务器，都装了mysql分别为A和B。那么你可以配置两个instance.properties一个监听A服务器，一个监听B服务器。至于怎么配置，后面说。 instance.properties ：包括mysql服务器的账号密码，ip端口等信息。

配置 canal.properties

canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk
canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = tcp
# 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 = false
canal.instance.filter.query.ddl = false
canal.instance.filter.table.error = false
canal.instance.filter.rows = false
canal.instance.filter.transaction.entry = 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

# aliyun ak/sk , support rds/mq
canal.aliyun.accesskey =
canal.aliyun.secretkey =

#################################################
#########               destinations            #############
#################################################
canal.destinations = example
# 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

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 = 127.0.0.1:1099
#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                      #############
##################################################
canal.mq.servers = 127.0.0.1:6667
canal.mq.retries = 0
canal.mq.batchSize = 16384
canal.mq.maxRequestSize = 1048576
canal.mq.lingerMs = 1
canal.mq.bufferMemory = 33554432
canal.mq.canalBatchSize = 50
canal.mq.canalGetTimeout = 100
canal.mq.flatMessage = true
canal.mq.compressionType = none
canal.mq.acks = all

若开启了中间件，那么要将canal.serverMode改成kafka或者RocketMQ

#tcp, kafka, RocketMQ
canal.serverMode = tcp

目前只需要改一个地方，设置canal.ip，将其设置为当前服务器的IP即可

canal.ip =192.168.102.55

配置 instance.properties

#################################################
## mysql serverId , v1.0.26+ will autoGen
# canal.instance.mysql.slaveId=0

# enable gtid use true/false
canal.instance.gtidon=false

# position info
canal.instance.master.address=127.0.0.1:3306
canal.instance.master.journal.name=
canal.instance.master.position=
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
canal.instance.defaultDatabaseName =test
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=

# mq config
canal.mq.topic=example
canal.mq.partition=0
# hash partition config
#canal.mq.partitionsNum=3
#canal.mq.partitionHash=mytest.person:id,mytest.role:id
#################################################

配置要订阅的mysql的实例信息

打开canal.instance.mysql.slaveId，用于配置从机(slave)Id。必须与mysql中的server-id不同。

canal.instance.mysql.slaveId=2

position info 用于配置从主机那个位置开始订阅消息（图1959）

mysql-bin.* 文件，当mysql重启时会滚动一个文件，或文件数据达到500M时滚动一个文件。

问题来了，000012 有154个字节数据，那么需要从那个字节开始读呢？于是需要执行一个sql命令查看。

执行 SHOW MASTER STATUS; sql 命名

该命令记录当前主机写入的数据最新位置。

# position info
canal.instance.master.address=127.0.0.1:3306 # mysql服务器地址
canal.instance.master.journal.name=mysql-bin.000012 # SHOW MASTER STATUS; 查询的结果
canal.instance.master.position=154 # SHOW MASTER STATUS; 查询的结果
canal.instance.master.timestamp= #暂时不用配
canal.instance.master.gtid= #暂时不用配

配置canal用户(登录上主机，并拥有从机的同步权限)

通常默认的用户和密码就是canal

# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

我们不改他，在mysql中直接创建canal用户

mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';
ERROR 1819 (HY000): Your password does not satisfy the current policy requirements

若出现上面类型的错误；说明是密码的策略达不到mysql的要求。解决方案降低密码强度，修改策略

SHOW VARIABLES LIKE 'validate_password%'; 可以查看密码策略

降低密码强度

set global validate_password_policy=0;
set global validate_password_length=1;

重新执行

mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';

修改完成之后，记得还原

set global validate_password_length=8; 
set global validate_password_policy=MEDIUM;

为了是账号立即生效，需要执行FLUSH PRIVILEGES;命令

mysql> FLUSH PRIVILEGES;
Query OK, 0 rows affected (0.00 sec)

验证canal用户是否成功创建成功

mysql> select host,user from mysql.user ;
+-----------+-----------+
| host      | user      |
+-----------+-----------+
| %         | canal     |
| %         | root      |
| localhost | mysql.sys |
+-----------+-----------+
3 rows in set (0.00 sec)