网上搜索 canal 的原理比比皆是，这里就不介绍了，记录下自己在使用 canal 的搭建过程。
我是基于 docker 搭建的的 canal-server
首先查看 MySQL 是否开启了binlog模式。（我是基于 docker 创建的 MySQL 容器）
第一步、进入 MySQL 容器
[root@localhost ~] docker exec -it my_mysql /bin/bash
第二步、进入 MySQL
[root@943f5d4d225f:/ ] mysql -uroot -p
第三步、查看是否开启binlog模式
mysql> SHOW VARIABLES LIKE '%log_bin%';
+---------------------------------+-----------------------------+
| Variable_name | Value |
+---------------------------------+-----------------------------+
| log_bin | ON |
| log_bin_basename | /var/lib/mysql/binlog |
| log_bin_index | /var/lib/mysql/binlog.index |
| log_bin_trust_function_creators | OFF |
| log_bin_use_v1_row_events | OFF |
| sql_log_bin | ON |
+---------------------------------+-----------------------------+
此时显示的 log_bin 是 ON 则是开启了，如果不是则需要开启，下面是开启 MySQL binlog模式的步骤
第一步、修改/etc/my.cnf 需要开启binlog模式。
log-bin=mysql-bin
binlog-format=ROW#设置行模式，默认是 mixed
server_id=1 #此处是 server_id不能与 canal 的 id 相同
第二步、进入mysql（同上）
第三步、创建账号 ，使用root账号创建用户并授予权限
mysql> create user canal@'%' IDENTIFIED by 'canal';
mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';
mysql> FLUSH PRIVILEGES;
第四步、重启 mysql 容器
[root@localhost ~] docker restart my_mysql
其次进行 canal 的配置
第一步、从 docker 仓库拉去镜像
[root@localhost ~] docker pull canal/canal-server
第二步、通过镜像启动 canal 容器
[root@localhost ~] docker run --name my_canal -p 11111:11111 -d canal/canal-server
第三步、进行 canal 的配置，进入容器
[root@localhost ~] docker exec -it my_canal /bin/bashi
第四步、配置canal.properties文件（如下图）
[root@1469b8913eb9 admin] vi /home/admin/canal-server/conf/canal.properties
image.png
第五步、配置instance.properties文件
[root@1469b8913eb9 admin]#vi /home/admin/canal-server/conf/example/instance.properties
image.png
第六步、设置开机启动、并重启
[root@localhost ~] docker update --restart=always my_canal
[root@localhost ~] docker restart my_canal
上面进行完成后进行测试
创建测试项目canal_test，pom 引入
<dependency>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.client</artifactId>
<version>1.1.0</version>
</dependency>
具体方法,
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 java.net.InetSocketAddress;
import java.util.List;
public class sample {
public static void main(String args[]) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("172.16.153.2",
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<CanalEntry.Entry> entrys) {
for (CanalEntry.Entry entry : entrys) {
if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
continue;
}
CanalEntry.RowChange rowChage = null;
try {
rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
CanalEntry.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 (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
if (eventType == CanalEntry.EventType.DELETE) {
printColumn(rowData.getBeforeColumnsList());
} else if (eventType == CanalEntry.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<CanalEntry.Column> columns) {
for (CanalEntry.Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
}
启动后控制台输出,代表监听成功
empty count : 1
empty count : 2
empty count : 3
empty count : 4
如果监听失败，查看 canal 日志
进入容器
[root@localhost ~] docker exec -it mall_canal /bin/bash
查看日志文件
[root@1469b8913eb9 admin] tail -n 100 /home/admin/canal-server/logs/example/example.log
可能会出现连接到数据库的情况，对应的问题对应解决。
