canal实现mysql数据同步

目录

1、canal下载

2、mysql同步用户创建和授权

3、canal admin安装和启动

4、canal server安装和启动

5、java 端集成监听canal 同步的mysql数据

6、java tcp同步只是其中一种方式，还可以通过kafka、rabbitmq等方式进行数据同步

---

1、canal下载

canal实现mysql数据同步可以直接安装canal server就可以了，但是为了方便管理(instance配置，canal server状态管理，集群等)，需要安装canal admin，应用下载地址：Releases · alibaba/canal · GitHub

进入页面可以选择需要安装的版本

下载canal.deployer-1.1.8.tar.gz和canal.admin-1.1.8.tar.gz

2、mysql同步用户创建和授权

登录mysql
mysql -h 127.0.0.1 -P 3306 -u root -p创建同步用户 repl 密码设为123456
CREATE USER 'repl'@'%' IDENTIFIED BY '123456';给予同步权限
GRANT REPLICATION SLAVE ON *.* to 'repl'@'%' identified by '123456';给予repl只读test库的权限，test库是用来同步数据的
GRANT SELECT ON test.* to 'repl'@'%' identified by '123456';canal_manager是canal admin需要的，给予repl对该库的读写权限
GRANT ALL PRIVILEGES ON canal_manager.* to 'repl'@'%' identified by '123456';mysql my.cnf配置文件增加主从配置master数据库的配置信息
#主数据主从配置 唯一id
server_id=1
#开启logbin
log-bin=mysql-bin
#写入模式 row
binlog-format=ROW
#需要同步的库
binlog-do-db=test
#忽略的数据库
replicate-ignore-db=mysql
replicate-ignore-db=sys
replicate-ignore-db=information_schema
replicate-ignore-db=performance_schema

在canal-admin解压文件的conf中有一个canal_manager.sql，导入到master数据库

3、canal admin安装和启动

把canal.admin-1.1.8.tar.gz上传到linux

解压 tar -zvxf canal.admin-1.1.8.tar.gz 

进入conf目录下，编辑application.yml配置文件。

server:port: 8089
spring:jackson:date-format: yyyy-MM-dd HH:mm:sstime-zone: GMT+8spring.datasource:address: 127.0.0.1:3306database: canal_managerusername: replpassword: 123456driver-class-name: com.mysql.jdbc.Driverurl: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=truehikari:maximum-pool-size: 30minimum-idle: 1canal:adminUser: adminadminPasswd: 123456

重点介绍以下几个参数：

address：我们需要订阅（也就是mysql master服务器）mysql所在的服务器IP和数据库端口。

database：canal.admin web系统必须的几张表，需要在mysql master服务器上初始化conf/canal_manager.sql文件。

sername和password就是mysql master服务器创建的用于复制的用户和密码，也就是我们在canal server中配置的repl 和 123456。

driver-class-name：mysql的驱动，默认是MYSQL5的驱动，如果你的MYSQL是8的（我的就是），要将驱动改为com.mysql.cj.jdbc.Driver。

另外，还需要在mysql连接后面加上allowPublicKeyRetrieval=true，不然启动时，有可能报错。

启动canal.admin

进入bin目录，执行如下命令，启动canal.admin：

./startup.sh

查看 admin 日志

2022-12-10 03:13:58.995 [main] INFO  o.s.jmx.export.annotation.AnnotationMBeanExporter - 
Located MBean 'dataSource': registering with JMX server as MBean [com.zaxxer.hikari:name=dataSource,type=HikariDataSource]
2022-12-10 03:13:59.015 [main] INFO  org.apache.coyote.http11.Http11NioProtocol - Starting ProtocolHandler ["http-nio-8089"]
2022-12-10 03:13:59.038 [main] INFO  org.apache.tomcat.util.net.NioSelectorPool - Using a shared selector for servlet write/read
2022-12-10 03:13:59.214 [main] INFO  o.s.boot.web.embedded.tomcat.TomcatWebServer - Tomcat started on port(s): 8089 (http) with context path ''
2022-12-10 03:13:59.221 [main] INFO  com.alibaba.otter.canal.admin.CanalAdminApplication - Started CanalAdminApplication in 14.281 seconds (JVM running for 15.894)

如果出现上述日志，说明启动成功！

登录admin

通过http://127.0.0.1:8089/访问，默认密码：admin/123456。

注意，IP和密码需要改成你自己配置的。如果是在服务器上配置的，别忘记放开8089端口。

输入用户名和密码之后，出现上述页面说明配置成功！

如果需要修改密码，直接通过执行 select upper(sha1(unhex(sha1('1234567')))) 这个sql得到结果，然后复制到canal_manager库的canal_user表的password字段中就可以了，其中1234567是明文密码，执行上述sql会得到一个密码。

4、canal server安装和启动

把canal.deployer-1.1.8.tar.gz上传到linux

解压 tar -zvxf ccanal.deployer-1.1.8.tar.gz

进入conf目录下，编辑canal.properties配置文件。

注意，如果直接编辑canal.properties，可能无法启动，报如下错误：

可以通过如下方式修改

mv canal.properties canal.properties_bak
cp canal_local.properties canal.properties
vim canal.properties

# register ip
canal.register.ip =# canal admin config  canalAdmin 的链接、端口、用户名和MD5密码
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd =6F32482BAFC60F23B7736044CEFC1799166E5CDB
# admin auto register canal server启动后自动注入到canal admin管理模块
canal.admin.register.auto = true
canal.admin.register.cluster =
canal.admin.register.name =

一般只需要修改下面这3个

canal.admin.manager = 127.0.0.1:8089
canal.admin.user = admin
canal.admin.passwd =6F32482BAFC60F23B7736044CEFC1799166E5CDB

启动canal.server

进入bin目录，执行如下命令，启动canal.server：

./startup.sh

查看canal日志

启动后，canalAdmin的server管理模块，对应创建的canal server会动态识别到，状态变为启动

5、java 端集成监听canal 同步的mysql数据

1、引入依赖

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.4</version>
</dependency>

2、编写测试代码

package com.hy.das.config;import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;import java.net.InetSocketAddress;
import java.util.List;@Component
public class CanalClient implements InitializingBean{private final static int BATCH_SIZE = 1000;@Overridepublic void afterPropertiesSet() throws Exception {// 创建链接 此处的11111为tcp端口 在canal admin Server管理模块可以查看CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111),"test", "", "");try {//打开连接connector.connect();//订阅数据库表,全部表connector.subscribe(".*\\..*");//回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿connector.rollback();while (true) {// 获取指定数量的数据Message message = connector.getWithoutAck(BATCH_SIZE);System.out.println(message.getEntries().size());//获取批量IDlong batchId = message.getId();//获取批量的数量int size = message.getEntries().size();//如果没有数据if (batchId == -1 || size == 0) {try {//线程休眠2秒Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}} else {System.out.println("----------------");//如果有数据,处理数据//遍历entries，单条解析for (CanalEntry.Entry entry : message.getEntries()) {//获取表名String tableName = entry.getHeader().getTableName();//获取类型CanalEntry.EntryType entryType = entry.getEntryType();//获取序列化后的数据ByteString storeValue = entry.getStoreValue();//判断entry类型是否为ROWDATA类型if (CanalEntry.EntryType.ROWDATA.equals(entryType)){//反序列化CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);//获取当前事件操作类型CanalEntry.EventType eventType = rowChange.getEventType();//获取数据集List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();//遍历for (CanalEntry.RowData rowData : rowDatasList) {//改变前数据JSONObject jsonObjectBefore = new JSONObject();List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();for (CanalEntry.Column column : beforeColumnsList) {jsonObjectBefore.put(column.getName(),column.getValue());}//改变后数据JSONObject jsonObjectAfter = new JSONObject();List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();for (CanalEntry.Column column : afterColumnsList) {jsonObjectAfter.put(column.getName(),column.getValue());}System.out.println("Table:"+tableName+",EventTpye:"+eventType+",Before:"+jsonObjectBefore+",After:"+jsonObjectAfter);}}else {System.out.println("当前操作类型为："+entryType);}}}//进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。connector.ack(batchId);}} catch (Exception e) {e.printStackTrace();} finally {connector.disconnect();}}
}

newSingleConnector方法里面的test是一个instance实列，定义了需要同步的master库的信息(ip、端口、用户名、密码、binlog文件名称、同步位置、需要同步的库、不需要同步的库等)

在canal admin web管理界面的Instance 管理模块，点击新建Instance进行创建，新建页面的Instance名称就是test，这个可以随便填写，代码对应修改就行，所属集群/主机，因为我这里是单机部署，直接选择自动注入的canal server就行，点击载入模板，获取配置初始信息，下图中标出的信息按照实际的修改填入就行，点击保存后，启动这个Instance。

3、启动服务，对test库的sys_user表进行数据更新，可以看到后台已经收到变更数据

6、java tcp同步只是其中一种方式，还可以通过kafka、rabbitmq等方式进行数据同步

注意上面需要提供对外访问的端口需要开通安全组，比如8089、11111等端口。

参考文章:

【CanalAdmin部署文档】_canal-admin-CSDN博客

https://zhuanlan.zhihu.com/p/590705531