• Canal1--搭建Canal监听数据库变化


    1.安装mysql

    默认安装了mysql(版本8.0.x);

    新创建用户

    -- 创建用户 用户名:canal 密码:Canal@123456
    create user 'canal'@'%' identified by 'Canal@123456';
    
    • 1
    • 2

    授权

    grant SELECT, REPLICATION SLAVE, REPLICATION CLIENT on *.* to 'canal'@'%' with grant option;
    
    flush privileges;
    
    • 1
    • 2
    • 3

    在这里插入图片描述

    查看MySQL是否开启binlog模式

    show variables like 'log_bin';
    
    • 1

    在这里插入图片描述
    查看当前正在写入的binlog日志:

    show master status;
    
    
    • 1
    • 2

    在这里插入图片描述
    记住文件名和偏移量

    2.安装Canal

    官网下载页面进行下载;

    我这里下载的是1.1.7的版本:
    在这里插入图片描述
    解压canal.deployer-1.1.7.tar.gz,我们可以看到里面有五个文件夹
    在这里插入图片描述
    打开配置文件conf/example/instance.properties,配置信息如下:

    ## mysql serverId , v1.0.26+ will autoGen
    ## v1.0.26版本后会自动生成slaveId,所以可以不用配置
    # canal.instance.mysql.slaveId=0
    
    # 数据库地址
    canal.instance.master.address=127.0.0.1:3306
    # binlog日志名称
    canal.instance.master.journal.name=自己的日志名称
    # mysql主库链接时起始的binlog偏移量
    canal.instance.master.position=日志的偏移量
    # mysql主库链接时起始的binlog的时间戳
    canal.instance.master.timestamp=
    canal.instance.master.gtid=
    
    # username/password
    # 在MySQL服务器授权的账号密码
    canal.instance.dbUsername=canal
    canal.instance.dbPassword=Canal@123456
    # 字符集
    canal.instance.connectionCharset = UTF-8
    # enable druid Decrypt database password
    canal.instance.enableDruid=false
    
    # table regex .*\\..*表示监听所有表 也可以写具体的表名,用,隔开
    canal.instance.filter.regex=.*\\..*
    # mysql 数据解析表的黑名单,多个表用,隔开
    canal.instance.filter.black.regex=
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

    开启Canal服务端:
    进入bin目录

    .\startup.bat
    
    • 1

    3.Java客户端操作

    首先引入maven依赖:

    <!--canal客户端-->
            <dependency>
                <groupId>com.alibaba.otter</groupId>
                <artifactId>canal.client</artifactId>
                <version>1.1.7</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba.otter</groupId>
                <artifactId>canal.protocol</artifactId>
                <version>1.1.7</version>
            </dependency>
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    然后创建一个CanalClient类

    import com.alibaba.otter.canal.client.*;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.stereotype.Component;
    import com.alibaba.otter.canal.protocol.Message;
    import com.alibaba.otter.canal.protocol.CanalEntry.*;
    
    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<Entry> entrys) {
            for (Entry entry : entrys) {
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                    //开启/关闭事务的实体类型,跳过
                    continue;
                }
                //RowChange对象,包含了一行数据变化的所有特征
                //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
                RowChange rowChage;
                try {
                    rowChage = 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类型
                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 (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());
            }
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93
    • 94
    • 95
    • 96
    • 97
    • 98
    • 99
    • 100
    • 101
    • 102
    • 103
    • 104
    • 105
    • 106
    • 107
    • 108
    • 109
    • 110
    • 111
    • 112

    完成之后再对应数据库进行操作,控制台会打印对应的操作,说明对数据的写入操作进行了有效的监控;
    注意只读操作并不会写入binlog也不会被Canal监控到(也没必要监控读取操作)。
    在这里插入图片描述

  • 相关阅读:
    Sentinel vs Hystrix 限流到底怎么选?(荣耀典藏版)
    帧间预测一些概念
    异常和异常处理
    【树莓派】如何用电脑连接树莓派的远程桌面,灰屏解决
    设计模式之单例模式
    sql语句 (增删改查)
    什么是虚拟主播?
    【祝福伟大的祖国】Java Web 9.2 Request 对象 9.2.2 Request 获取请求数据
    【深度学习】高速神经网络(Highway Network)
    基于nodejs+vue 衣服穿搭推荐系统
  • 原文地址:https://blog.csdn.net/joey_ro/article/details/138078255