DOCKER 部署 CANAL

DOCKER 部署 CANAL

1、MYSQL开启binlog

前提MYSQL已经安装完成，canal采用读取Mysql的binlog日志来实现数据同步，需要修改mysql配置为难my.cnf,并将binlog的格式模式设置为ROW，其中server-id必须与后边canal配置文件中的server-id不相同。

# binlog
log-bin=mysql-bin
binlog_format=ROW
server-id=1

配置修改完后，重新启动mysql，并binlog是否启动成功

# 重启mysql
systemctl mysqld restart
# 查看binlog是否启动成功【结果显示为:（log_bin ON）表示启动成功】
SHOW VARIABLES LIKE 'log_bin'

2、mysql授权canal用户

#创建canal用户
CREATE USER 'canal' IDENTIFIED BY 'canal'
# 授权canal用户 查询、从复制、客户端复制权限
GRANT SELECT,REPLICATION SLAVE ,REPLICATION CLIENT ON *.* to 'canal'@'%' IDENTIFIED BY "canal";
# 刷新权限
flush PRIVILEGES;

如果出现【1819 - Your password does not satisfy the current policy requirements, Time: 0.043000s】异常，表明密码级别过高，可以适当降低密码级别

# 查看密码校验信息
SHOW VARIABLES LIKE 'validate_password%';
# 设置为校验策略为低级别
SET GLOBAL validate_password_policy=LOW;
# 设置密码校验长度为5
SET GLOBAL validate_password_length=5;
# 再次授权
GRANT SELECT,REPLICATION SLAVE ,REPLICATION CLIENT ON *.* to 'canal'@'%' IDENTIFIED BY "canal";
# 刷新权限
FLUSH PRIVILEGES;

3、创建并启动canal容器

创建存放canal配置文件目录【目录可自行指定】

mkdir -p /opt/canal/conf

创建canal配置文件

vim /opt/canal/conf/instance.properties

canal配置内容如下

#################################################
## 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=120.48.130.125: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
# enable druid Decrypt database password
canal.instance.enableDruid=false
#canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==

# table regex
# canal.instance.filter.regex=.*\\..*
canal.instance.filter.regex=.*\\..*
# 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=example
# dynamic topic route by schema or table regex
#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..*
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.properties】配置文件解析

canal.instance.master.address 需要监听的mysql主机ip:端口
canal.instance.dbUsername canal订阅binlog使用的用户
canal.instance.dbPassword canal订阅binlog使用的用户密码
canal.instance.connectionCharset = UTF-8 canal订阅的字符集
canal.instance.filter.regex 需要监听的mysql库和表
mysql库和表表达式：【备注：配置全库或库下所有表有可能出现 （errorNumber=1146, fieldCount=-1, message=Table 'home.BASE TABLE' doesn't exist, sqlState=42S02, sqlStateMarker=#)异常，尽量配置到指定库下的指定表】
1 .*\\..* : 全库
2 canal\\..* : 指定库下的所有表
3 canal\\.canal,test\\.test : 指定库下的指定表

启动docker容器

docker run -d \
-v /opt/canal/conf/instance.properties:/home/admin/canal-server/conf/example/instance.properties \
-p 11111:11111 \
--name canal \
canal/canal-server:v1.1.6

查看是否启动成功

docker logs -f canal

4、客户端监听canal

创建需要监听的表

CREATE TABLE `sys_user_role` (
`user_id` bigint(20) NOT NULL COMMENT '用户ID',
`role_id` bigint(20) NOT NULL COMMENT '角色ID',
PRIMARY KEY (`user_id`,`role_id`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='用户和角色关联表'

4.1 代码实现监听所有表变动

引入依赖

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

代码实现【摘自他人】

package com.pango.system.service.impl;

import java.net.InetSocketAddress;
import java.util.List;

import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;

/**
* @date 2022/10/9
*/
public class SimpleCanalClientExample {

public static void main(String args[]) {

// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("120.48.130.125",
11111), "example", "", "");
int batchSize = 1000;
int emptyCount = 0;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
int totalEmptyCount = 120;
while (emptyCount < totalEmptyCount) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId();
int size = message.getEntries().size();
if (batchId == -1 || size == 0) {
emptyCount++;
System.out.println("empty count : " + emptyCount);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
}
} else {
emptyCount = 0;
// System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
printEntry(message.getEntries());
}
connector.ack(batchId); // 提交确认
// connector.rollback(batchId); // 处理失败, 回滚数据
}
System.out.println("empty too many times, exit");
} finally {
connector.disconnect();
}

}

private static void printEntry(List<Entry> entrys) {
for (Entry entry : entrys) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue;
}
RowChange rowChage = null;
try {
rowChage = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
EventType eventType = rowChage.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
for (RowData rowData : rowChage.getRowDatasList()) {
if (eventType == EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
printColumn(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
printColumn(rowData.getAfterColumnsList());
}
}
}
}

private static void printColumn(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}

验证

/** 插入数据 */
INSERT INTO sys_user_role VALUES (3,3)
/** 测试结果： ================> binlog[mysql-bin.000001:167986] , name[pango,sys_user_role] , eventType : INSERT
user_id : 3 update=true
role_id : 3 update=true
**/
/** 更新数据 */
UPDATE sys_user_role SET role_id = 4 WHERE user_id = 3
/** 测试结果：================> binlog[mysql-bin.000001:168265] , name[pango,sys_user_role] , eventType : UPDATE
-------> before
user_id : 3 update=false
role_id : 3 update=false
-------> after
user_id : 3 update=false
role_id : 4 update=true
**/
/** 删除数据 */
DELETE FROM sys_user_role WHERE user_id = 3
/** 测试结果：================> binlog[mysql-bin.000001:168562] , name[pango,sys_user_role] , eventType : DELETE
user_id : 3 update=false
role_id : 4 update=false
**/

4.2 代码实现监听单表变动

引入依赖

<dependency>
<groupId>top.javatool</groupId>
<artifactId>canal-spring-boot-starter</artifactId>
<version>1.2.1-RELEASE</version>
</dependency>

编写配置【可以查看canal.properties,该文件有canal的集群名字等配置描述】

canal:
destination: example # canal的集群名字，要与安装canal时设置的名称一致
server: 120.48.130.125:11111 # canal服务地址

代码实现

• 表数据接收类，该类在类定义上一定要标注@Table(name = "sys_user_role")注解，且属性上一定要标注 @Column(name = "user_id")注解，否则无法接收到表数据

package com.pango.system.domain;

import com.baomidou.mybatisplus.annotation.TableName;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;

import javax.persistence.Column;
import javax.persistence.Table;
import java.io.Serializable;

/**
* 用户和角色关联 sys_user_role
*
*/
@Table(name = "sys_user_role")
public class SysUserRole implements Serializable
{
/** 用户ID */
@Column(name = "user_id")
private Long userId;

/** 角色ID */
@Column(name = "role_id")
private Long roleId;

public Long getUserId()
{
return userId;
}

public void setUserId(Long userId)
{
this.userId = userId;
}

public Long getRoleId()
{
return roleId;
}

public void setRoleId(Long roleId)
{
this.roleId = roleId;
}

@Override
public String toString() {
return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE)
.append("userId", getUserId())
.append("roleId", getRoleId())
.toString();
}
}

• 监听处理类

package com.pango.system.handler;

import com.pango.system.domain.SysUserRole;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

/**
* @date 2022/10/9
*/
@CanalTable(value = "sys_user_role")
@Component
public class UserRoleHandler implements EntryHandler<SysUserRole> {

/**
* 插入表sys_user_role数据时，触发该方法
* 如：INSERT INTO sys_user_role VALUES (3,3)
* @param sysUserRole
*/
@Override
public void insert(SysUserRole sysUserRole) {
System.out.println(sysUserRole);
}

/**
* 更新表sys_user_role数据时，触发该方法
* 如：UPDATE sys_user_role SET role_id = 4 WHERE user_id = 3
* @param before
* @param after
*/
@Override
public void update(SysUserRole before, SysUserRole after) {
System.out.println("before:"+before);
System.out.println("after:"+after);
}

/**
* 删除表sys_user_role数据时，触发该方法
* 如：DELETE FROM sys_user_role WHERE user_id = 3
* @param sysUserRole
*/
@Override
public void delete(SysUserRole sysUserRole) {
System.out.println(sysUserRole);
}
}