Mysql和Redis数据同步之Canal

逃离我推掉我的手 2023-10-11 21:28 58阅读 0赞

canal是阿里研发的一款用于数据同步的中间件，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。工作原理如下：

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

## 准备阶段： ##

必须有Mysql数据库

## 第一步：下载canal工具并解压 ##

[Releases · alibaba/canal (github.com)][Releases _ alibaba_canal _github.com]

## 第二步：在Mysql主库中创建同步账号 ##

在mysql中执行下面命令

#创建一个名为 canal 密码为 root 的账户
    CREATE USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'root';
    
    #给canal账户授权
    GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    
    #刷新权限信息
    FLUSH PRIVILEGES;

## 第三步：canal环境搭建 ##

点击上面第一步解压的文件，找到并编辑`conf/example/instance.properties`文件

# slaveId随便取,只要不和其他从库id重复即可
    canal.instance.mysql.slaveId=1234567
    
    #对应需要canal监听的账户和密码
    canal.instance.dbUsername=canal
    canal.instance.dbPassword=root
    
    #这里还需要把下面位置连接到主库地址
    # position info
    canal.instance.master.address=127.0.0.1:3306

## 第四步：启动canal ##

进入canal解压的文件夹，找到canal中bin目录下的startup.bat，点击启动

## 第五步：java操作canal ##

导入依赖

<dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>2.0.12</version>
    </dependency>
    
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.2</version>
    </dependency>

写一个测试类：

package com.jdh.canal;
    
    import com.alibaba.fastjson2.JSONObject;
    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 com.google.protobuf.ByteString;
    import com.google.protobuf.InvalidProtocolBufferException;
    
    import java.net.InetSocketAddress;
    import java.net.SocketAddress;
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @ProjectName: springboot_java_study
     * @PackageUrl: com.jdh.canal
     * @ClassName: TestCanal
     * @Author: jdh
     * @CreateTime: 2023-04-03  15:25
     * @Description:
     * @Version: 1.0
     */
    public class TestCanal {
    
        public static void main(String[] args) {
            //canal的网络通信地址
            SocketAddress address = new InetSocketAddress("127.0.0.1", 11111);
            //canal连接器
            CanalConnector connector = CanalConnectors.newSingleConnector(address, "example", "canal", "root");
            connector.connect();//建立链接
            connector.subscribe("study.*");//订阅study库中的所有表
            //回滚到未进行ack的地方，相当于一个标记，下次获取的时候，可以从最后一个没有ack的地方开始拿
            connector.rollback();
            while (true) {
    //            Message message = connector.getWithoutAck(100, 2L, TimeUnit.SECONDS);//尝试获取100条数据，可能没有那么多，甚至没有
                //下面这个就是只要有事件就会触发更新操作
                Message message = connector.getWithoutAck(100);
                List<CanalEntry.Entry> entries = message.getEntries();//获取所有的数据
                if (entries.isEmpty()) {
    //                System.out.println("暂无数据变更");
                } else {
                    processEntries(entries);
                }
                //回执一个信息，表示当前这些发生变更的数据已经处理
                connector.ack(message.getId());
            }
        }
    
        private static void processEntries(List<CanalEntry.Entry> entries) {
            for (CanalEntry.Entry entry : entries) {
                //获取表名
                String tableName = entry.getHeader().getTableName();
                System.out.println("当前数据变更来源于：" + tableName);
                //获取类型
                CanalEntry.EntryType entryType = entry.getEntryType();
                if (entryType == CanalEntry.EntryType.ROWDATA) {//数据变更类型如果是一行数据
                    ByteString storeValue = entry.getStoreValue();//获取变更项序列化后的数据
                    try {
                        //解析数据，得到改变的行信息
                        CanalEntry.RowChange change = CanalEntry.RowChange.parseFrom(storeValue);
                        //获取触发改变的事件类型：插入，修改，删除
                        CanalEntry.EventType eventType = change.getEventType();
                        //获取所有的改变的数据行
                        List<CanalEntry.RowData> rowDataList = change.getRowDatasList();
                        for (CanalEntry.RowData rowData : rowDataList) {
                            processRowData(eventType, rowData);
                        }
                    } catch (InvalidProtocolBufferException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    
        private static void processRowData(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
    
    
            if (eventType == CanalEntry.EventType.DELETE) {//如果是删除动作
    //            Optional<CanalEntry.Column> opt = beforeColumnsList.stream().filter(column -> "id".equals(column.getName())).findFirst();
    //            if(opt.isPresent()){
    //                //获取id
    //                Long id = Long.valueOf(opt.get().getValue());
    //                //根据id去es中删除对应数据，需要考虑是sku的id还是spu的id
    //                System.out.println("根据id去删除es中的数据：" + id);
    //            }
    
                List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();//操作之前的数据
                JSONObject deleteBefore = getJsonObject(beforeColumnsList);
                System.out.println("删除操作之前的数据：" + deleteBefore.toString());
            } else if (eventType == CanalEntry.EventType.UPDATE) {//更新事件
                //更新事件需要考虑是审核状态的更新还是数据的更新
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                JSONObject json = new JSONObject();
                afterColumnsList.forEach(column -> json.put(column.getName(), column.getValue()));
                System.out.println("监听到改变数据：" + json);
            } else if (eventType == CanalEntry.EventType.INSERT) {//如果是插入动作
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();//插入操作之后的数据
                JSONObject insertAfter = getJsonObject(afterColumnsList);
                System.out.println("监听到插入之后的数据：" + insertAfter);
            }
        }
    
        private static JSONObject getJsonObject(List<CanalEntry.Column> ColumnsList) {
            JSONObject jsonObject = new JSONObject();
            for (CanalEntry.Column column : ColumnsList) {
                String columnName = column.getName();
                String columnValue = column.getValue();
                jsonObject.put(columnName, columnValue);
            }
            return jsonObject;
        }
    }

## 第六步：测试 ##

先启动上面的java代码

然后再mysql中随意进行数据的增删改操作

然后看java控制台打印信息

如果监听到对应的数据操作就会打印出来

第七步：同步redis

这步就是拿到canal监听到变化的数据，并对redis中的数据进行更新即可

演示略

[Releases _ alibaba_canal _github.com]: https://github.com/alibaba/canal/releases