当前位置： 首页 > news >正文

canal的入门与使用

news 来源：原创 2024/5/2 22:54:53

阿里巴巴 MySQL binlog 增量订阅&消费组件

github地址：https://github.com/alibaba/canal

1、介绍

在这里插入图片描述


canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费
早期阿里巴巴因为杭州和美国双机房部署，存在跨机房同步的业务需求，实现方式主要是基于业务 trigger 获取增量变更。从 2010 年开始，业务逐步尝试数据库日志解析获取增量变更进行同步，由此衍生出了大量的数据库增量订阅和消费业务。
基于日志增量订阅和消费的业务包括
数据库镜像
数据库实时备份
索引构建和实时维护(拆分异构索引、倒排索引等)
业务 cache 刷新
带业务逻辑的增量数据处理
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

2、工作原理

2.1、MySQL主备复制原理

在这里插入图片描述

  • MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
  • MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
  • MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

2.2、canal 工作原理

  • canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
  • MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
  • canal 解析 binary log 对象(原始为 byte 流)

3、QuickStart

3.1、准备

1、mysql配置准备

  • 对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
  • 注意：针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步
  • 授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant
CREATE USER canal IDENTIFIED BY 'canal';  
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;

查看mysql的log_bin是否开启：

SHOW VARIABLES LIKE 'log_bin';

如果Value 为 OFF 则未开启日志文件。
查询binlog 变动信息 :

SHOW BINLOG EVENTS;

windows版本文件位置：

D:\developSoftware\mysql-5.7.20\my.ini

2、canal准备

1、下载

D:\developsoftware2\canal\canal.deployer-1.1.4

2、修改配置文件

D:\developsoftware2\canal\canal.deployer-1.1.4\conf\example\instance.properties

#################################################
## 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=127.0.0.1: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=123456
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=.*\\..*
# table black regex
canal.instance.filter.black.regex=
# 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.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型，比如 UTF-8，GBK , ISO-8859-1
如果系统是1个 cpu，需要将 canal.instance.parser.parallel 设置为 false
3、启动

D:\developsoftware2\canal\canal.deployer-1.1.4\bin\startup.bat

去除启动参数：

-Dlogback.configurationFile="%logback_configurationFile%"

4、日志查看

D:\developsoftware2\canal\canal.deployer-1.1.4\logs

5、关闭

3.2、ClientSample

a、首先启动Canal Server，可参见QuickStart
b、

  1. 可以在idea 里，直接打开com.alibaba.otter.canal.example.SimpleCanalClientTest，直接运行
  2. 在工程的example目录下运行命令行：

mvn exec:java -Dexec.mainClass=“com.alibaba.otter.canal.example.SimpleCanalClientTest”

  1. 下载example包: https://github.com/alibaba/canal/releases，解压缩后，直接运行sh startup.sh脚本

c、 触发数据变更
更新数据库test2中lesson表的信息
d、 在控制台或者logs中查看，可以看到如下信息 ：

================> binlog[mysql-bin.000001:1158] , executeTime : 1610087664000(2021-01-08 14:34:24) , gtid : () , delay : 884ms
 BEGIN ----> Thread id: 3
----------------> binlog[mysql-bin.000001:1288] , name[test2,lesson] , eventType : UPDATE , executeTime : 1610087664000(2021-01-08 14:34:24) , gtid : () , delay : 885 ms
id : 1    type=int(11)
name : 大学物理    type=varchar(36)
credit : 5    type=int(11)
teacher : 张思清    type=varchar(36)    update=true
----------------
 END ----> transaction id: 1146
================> binlog[mysql-bin.000001:1388] , executeTime : 1610087664000(2021-01-08 14:34:24) , gtid : () , delay : 886ms

3.3 普通工程

1、导入依赖

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

2、测试代码

package com.zengqingfa.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

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

public class SimpleCanalClientExample {


    public static void main(String args[]) {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
                11111), "example", "canal", "123456");
        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("================&gt; 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("-------&gt; before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------&gt; 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、启动server
4、启动client
5、触发数据变更
6、观察控制台

================&gt; binlog[mysql-bin.000001:1935] , name[test2,lesson] , eventType : UPDATE
-------&gt; before
id : 2    update=false
name : 高等数学    update=false
credit : 5    update=false
teacher : 李佛1    update=false
-------&gt; after
id : 2    update=false
name : 高等数学    update=false
credit : 5    update=false
teacher : 李佛11    update=true

3.4 与springboot集成

1、pom.xml

<dependencies>
  <dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
  </dependency>
  <dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.0</version>
  </dependency>
  <dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jooq</artifactId>
  </dependency>
</dependencies>

2、application.yml

canal:
  server: localhost:11111
  destination: example
spring:
  application:
    name: canal-example
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://localhost:3306/test2
    username: canal
    password: 123456
  jooq:
    sql-dialect: mysql
  flyway:
    baseline-on-migrate: true

3、订阅数据库的增删改操作

a、实现EntryHandler 接口，泛型为想要订阅的数据库表的实体对象， 该接口的方法为 java 8 的 default 方法，方法可以不实现，如果只要监听增加操作，只实现增加方法即可

package com.zengqingfa.canal.handler;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

import java.io.Serializable;
@Slf4j
@Component
@CanalTable(value = "lesson")
public class UserHandler implements EntryHandler<UserHandler.Lesson> {

    @Override
    public void insert(Lesson lesson) {
        log.info("insert message  {}", lesson);
    }

    @Override
    public void update(Lesson before, Lesson after) {
        log.info("update before {} ", before);
        log.info("update after {}", after);
    }

    @Override
    public void delete(Lesson lesson) {
        log.info("delete  {}", lesson);
    }

    @Data
    public static class Lesson implements Serializable {

        private Integer id;
        private String name;
        private Integer credit;
        private String teacher;

    }

}

b、也支持统一的处理@CanalTable(value=“all”),这样除去存在EntryHandler的表以外，其他所有表的处理将通过该处理器,统一转为Map<String, String>对象

package com.zengqingfa.canal.handler;
import org.jooq.DSLContext;
import org.jooq.Field;
import org.jooq.Param;
import org.jooq.impl.DSL;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.context.CanalContext;
import top.javatool.canal.client.handler.EntryHandler;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static org.jooq.impl.DSL.field;
import static org.jooq.impl.DSL.table;

@CanalTable(value = "all")
@Component
public class DefaultEntryHandler implements EntryHandler<Map<String, String>> {

    @Resource
    private DSLContext dsl;

    private Logger logger = LoggerFactory.getLogger(DefaultEntryHandler.class);

    @Override
    public void insert(Map<String, String> map) {
        logger.info("增加 {}", map);
        String table = CanalContext.getModel().getTable();
        List<Field<Object>> fields = map.keySet().stream().map(DSL::field).collect(Collectors.toList());
        List<Param<String>> values = map.values().stream().map(DSL::value).collect(Collectors.toList());
        int execute = dsl.insertInto(table(table)).columns(fields).values(values).execute();
        logger.info("执行结果 {}", execute);
    }

    @Override
    public void update(Map<String, String> before, Map<String, String> after) {
        logger.info("修改 before {}", before);
        logger.info("修改 after {}", after);
        String table = CanalContext.getModel().getTable();
        Map<Field<Object>, String> map = after.entrySet().stream().filter(entry -> before.get(entry.getKey()) != null)
                .collect(Collectors.toMap(entry -> field(entry.getKey()), Map.Entry::getValue));
        dsl.update(table(table)).set(map).where(field("id").eq(after.get("id"))).execute();
    }

    @Override
    public void delete(Map<String, String> map) {
        logger.info("删除 {}", map);
        String table = CanalContext.getModel().getTable();
        dsl.delete(table(table)).where(field("id").eq(map.get("id"))).execute();
    }
}

如果你想获取除实体类信息外的其他信息，可以使用
CanalModel canal = CanalContext.getModel();

3.5 与消息队列集成

相关文章：

  • 病毒营销增长策略：Facebook和 Airbnb的5个病毒传播方法
  • C++：特殊类设计
  • Spring如何记忆
  • 材料科学基础学习指导-吕宇鹏-名词和术语解释-第二章晶体缺陷
  • 线段树2----简单拓展
  • 【学姐面试宝典】前端基础篇Ⅳ（JavaScript）
  • 本地JAR文件作为Gradle依赖项
  • Linux软件安装的4种方式
  • 自然语言处理Transformer模型最详细讲解（图解版）
  • JVM的组成
  • 【Rust日报】2022-10-15 Frui: 一个rust写的开发者友好的UI框架
  • Text Preprocessing - 文本预处理(RNN循环神经网络)
  • 【JavaScript设计模式】观察者模式
  • 【漏洞复现-splunk-信息泄露】vulfocus/splunk-cve_2018_11409
  • 赶紧进来看看---详解C/C++中的自定义类型:枚举和联合体
  • ➹使用webpack配置多页面应用(MPA)
  • download使用浅析
  • go append函数以及写入
  • Js基础知识（四） - js运行原理与机制
  • Koa2 之文件上传下载
  • Linux后台研发超实用命令总结
  • mongodb--安装和初步使用教程
  • Node项目之评分系统（二）- 数据库设计
  • rc-form之最单纯情况
  • Redis的resp协议
  • SAP云平台里Global Account和Sub Account的关系
  • Spring思维导图，让Spring不再难懂（mvc篇）
  • 服务器之间，相同帐号，实现免密钥登录
  • 基于axios的vue插件，让http请求更简单
  • 十年未变！安全，谁之责？（下）
  • 携程小程序初体验
  • 教程：使用iPhone相机和openCV来完成3D重建（第一部分） ...
  • 如何在 Intellij IDEA 更高效地将应用部署到容器服务 Kubernetes ...
  • ​直流电和交流电有什么区别为什么这个时候又要变成直流电呢？交流转换到直流（整流器）直流变交流（逆变器）​
  • # 学号 2017-2018-20172309 《程序设计与数据结构》实验三报告
  • # 再次尝试 连接失败_无线WiFi无法连接到网络怎么办【解决方法】
  • (01)ORB-SLAM2源码无死角解析-(56) 闭环线程→计算Sim3:理论推导(1)求解s,t
  • (1) caustics\
  • （145）光线追踪距离场柔和阴影
  • (C#)Windows Shell 外壳编程系列9 - QueryInfo 扩展提示
  • (Matlab)基于蝙蝠算法实现电力系统经济调度
  • （翻译）Entity Framework技巧系列之七 - Tip 26 – 28
  • （附源码）spring boot校园健康监测管理系统 毕业设计 151047
  • （附源码）springboot 基于HTML5的个人网页的网站设计与实现 毕业设计 031623
  • （附源码）ssm高校实验室 毕业设计 800008
  • （六）软件测试分工
  • （一）Thymeleaf用法——Thymeleaf简介
  • ./indexer: error while loading shared libraries: libmysqlclient.so.18: cannot open shared object fil
  • .bat批处理（九）：替换带有等号=的字符串的子串
  • .net core 客户端缓存、服务器端响应缓存、服务器内存缓存
  • .Net小白的大学四年，内含面经
  • .NET中使用Redis (二)
  • .Net转前端开发-启航篇，如何定制博客园主题
  • /etc/shadow字段详解
  • /etc/sudoer文件配置简析