十六、Redis与MySQL数据双写一致性 —— Canal Demo

一、认识canal

1、是什么？

canal，中文翻译为水道/管道/沟渠/运河，主要用途是用于 MySQL 数据库增量日志（binlog）数据的订阅、消费和解析，是阿里巴巴开发并开源的，采用Java语言开发；

历史背景是早期阿里巴巴因为杭州和美国双机房部署，存在跨机房数据同步的业务需求，实现方式主要是基于业务 trigger（触发器）获取增量变更。从2010年开始，阿里巴巴逐步尝试采用解析数据库日志获取增量变更进行同步，由此衍生出了canal项目。

2、能干嘛？

数据库镜像
数据库实时备份
索引构建和实时维护(拆分异构索引、倒排索引等)
业务 cache 刷新
带业务逻辑的增量数据处理

3、去哪下？

下载canal：https://github.com/alibaba/canal/wiki/QuickStart

java案例：https://github.com/alibaba/canal/wiki/ClientExample

二、canal工作原理

1、传统MySQL主从复制工作原理

MySQL的主从复制将经过如下步骤：

1、当 master 主服务器上的数据发生改变时，则将其改变写入二进制事件日志文件中（binlog）；
2、salve 从服务器会在一定时间间隔内对 master 主服务器上的二进制日志进行探测，探测其是否发生过改变，如果探测到 master 主服务器的二进制事件日志发生了改变，则开始一个 I/O Thread 请求 master 二进制事件日志；
3、同时 master 主服务器为每个 I/O Thread 启动一个dump Thread，用于向其发送二进制事件日志；
4、slave 从服务器将接收到的二进制事件日志保存至自己本地的中继日志文件中；
5、salve 从服务器将启动 SQL Thread 从中继日志中读取二进制日志，在本地重放，使得其数据和主服务器保持一致；
6、最后 I/O Thread 和 SQL Thread 将进入睡眠状态，等待下一次被唤醒；

2、canal工作原理

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

3、结论

分布式系统只有最终一致性，很难做到强一致性。

三、mysql配置

1、查看是否开启binlog

SHOW VARIABLES LIKE 'log_bin';

2、开启 MySQL的binlog写入功能

Windows文件名为：my.ini

linux文件名为：my.cnf

log-bin=mysql-bin #开启 binlog
binlog-format=ROW #选择 ROW 模式
server_id=1    #配置MySQL replaction需要定义，不要和canal的 slaveId重复

ROW：模式除了记录sql语句之外，还会记录每个字段的变化情况，能够清楚的记录每行数据的变化历史，但会占用较多的空间。
STATEMENT：模式只记录了sql语句，但是没有记录上下文信息，在进行数据恢复的时候可能会导致数据的丢失情况；
MIX：模式比较灵活的记录，理论上说当遇到了表结构变更的时候，就会记录为statement模式。当遇到了数据更新或者删除情况下就会变为row模式；

3、重启mysql

service mysqld restart

4、授权canal连接MySQL账号

mysql默认的用户在mysql库的user表里

默认没有canal账户，此处新建+授权

DROP USER 'canal'@'%';
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';  
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';  
FLUSH PRIVILEGES;

如果提示密码策略限制，使用如下策略：

set global validate_password_policy=LOW;
set global validate_password_length=5;

四、canal服务端

1、下载

如以 1.0.17 版本为例

wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.deployer-1.0.17.tar.gz

2、创建解压文件夹

mkdir /tmp/canal

3、解压

tar zxvf canal.deployer-$version.tar.gz  -C /tmp/canal

如果无法联网下载，则进入tar包文件目录打开终端解压。下载地址

https://github.com/alibaba/canal/releases

解压后的文件canal文件目录结构

bin：启动、停止等脚本命令
conf：配置文件目录
lib：相关依赖jar包
logs：日志目录
plugin：相关mq依赖插件jar包

4、修改配置

vi conf/example/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.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型，比如 UTF-8，GBK , ISO-8859-1
如果系统是1个 cpu，需要将 canal.instance.parser.parallel 设置为 false

5、启动及其他操作命令

在../canal.deployer-1.1.5/bin路径下执行

./startup.sh

五、Springboot 配置

1、POM文件

        
            com.alibaba.otter
            canal.client
            1.1.0
        
        
        
            redis.clients
            jedis
            3.1.0
        

        
            com.alibaba
            fastjson
            1.2.73

2、业务类——RedisUtils

package canal_demo.shiwn.config;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisUtils {
    public static JedisPool jedisPool;

    static {
        JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(20);
        jedisPoolConfig.setMaxIdle(10);
        jedisPool=new JedisPool(jedisPoolConfig,"192.168.1.123",6379);
    }

    public static Jedis getJedis() throws Exception {
        if(null!=jedisPool){
            return jedisPool.getResource();
        }
        throw new Exception("Jedispool is not ok");
    }

}

3、业务类——RedisCanalClientExample

package canal_demo.shiwn;

import canal_demo.shiwn.config.RedisUtils;
import com.alibaba.fastjson.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 redis.clients.jedis.Jedis;

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

public class RedisCanalClientExample {
    public static final Integer _60SECONDS = 60;

    public static void main(String args[]) {

        // 创建链接canal服务端，默认端口11111
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.1.123",
                11111), "example", "", "");
        // 查询条数
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            //connector.subscribe(".*\\..*");
            //  指定数据库及表
            connector.subscribe("db2022.t_user");
            connector.rollback();
            int totalEmptyCount = 10 * _60SECONDS;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    emptyCount = 0;
                    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 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("================> 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.INSERT) {
                    redisInsert(rowData.getAfterColumnsList());
                } else if (eventType == EventType.DELETE) {
                    redisDelete(rowData.getBeforeColumnsList());
                } else {//EventType.UPDATE
                    redisUpdate(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private static void redisInsert(List columns) {
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            jsonObject.put(column.getName(), column.getValue());
        }
        if (columns.size() > 0) {
            try (Jedis jedis = RedisUtils.getJedis()) {
                jedis.set(columns.get(0).getValue(), jsonObject.toJSONString());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static void redisDelete(List columns) {
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns) {
            jsonObject.put(column.getName(), column.getValue());
        }
        if (columns.size() > 0) {
            try (Jedis jedis = RedisUtils.getJedis()) {
                jedis.del(columns.get(0).getValue());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static void redisUpdate(List columns) {
        JSONObject jsonObject = new JSONObject();
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            jsonObject.put(column.getName(), column.getValue());
        }
        if (columns.size() > 0) {
            try (Jedis jedis = RedisUtils.getJedis()) {
                jedis.set(columns.get(0).getValue(), jsonObject.toJSONString());
                System.out.println("---------update after: " + jedis.get(columns.get(0).getValue()));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Redis

十六、Redis与MySQL数据双写一致性 —— Canal Demo

一、认识canal

1、是什么？

2、能干嘛？

3、去哪下？

二、canal工作原理

1、传统MySQL主从复制工作原理

2、canal工作原理

3、结论

三、mysql配置

1、查看是否开启binlog

2、开启 MySQL的binlog写入功能

3、重启mysql

4、授权canal连接MySQL账号

四、canal服务端

1、下载

2、创建解压文件夹

3、解压

4、修改配置

5、启动及其他操作命令

五、Springboot 配置

1、POM文件

2、业务类——RedisUtils

3、业务类——RedisCanalClientExample

相关

ceryx+redis实现nginx动态路由功能、动态管理upstream

Go入门笔记-20 访问redis

为毛我的redis 非保护启动修改文件有问题呢？

org.springframework.data.redis.RedisConnectionFailureException

redis集群模式

hyperf 框架 redis 安装和使用

Laravel中Redis的使用

Win10安装Redis

redis哨兵原理

Redis高可用方案：Sentinel哨兵

那一天，我被Redis主从架构支配的恐惧

Redis | 第一部分：数据结构与对象上篇《Redis设计与实现》

标签