java整合Canal实现数据库监听(附完整的踩坑总结)
1. 预备工作1.1. MySQL开启binlog日志
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复 修改完成后输入: show variables like 'log_bin';查看binlog是否生效
show variables like 'log_bin';
https://i-blog.csdnimg.cn/blog_migrate/2ec6406165175c3bba44719864f59fc1.png
1.2. 授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限
-- 创建一个用户名密码均为canal的账户
CREATE USER canal IDENTIFIED BY 'canal';
-- 授予用户'canal'在所有数据库和表上进行SELECT查询以及复制从库和复制客户端的权限，并且允许该用户从任何主机连接到数据库服务器。
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- 刷新权限
FLUSH PRIVILEGES;
2. 正式开始
2.1. 采用doker部署canal-server
2.1.1. 使用docker拉取最新版canal服务端
docker pull canal/canal-server:latest 2.1.2. 使用docker run命令创建容器并挂载设置文件所在的数据卷
# -v 本地的instance.properties:容器的instance.properties将容器的instance.properties配置文件挂载到宿主机，方便后续变更
docker run \
--name mycanal \
-p 11111:11111 \
-v /tmp/canal/conf2/instance.properties:/home/admin/canal-server/conf/example/instance.properties \
-d \
canal/canal-server 2.2. 修改设置文件
进入数据卷挂载的目录新建一个instance.properties文件 修改设置如下
## mysql serverId 不能和mysql相同
canal.instance.mysql.slaveId = 1234
#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.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#username/password，需要改成自己的数据库信息 此处采用刚刚创建的用户
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
#需要改成同步的数据库表规则
canal.instance.filter.regex = .\*\\\\..\* 设置参考 此处引用csdn博客
mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\)
常见例子：
[*]所有表：. or .\..*
[*]canal schema下所有表： canal\..*
[*]canal下的以canal打头的表：canal\.canal.*
[*]canal schema下的一张表：canal.test1
[*]多个规则组合使用：canal\..*,mysql.test1,mysql.test2 (逗号分隔)
注意：此过滤条件只针对row模式的数据有用(ps. mixed/statement由于不解析sql，以是无法准确提取tableName举行过滤)
修改完成后重启 canal容器即可。
docker restartmycanal; 至此,canal服务端就算设置完成
3. java示例
3.1. 导入相关maven依靠
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.0</version>
</dependency> 3.2. 创建官方示例代码
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.common.utils.AddressUtils;
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;
public class SimpleCanalClientExample {
public static void main(String args[]) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
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 \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());
}
}
} 3.3. 运行Client
首先启动Canal Server
启动Canal Client后，可以从控制台从看到雷同消息：
empty count : 1
empty count : 2
empty count : 3
empty count : 4 此时代表当前数据库无变更数据
改改数据库,如修改或添加数据时,会出现如下提示:
================> binlog , name , eventType : UPDATE
-------> before
123 : 123 update=false
-------> after
123 : 1233 update=true 出现如上消息代表数据库被改变 至此,恭喜你成功设置.
4. 踩坑总结
4.1. java客户端无法连接到canal的服务端
如下图所示:
https://i-blog.csdnimg.cn/blog_migrate/2d590aeed59e956a76faac36549cf9e6.png
需要检查下地址的设置,确认能ping通
https://i-blog.csdnimg.cn/blog_migrate/5d1e13e5c55392a214d583ff5b675910.png
如果可以ping通但是仍然无法连接,查看是否是捏造机的防火墙是否打开,11111端口可能被拦截。
4.2. java客户端连接到了canal但是修改数据库无日志打印
4.2.1. 缘故原由1: canal无法连接到MySQL
[*]首先检查数据库中是否有设置文件设置的账户和暗码,如本文中的 账号:canal 暗码:canal
[*]如果是MySQL 8.0以上,需要修改身份验证插件
## 自MySQL 8.0.3开始，身份验证插件默认使用caching_sha2_password
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'password';
[*]如果是在docker中部署的MySQL,需要注意此时的设置文件 canal.instance.master.address = 127.0.0.1:3306 的地址需要是docker的地址,通过 ifconfig查看
https://i-blog.csdnimg.cn/blog_migrate/1534a355101244b260c3e6805246cc25.png
4.2.2. 缘故原由2: MySQL的binlog日志未开启,此处参考第一步开启
4.2.3. 缘故原由3: MySQL的binlog日志的binlog-format非ROW模式,修改即可
4.3. 其他缘故原由,手动查看canal的日志,排查题目
输入 docker ps 查看正在运行的容器
# docker ps
CONTAINER ID IMAGE COMMAND CREATED STATUS PORTS NAMES
b4b3328b1a08 canal/canal-server "/alidata/bin/main.s…" 55 minutes ago Up 50 minutes 9100/tcp, 11110/tcp, 11112/tcp, 0.0.0.0:11111->11111/tcp, :::11111->11111/tcp mycanal2
c34cdcfea74e mysql "docker-entrypoint.s…" 2 hours ago Up 2 hours 0.0.0.0:3306->3306/tcp, :::3306->3306/tcp, 33060/tcp mymysql
460bf1f5daaf canal/canal-server "/alidata/bin/main.s…" 4 hours ago Up 4 hours 9100/tcp, 11110-11112/tcp zealous_hermann 找到canal的容器id 输入 docker exec -it [容器id] /bin/bash 进入容器内部
进入 /canal-server/logs/example目录输入 cat example.log 查看日志根据具体内容排查
比方:
https://i-blog.csdnimg.cn/blog_migrate/5cdde6aa497cc6cc7c1ecfb118d9494d.png
参考文档:
java整合canal 实现数据同步_java 融合canal-client-CSDN博客
canal官方wiki
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]