目录
前言
简单原理
1.mysql数据库开启Binlog模式
1.docker 安装 canal 服务端
3.实现canal客户端
前言
是由公司业务改造搜索功能，使用ES搜索引擎中间件，那么我们需要将mysql中的数据同步至ES服务中，最总选择使用alibaba的canal增量订阅和解析工具。
简单原理
- canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
- mysql master收到dump请求，开始推送binary log给slave(也就是canal)
- canal解析binary log对象(原始为byte流)
1.mysql数据库开启Binlog模式
注：如未安装mysql可参考《docker环境中安装mysql》进行安装。
针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限，无需任何操作，可直接进行下一步
docker进入已安装的mysql容器，找到my.cnf文件编辑
docker exec -it 容器id bash
ls
cd etc
ls
编辑my.cnf文件
vim my.cnf文件内容追加以下配置并保存，最后重启mysql容器
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复知识点补充：
mysql binlog的格式有三种：STATEMENT、ROW、MIXED
- ROW 模式（一般就用它）
日志会记录每一行数据被修改的形式，不会记录执行 SQL 语句的上下文相关信息，只记录要修改的数据，哪条数据被修改了，修改成了什么样子，只有 value，不会有 SQL 多表关联的情况。
优点：它仅仅只需要记录哪条数据被修改了，修改成什么样子了，所以它的日志内容会非常清楚地记录下每一行数据修改的细节，非常容易理解。
缺点：ROW 模式下，特别是数据添加的情况下，所有执行的语句都会记录到日志中，都将以每行记录的修改来记录，这样会产生大量的日志内容。- STATEMENT 模式
每条会修改数据的 SQL 语句都会被记录下来。
缺点：由于它是记录的执行语句，所以，为了让这些语句在 slave 端也能正确执行，那他还必须记录每条语句在执行过程中的一些相关信息，也就是上下文信息，以保证所有语句在 slave 端被执行的时候能够得到和在 master 端执行时候相同的结果。
但目前例如 step()函数在有些版本中就不能被正确复制，在存储过程中使用了 last-insert-id()函数，可能会使 slave 和 master 上得到不一致的 id，就是会出现数据不一致的情况，ROW 模式下就没有。- MIXED 模式
以上两种模式都使用。引用至：Canal：同步mysql增量数据工具，一篇详解核心知识点 - 知乎
新建一个mysql账号用于canal连接（不建议使用root账号），并具有作为 MySQL slave 的权限
# 新增账号
CREATE USER canal IDENTIFIED BY 'canal';
#设置账号操作权限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
#刷新权限配置
FLUSH PRIVILEGES;1.docker 安装 canal 服务端
拉取最新版本canal-server镜像
docker pull canal/canal-server:latest
构建canal-server容器
docker run --name canal -p 11111:11111 -d canal/canal-server:latest查看安装情况，并进入容器
docker ps
docker exec -it 容器id bash
找到文件instance.properties
编辑instance.properties
vim instance.properties具体修改配置
## mysql serverId
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
#table regex
canal.instance.filter.regex = .\*\\\\..\*主要关注属性：
canal.instance.master.address = 数据库ip地址:3306
canal.instance.dbUsername = 数据库账号
canal.instance.dbPassword = 数据库密码
canal.instance.filter.regex = .\*\\\\..\* 关注的表（默认全部表数据）配置的过滤正则，更多过滤格式可参考：Canal配置connector.subscribe和canal.instance.filter.regex遇到的坑_Alice_qixin的博客-CSDN博客_canalconnector
最后保存配置文件退出，并重启canal-server容器
3.实现canal客户端
核心代码实现
public static void main(String[] args) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("服务器ip地址",
11111), "example", "", "");
int batchSize = 1000;
int emptyCount = 0;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
int totalEmptyCount = 1000;
while (emptyCount < totalEmptyCount) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
System.out.println("message:" + message);
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);
}
System.out.println("entry======:" + entry);
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());
}
}运行main方法，并查看控制台日志
在这期间，操作关注的数据库表数据
查看控制台日志，显示为update操作，获取binlog成功
注：在实际生产业务中可把canal逻辑放在定时任务中，定时获取mysql binlog 数据同步
