首页 > 数据库 >Canal 安装 && 使用,监听Mysql数据库

Canal 安装 && 使用,监听Mysql数据库

时间:2023-05-27 19:00:43浏览次数:47  
标签:Canal canal binlog CanalEntry Mysql instance && mysql entry

1.下载canal.deployer-1.1.5.tar.gz

2.查看数据库是否开启binlog模式(必须)

show variables like 'log_%';
log_bin属性值为ON,则binlog模式开启;为OFF则binlog模式关闭。
若binlog模式关闭,则在MySQL配置文件my.cnf(my.ini)设置如下信息:
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复

3.查看binlog日志文件列表(1.1.5不需要配置)

show binary logs;

4.查看当前正在写入的binlog文件(1.1.5不需要配置)

show master status;

5.配置mysql

若binlog模式关闭,则在MySQL配置文件my.cnf(my.ini)设置如下信息:
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复
---------------------------------以上必须--------------------------------------
#binlog数据中包含server_id,标识该数据是由那个server同步过来的
server-id = 1
#mysql向文件名前缀添加数字后缀来按顺序创建二进制日志文件如mysql-binlog.000006 
log-bin = /var/lib/mysql/mysql-bin
#选择基于行的日志记录方式
binlog_format = ROW

6.配置canal

(1.1.5不配置直接启动也行)

conf/example/instance.properties

# mysql serverId,随便设置  (可以不配)
canal.instance.mysql.slaveId = 1234
#position info,需要改成自己的数据库信息
canal.instance.master.address=127.0.0.1:3306
#一定要注释掉下面这个参数,这样就会扫描全库
#canal.instance.defaultDatabaseName = test
# username/password 数据库用户名密码
canal.instance.dbUsername=root
canal.instance.dbPassword=root

7.启动canal

bin/startup.sh
#查看
canal.deployer-1.1.5\logs\canal\canal.log 是否成功
#如果失败 尝试修改  conf/canal.properties下 不再注释这句
canal.instance.parser.parallelThreadSize = 16

8.监听多个数据库

复制canal.deployer-1.1.5\conf\example重命名(对接的java代码中的example需要替换)
配置instance.properties
canal.instance.master.address=152.136.150.136:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=123456

9.Pom.xml

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

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.47</version>
        </dependency>

10.示例:CanalClient

package com.example.canaldemo.canal;

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 org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

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

@Component
public class CanalClient implements InitializingBean {

    private final static int BATCH_SIZE = 1000;
 
    @Override
    public void afterPropertiesSet() throws Exception {
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
        try {
            //打开连接
            connector.connect();
            //订阅数据库表,全部表
            connector.subscribe(".*..*");
            //回滚到未进行ack的地方,下次fetch的时候,可以从最后一个没有ack的地方开始拿
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(BATCH_SIZE);
                //获取批量ID
                long batchId = message.getId();
                //获取批量的数量
                int size = message.getEntries().size();
                //如果没有数据
                if (batchId == -1 || size == 0) {
                    try {
                        //线程休眠2秒
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    //如果有数据,处理数据
                    printEntry(message.getEntries());
                }
                //进行 batch id 的确认。确认之后,小于等于此 batchId 的 Message 都会被确认。
                connector.ack(batchId);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            connector.disconnect();
        }
    }
 
    /**
     * 打印canal server解析binlog获得的实体类信息
     */
    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;
            }
            //RowChange对象,包含了一行数据变化的所有特征
            //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
            CanalEntry.RowChange rowChage;
            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);
            }
            //获取操作类型:insert/update/delete类型
            CanalEntry.EventType eventType = rowChage.getEventType();
            //打印Header信息
            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));
            //判断是否是DDL语句
            if (rowChage.getIsDdl()) {
                System.out.println("================》;isDdl: true,sql:" + rowChage.getSql());
            }
            //获取RowChange对象里的每一行数据,打印出来
            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());
        }
    }
}
spring:
  datasource:
    driver-class-name: com.mysql.jdbc.Driver
    url: jdbc:mysql://127.0.0.1:3306/db1?useSSL=false&useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull
    username: root
    password: root
  main:
    allow-bean-definition-overriding: true
server:
  port: 9011




标签:Canal,canal,binlog,CanalEntry,Mysql,instance,&&,mysql,entry
From: https://blog.51cto.com/u_16085354/6362920

相关文章

  • MySQL增量备份的使用
    登录mysql数据库创建数据库HB3051,数据库中创建student表设置结构表中插入数据备份HB3051数据库中的student表备份HB3051数据库备份mysql数据库中的user表和server表备份所有数据库删除HB3051数据库恢复备份误删除的HB3051数据库模拟HB3051数据库中的student表备份删除查看表是否删......
  • MySQL的sql语句执行流程(简述)
    导言:MySQL和服务器端对接的时候,我们知道一般就是服务器端会打包一些SQL命令去增删改查数据库,这个打包的数据库SQL语句数据包一般为4MB,再大一些就不会被数据库端接收了但是我们可以自己更改默认大小,当数据包到达数据库端以后,它们会经历那些操作然后返回结果给服务器端呢?大概有:查......
  • 在 Kubernetes 上部署 RadonDB MySQL 集群
    1.mysql部署部署参考文档:https://radondb.com/docs/mysql/v2.2.0/installation/on_kubernetes/#content参数:https://github.com/radondb/radondb-mysql-kubernetes/blob/main/docs/zh-cn/config_para.md官网:https://radondb.comhelmrepoaddradondbhttps://radondb.github.......
  • 一个mysql的group_concat导致的问题
    好久都没有写点东西了,是时候有点写东西的必要了。去年下年底离职了,躺了几个月,最近又兜兜转转换了一家公司继续当牛马了,前段时间八股文背了好多,难受呀,不过我也趁着前段时间自己也整理了属于我自己的八股文,有好几万字吧,哈哈哈,以后就不用到处去找八股文了。说回正题,这......
  • Mysql练习
    1. 练习11.1 选择题   811代码在E:\java学习\初级\course166\db_exercise01--选择题写法对错判断811--错误,as可以省略但是AnnualSalary中间有空格会被认为Salary是最终的别名SELECTename,sal*12ASAnnualSalaryFROMempSELECTename,sal*12AS"AnnualSalary"FR......
  • MySQL学习进阶篇Day2
    2.索引2.1索引概述2.1.1介绍索引(index)是帮助MySQL高效获取数据的数据结构(有序)。在数据之外,数据库系统还维护着满足特定查找算法的数据结构,这些数据结构以某种方式引用(指向)数据,这样就可以在这些数据结构上实现高级查找算法,这种数据结构就是索引。 一提到数据结构,大......
  • MySQL数据类型
    数据类型的选取原则:更小的通常更好:尽量使用可以正确存储数据的最小数据类型(它们占用更少的磁盘,内存,cpu缓存,处理时需要的cpu周期也更少)简单:简单数据类型的操作通常所需cpu周期更少,例如整型就是比字符串类型的操作代价更低尽量避免NULL:最好指定列为NOTNULL,如果......
  • ubuntu 20.4 安装msyql+创建mysql用户
    sudoapt-getupdatesudoapt-getupgratesudoaptinstallmysql-server设置运行远程访问修改:sudovim/etc/mysql/mysql.conf.d/mysqld.cnf用#注释掉bind-address=127.0.0.1进入mysql创建用户mysql-uroot-pCREATEUSER'test'@'%'IDENTIFIEDBY'12345......
  • MySQL的间隙锁
    什么是间隙锁?间隙锁就是在MySQL的一个范围锁,对某个不存在数据的范围进行加锁,加锁后,不能在这个范围内插入数据;在可重复读事务隔离级别下,默认使用的是next-keyLock(行锁+间隙锁);间隙锁是为了解决什么问题?间隙锁就是MySQL在Innodb存储引擎在可重复读的事务隔离级别下为了解决当前读......
  • MySQL刷题记录
    1. select*fromemployeesorderbyhire_datedesclimit1; 笔记: limit 0,1;使用limit关键字从第0条记录向后读取一个,也就是第一条记录  2.select*fromtestLIMIT 3OFFSET 1;(在mysql 5以后支持这种写法)当limit和offset组合使用的时候,limit后面只......