搭建本地Canal中间件进行数据迁移 —— 从缓存击穿得到的启发

🍊 Java学习：Java从入门到精通总结

🍊 Spring系列推荐：Spring源码解析

🍊 绝对不一样的职场干货：大厂最佳实践经验指南


📆 最近更新：2022年4月21日

🍊 个人简介：通信工程本硕💪、Java程序员🌕。我的故事充满机遇、挑战与翻盘，欢迎关注作者来共饮一杯鸡汤

文章目录

• 先从缓存击穿说起
• MySQL数据同步
• Canal工作原理
• 搭建本地Canal中间件进行数据迁移

先从缓存击穿说起

所谓缓存击穿，就是热点数据在缓存中没有数据，大量用户请求直接打在数据库上，这是一种非常危险的情况，我们应该在开发过程中避免此过程的发生。

目前我们经常使用的解决方案一共有3种：

• 缓存过期策略方案： 使用读写分离架构，读取永远从缓存里读，DB和缓存之间的同步使用Canal

• 热点缓存策略： 甄别出热点数据后，将针对热点数据的请求打到特别的一个区域

• 互斥锁： 在数据快要到期之前，给key加上一个互斥锁，如果其他请求过来时，如果发现当前key已经处于锁定状态，说明在到期时前已经有一个请求去后台获取数据值了，其他的请求可以大胆地从缓存中读取数据。去数据库拿数据的请求可以异步刷新缓存，刷新完成之后可以将互斥锁去掉

Guava的解决方案：

• Guava Cache 在load 的时候做了并发控制，在多个线程请求一个不存在或者过期的缓存项时保证只有一个线程进入load 方法，其他线程等待直到缓存项被生成，这样就避免了大量的线程击穿缓存直达DB 。
• 通过refreshAfterWrite 实现的，在配置刷新策略后，对应的缓存项会按照设
  定的时间定时刷新，避免线程阻塞的同时保证缓存项处于最新状态；前提是已经生成缓存项了。在实际生产情况下我们可以做缓存预热，提前生成缓存项，避免流量洪峰造成的线程堆积。

具体如何取舍，还是要结合义务来看。如果要使用读写分离架构的话，就涉及到使用Canal中间件来进行数据的同步，本文就来带领大家搭建一个本地Demo来看一看是怎么个玩法~

MySQL数据同步

数据同步的主要流程如下图所示：

Master每次更新数据时，都会把它写入到本地的bin log里，Slave通过IO线程把Master的bin log同步到本地的relay log里，然后再通过一个SQL线程把relay log执行到数据库当中

Canal工作原理

Canal通过伪造Dump协议把自己伪装成一个Slave来向Master同步数据，

搭建本地Canal中间件进行数据迁移

下载MySQL-community版
https://dev.mysgl.com/downloads/mysql/

下载Canal
https://github.com/alibaba/canal/releases/tag/canal-1.1.4

windows下有一个小bug需要改（mac版的电脑可以无视），编辑bin下面的startup.bat：

之后双击启动startup.bat，如果logs文件夹下的日志里没有报错信息则启动成功

开启MySQL的Bin Log
找到MySQL配置文件my.cnf，把下面代码段放到配置文件里

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

创建测试库
在测试库test下创建canal_test表，只设置一个id字段

创建用户

CREATE USER canal IDENTIFIED BY 'canal'; 
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; 
FLUSH PRIVILEGES;

Canal使用这个用户来连接上目标数据库，并且消费bin log的变更

pom坐标

<dependency>
    <groupId>com.alibaba.ottergroupId>
    <artifactId>canal.clientartifactId>
    <version>1.1.0version>
dependency>

Java代码

public class CanalStarter {
    public static void main(String[] args) throws InterruptedException, InvalidProtocolBufferException {
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(AddressUtils.getHostIp(), 11111),
                "example",  // destination
                "",          // username
                ""           // password
        );
        try{
            connector.connect();
            connector.subscribe(".*\..*"); // .*.\.* 表示所有数据库的变更
            connector.rollback();   // 回到上次读取的位置，即回到上一个数据库的bin log消费到的这条记录的位置上

            while (true){
                Message msg = connector.getWithoutAck(100);     // 拿100条数据

                if (msg == null || msg.getId() < 0 || msg.getEntries().size() == 0){
                    System.out.println("nothing consumed");
                    Thread.sleep(1000);
                    continue;
                }
                // 对变更的数据做处理
                printEntry(msg.getEntries());

                connector.ack(msg.getId()); // 确认这条消息已经被消费掉
            }
        }finally {
            connector.disconnect();
        }
    }
}

打印数据变更前后内容的方法：

    private static void printEntry(List<CanalEntry.Entry> entries) throws InvalidProtocolBufferException {
        for (CanalEntry.Entry entry : entries) {
            // 过滤掉一些类型
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ||
                    entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND
            ){
                continue;
            }

            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                System.out.println("event type " + rowChange.getEventType());

                System.out.println("********** before change");
                printRowData(rowData.getBeforeColumnsList());

                System.out.println("********** after change");
                printRowData(rowData.getAfterColumnsList());
            }

        }
    }

这里会过滤掉CanalEntry.EntryType.TRANSACTIONBEGIN和CanalEntry.EntryType.TRANSACTIONEND这种和数据本身变更无关的变更

打印某一行数据的方法：

    private static void printRowData(List<CanalEntry.Column> columns){
        if (CollectionUtils.isEmpty(columns)){
            return;
        }
        columns.forEach(e -> {
            System.out.println(e.getName() + " : " + e.getValue());
        });
    }

启动源码之后修改一下原表中的数据：

INSERT INTO canal_test (id) VALUES (2022);

由于是新插入的数据，所以before没有，after打印出了新插入的数据

UPDATE canal_test SET id = 2023 WHERE id = 2022;