Redis 与 MySQL 数据双写一致性工程落地案例

面试题

先动 MySQL，再动 Redis，两害相衡取其轻，避免 Redis 业务 Key 突然消失，多线程请求集火打满 MySQL
动，写操作，先更新数据库，再删除缓存。尝试使用双检加锁机制锁住 MySQL，只让一个请求线程回写 Redis，完成数据一致性
当 MySQL 有记录改动（有增删改操作），需要立刻同步反应到 Redis，该如何做？

canal

概述

官网：https://github.com/alibaba/canal/wiki (opens in a new tab)

canal [kə'næl]，中文翻译为水道/管道/沟渠/运河，主要用途是用于 MySQL 数据库增量日志数据的订阅、消费和解析，是阿里巴巴开发并开源的，采用 Java 语言开发。

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

作用

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

下载

https://github.com/alibaba/canal/releases (opens in a new tab)

工作原理，面试回答

传统 MySQL 主从复制工作原理

MySQL-主从复制

MySQL 的主从复制步骤：

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

Canal 工作原理

Redis-canal工作原理

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

mysql-canal-redis 双写一致性 Demo

Java Demo

https://github.com/alibaba/canal/wiki/CLientExample (opens in a new tab)

MySQL

查看 mysql 版本
```
select version();
mysql 5.7.28
```
当前的主机二进制日志
```
show master status;
```
查看
```
SHOW VARIABLES LIKE 'log_bin';
```
开启 MySQL 的 binlog 写入功能

window my.ini [mysqld]

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 模式。
重启 MySQL
再次查看
```
SHOW VARIABLES LIKE 'log_bin';
```

授权 canal 连接 MySQL 账号

mysql 默认的用户在 mysql 库的 user 表里
```
SELECT * FROM mysql.`user`
```

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

DROP USER IF EXISTS 'canal'@'%';
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';  
GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';  
FLUSH PRIVILEGES;
 
SELECT * FROM mysql.user;

canal 服务端

下载：注意发布时间和版本，2022.8.11 后发布的才用
解压：解压后整体放在 /mycanal 路径下

配置

修改 /mycanal/conf/example 路径下 instance.properties 文件

instance.properties

# 换成自己的 ip
canal.instance.master.address=127.0.0.1:3306
 
# 换成自己在 mysql 新建的 canal 账户
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal

启动：/opt/mycanal/bin 路径下执行 ./startup.sh
查看 canal 是否启动成功，查看 server 日志，查看样例 example 的日志

canal 客户端（Java 编写业务程序）

SQL 脚本

CREATE TABLE `t_user` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `userName` varchar(100) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8mb4

POM

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
 
    <groupId>dev.matrixlab.canal</groupId>
    <artifactId>canal_demo</artifactId>
    <version>1.0-SNAPSHOT</version>
 
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.14</version>
        <relativePath/>
    </parent>
 
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <junit.version>4.12</junit.version>
        <log4j.version>1.2.17</log4j.version>
        <lombok.version>1.16.18</lombok.version>
        <mysql.version>5.1.47</mysql.version>
        <druid.version>1.1.16</druid.version>
        <mapper.version>4.1.5</mapper.version>
        <mybatis.spring.boot.version>1.3.0</mybatis.spring.boot.version>
    </properties>
 
    <dependencies>
        <!--canal-->
        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>1.1.0</version>
        </dependency>
        <!--SpringBoot通用依赖模块-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
        <!--swagger2-->
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>2.9.2</version>
        </dependency>
        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>2.9.2</version>
        </dependency>
        <!--SpringBoot与Redis整合依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-pool2</artifactId>
        </dependency>
        <!--SpringBoot与AOP-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>
        <dependency>
            <groupId>org.aspectj</groupId>
            <artifactId>aspectjweaver</artifactId>
        </dependency>
        <!--Mysql数据库驱动-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.47</version>
        </dependency>
        <!--SpringBoot集成druid连接池-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid-spring-boot-starter</artifactId>
            <version>1.1.10</version>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>${druid.version}</version>
        </dependency>
        <!--mybatis和springboot整合-->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>${mybatis.spring.boot.version}</version>
        </dependency>
        <!--通用基础配置junit/devtools/test/log4j/lombok/hutool-->
        <!--hutool-->
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.2.3</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>${junit.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>${lombok.version}</version>
            <optional>true</optional>
        </dependency>
        <!--persistence-->
        <dependency>
            <groupId>javax.persistence</groupId>
            <artifactId>persistence-api</artifactId>
            <version>1.0.2</version>
        </dependency>
        <!--通用Mapper-->
        <dependency>
            <groupId>tk.mybatis</groupId>
            <artifactId>mapper</artifactId>
            <version>${mapper.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
        </dependency>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.8.0</version>
        </dependency>
    </dependencies>
 
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
 
</project>

改 YML

application.properties

server.port=5555
 
# ========================alibaba.druid=====================
spring.datasource.type=com.alibaba.druid.pool.DruidDataSource
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/bigdata?useUnicode=true&characterEncoding=utf-8&useSSL=false
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.druid.test-while-idle=false

业务类

RedisUtils.java

package dev.matrixlab.canal.util;
 
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
 
public class RedisUtils {
    public static final String  REDIS_IP_ADDR = "127.0.0.1";
    public static final String  REDIS_pwd = "111111";
    public static JedisPool jedisPool;
 
    static {
        JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
        jedisPoolConfig.setMaxTotal(20);
        jedisPoolConfig.setMaxIdle(10);
        jedisPool=new JedisPool(jedisPoolConfig,REDIS_IP_ADDR,6379,10000,REDIS_pwd);
    }
 
    public static Jedis getJedis() throws Exception {
        if (null!=jedisPool) {
            return jedisPool.getResource();
        }
        throw new Exception("Jedispool is not ok");
    }
 
}

RedisCanalClientExample.java

package dev.matrixlab.canal.biz;
 
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 dev.matrixlab.canal.util.RedisUtils;
import redis.clients.jedis.Jedis;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
 
public class RedisCanalClientExample {
    public static final Integer _60SECONDS = 60;
    public static final String  REDIS_IP_ADDR = "127.0.0.1";
 
    private static void redisInsert(List<Column> 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<Column> 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<Column> 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();
            }
        }
    }
 
    public static void printEntry(List<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }
 
            RowChange rowChage = null;
            try {
                //获取变更的row数据
                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.INSERT) {
                    redisInsert(rowData.getAfterColumnsList());
                } else if (eventType == EventType.DELETE) {
                    redisDelete(rowData.getBeforeColumnsList());
                } else {//EventType.UPDATE
                    redisUpdate(rowData.getAfterColumnsList());
                }
            }
        }
    }
 
 
    public static void main(String[] args) {
        System.out.println("--------- initCanal() main方法-----------");
 
        //=================================
        // 创建链接canal服务端
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(REDIS_IP_ADDR,
                11111), "example", "", "");
        int batchSize = 1000;
        //空闲空转计数器
        int emptyCount = 0;
        System.out.println("---------------------canal init OK，开始监听mysql变化------");
        try {
            connector.connect();
            //connector.subscribe(".*\\..*");
            connector.subscribe("bigdata.t_user");
            connector.rollback();
            int totalEmptyCount = 10 * _60SECONDS;
            while (emptyCount < totalEmptyCount) {
                System.out.println("我是canal，每秒一次正在监听:"+ UUID.randomUUID().toString());
                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("已经监听了" + totalEmptyCount + "秒，无任何消息，请重启重试......");
        } finally {
            connector.disconnect();
        }
    }
}

java 程序下 connector.subscribe 配置的过滤正则

范围	规则
全库全表	`connector.subscribe(".\\..")`
指定库全表	`connector.subscribe(test\\..*")`
单表	`connector.subscribe("test.user")`
多规则组合使用	`connector.subscribe("test\\..*", test2.user1, test3.user2)`

关闭资源代码简写，try-with-resources 释放资源

jdk1.7 后增加了 try-with-resources ，他是一个声明一个或多个资源的 try 语句。一个资源作为一个对象，必须在程序结束之后关闭。try-with-resources 语句确保在语句的最后每个资源都被关闭，任何实现了 java.lang.AutoCloseable 和 java.io.Closeable 的对象都可以使用 try-with-resources 来实现异常处理和关闭资源。

缓存双写一致性之更新策略案例落地实战 HyperLogLog GEO Bitmap