Redis和MySQL如何保持緩存一致性？阿里Canal告訴你

作者：王振軍 2023-07-27 08:29:09

Canal 是用 Java 開發的基于數據庫增量日志解析，提供增量數據訂閱&消費的中間件。目前，Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 來處理獲得的相關數據。（數據庫同步需要阿里的 Otter 中間件，基于 Canal）。

一、簡介

canal [k?'n?l]，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數據庫增量日志解析，提供增量數據訂閱和消費。早期阿里巴巴因為杭州和美國雙機房部署，存在跨機房同步的業務需求，實現方式主要是基于業務 trigger 獲取增量變更。從 2010 年開始，業務逐步嘗試數據庫日志解析獲取增量變更進行同步，由此衍生出了大量的數據庫增量訂閱和消費業務。

當前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x。

canal github地址：https://github.com/alibaba/canal。

二、MySQL 的 Binlog

1、Binlog介紹

MySQL 的二進制日志可以說 MySQL 最重要的日志了，它記錄了所有的 DDL 和 DML(除了數據查詢語句)語句，以事件形式記錄，還包含語句所執行的消耗的時間，MySQL 的二進制日志是事務安全型的。一般來說開啟二進制日志大概會有 1%的性能損耗。二進制有兩個最重要的使用場景：

MySQL Replication 在 Master 端開啟 Binlog，Master 把它的二進制日志傳遞給 Slaves 來達到 Master-Slave 數據一致的目的，這就是我們常用的主從復制。
就是數據恢復了，通過使用 MySQL Binlog 工具來使恢復數據，生產上要開啟，不然真的要刪庫跑路了。

2、Binlog 的分類

MySQL Binlog 的格式有三種，分別是 STATEMENT,MIXED,ROW。在配置文件中可以選擇配置 binlog_format= statement|mixed|row。

statement：語句級，binlog 會記錄每次一執行寫操作的語句。比如update user set create_date=now()優點：節省空間。缺點：有可能造成數據不一致。
row：行級， binlog 會記錄每次操作后每行記錄的變化。優點：保持數據的絕對一致性缺點：占用較大空間
mixed：statement 的升級版，一定程度上解決了，因為一些情況而造成的 statement 模式不一致問題，默認還是 statement，一些會產生不一致的情況還是會選擇row。

==綜合對比== Canal 想做監控分析，選擇 row 格式比較合適。

三、工作原理

1、MySQL主備復制原理

MySQL master 將數據變更寫入二進制日志( binary log, 其中記錄叫做二進制日志事件binary log events，可以通過 show binlog events 進行查看)。
MySQL slave 將 master 的 binary log events 拷貝到它的中繼日志(relay log)。
MySQL slave 重放 relay log 中事件，將數據變更反映它自己的數據

2、canal 工作原理

canal 模擬 MySQL slave 的交互協議，偽裝自己為 MySQL slave ，向 MySQL master 發送dump 協議。
MySQL master 收到 dump 請求，開始推送 binary log 給 slave (即 canal )。
canal 解析 binary log 對象(原始為 byte 流)。

總結：

我們可以把canal理解為從機，拿到數據然后進行后續操作，可以同步到redis上，再也不需要進行延遲雙刪來保證mysql和redis的數據一致性了，而且還不會出現各種各樣的問題！

四、canal使用場景

場景一：阿里 Otter 中間件的一部分 Otter 是阿里用于進行異地數據庫之間的同步框架，Canal 是其中一部分。

otter github地址：https://github.com/alibaba/otter。

場景二：保證緩存和數據庫一致性（我們今天要測試的）。

場景三：實時數據分析。

抓取業務表的新增變化數據，用于制作實時統計。

五、安裝MySQL、redis

1、安裝MySQL

sudo docker run -p 3306:3306 --name mysql \
-v /mydata/mysql/log:/var/log/mysql \
-v /mydata/mysql/data:/var/lib/mysql \
-v /mydata/mysql/conf:/etc/mysql \
-e MYSQL_ROOT_PASSWORD=root \
-d mysql:5.7

2、Docker配置MySQL

vim /mydata/mysql/conf/my.cnf # 創建并進入編輯

添加如下配置：

[client]
default-character-set=utf8
[mysql]
default-character-set=utf8
[mysqld]
init_cnotallow='SET collation_connection = utf8_unicode_ci'
init_cnotallow='SET NAMES utf8'
character-set-server=utf8
collation-server=utf8_unicode_ci
skip-character-set-client-handshake
skip-name-resolve
# 開啟binlog日志：目錄為docker里的目錄
log-bin=/var/lib/mysql/mysql-bin
# server_id 需保證唯一，不能和 canal 的 slaveId 重復
server-id=123456
binlog_format=row
# test數據庫開啟，不設置則所有庫開啟
binlog-do-db=test

3、重新啟動MySQL

docker restart mysql

4、創建用戶并賦權限

查看mysql的 id：

docker ps

進入docker容器：

docker exec -it 7d /bin/bash

連接到mysql：

mysql -u root -p

創建用戶并賦予權限：

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

刷新：

flush privileges;

5、Win10連接mysql創建user表

CREATE TABLE `user`  (
  `id` int(10) NOT NULL AUTO_INCREMENT,
  `name` varchar(25) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `sex` varchar(1) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 8 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;

6、創建redis

docker run -p 6379:6379 --name redis \
-v /mydata/redis/data:/data \
-v /mydata/redis/conf/redis.conf:/etc/redis/redis.conf \
-d redis redis-server /etc/redis/redis.conf

六、安裝canal

1、啟動容器

docker run -it --name canal -p 11111:11111 -d canal/canal-server:v1.1.5

查看三個容器：

docker ps

2、配置canal

進入容器：

docker exec -it 56 /bin/bash

切換目錄：

cd canal-server/conf/example

修改兩個地方：

「第一個是mysql的地址，第二個是我們創建數據庫名字（可以使用默認帶的，就是全部的庫都進行收集binlog日志）」

canal.instance.master.address=192.168.84.138:3306

canal.instance.filter.regex=test\..*

3、查看日志

我們查看一下canal的日志，看是否啟動成功！首先進入容器：

docker exec -it 56 /bin/bash

切換目錄：

cd canal-server/logs/example/

查看日志：

cat example.log

無報錯，剛剛新建的表這里也可以檢測到！

4、查看canal.properties

cd /canal-server/conf

cat canal.properties

我們可以看到有很多個模式，可以把canal收集到的binlog發送到三大MQ中，或者tcp。

本次以tcp為準測試，如果大家有需求可以進行發送到MQ，往下滑都有對應的配置！

七、簡單測試

1、新建springboot項目，導入依賴

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
    <exclusions>
        <exclusion>
            <groupId>org.junit.vintage</groupId>
            <artifactId>junit-vintage-engine</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.0</version>
</dependency>
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-core</artifactId>
    <version>2.8.6</version>
</dependency>
<dependency>
    <groupId>com.fasterxml.jackson.module</groupId>
    <artifactId>jackson-module-jaxb-annotations</artifactId>
    <version>2.8.6</version>
</dependency>

2、編寫測試文件

來自官方例子：

我把statis關鍵字刪除了，方便和redis進行整合。

例子地址：https://github.com/alibaba/canal/wiki/ClientExample。

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.client.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * @author wangzhenjun
 * @date 2022/6/29 9:31
 */
@Configuration
public class SimpleCanalClientExample {

//    private static String REDIS_DATABASE = "mall";
//    private static String REDIS_KEY_ADMIN = "ums:admin";

    @Bean
    public void canalSync() {
        // 創建鏈接,第一個參數是canal的ip，第二個參數是canal的端口號，
        // 第三個參數是canal虛擬的模塊名稱，canal是創建的數據庫賬號密碼
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.84.138",
                11111), "example", "canal", "canal");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            // 對應上面的配置只對test庫進行獲取binlog文件
            connector.subscribe("test\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 獲取指定數量的數據
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交確認
                // connector.rollback(batchId); // 處理失敗, 回滾數據
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<Entry> 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.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 void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}

3、啟動項目

4、插入一條數據

INSERT INTO user VALUES (1,'小紅','女');

總結： 我們測試是可以獲取到binlog日志的，下面我們進入實戰：實現redis緩存同步

八、實戰redis同步緩存

1、編寫redis序列化配置類

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @author wangzhenjun
 * @date 2022/6/30 9:24
 */
@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringRedisSerializer);
        redisTemplate.setHashKeySerializer(stringRedisSerializer);
        Jackson2JsonRedisSerializer<?> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

}

2、添加redis增刪改方法

主要添加了同步到redis的兩個方法，這里是2分鐘就會停止監聽，大家可以按自己的來調整：

int totalEmptyCount = 120;

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

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.client.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
/**
 * @author wangzhenjun
 * @date 2022/6/29 9:31
 */
@Configuration
public class SimpleCanalClientExample {

    @Autowired
    private RedisTemplate redisTemplate;

    private static final String KEY = "user:info";

    @Bean
    public void canalSync() {
        // 創建鏈接,第一個參數是canal的ip，第二個參數是canal的端口號，
        // 第三個參數是canal虛擬的模塊名稱，canal是創建的數據庫賬號密碼
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.84.138",
                11111), "example", "canal", "canal");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            // 對應上面的配置只對test庫進行獲取binlog文件
            connector.subscribe("test\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 獲取指定數量的數據
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    emptyCount++;
                    System.out.println("empty count : " + emptyCount);
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    emptyCount = 0;
                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);
                    printEntry(message.getEntries());
                }

                connector.ack(batchId); // 提交確認
                // connector.rollback(batchId); // 處理失敗, 回滾數據
            }

            System.out.println("empty too many times, exit");
        } finally {
            connector.disconnect();
        }
    }

    private void printEntry(List<Entry> 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.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                    // 同步到redis
                    delete(rowData.getBeforeColumnsList());
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList());
                    // 同步到redis
                    insertOrUpdate(rowData.getAfterColumnsList());
                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList());
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList());
                    // 同步到redis
                    insertOrUpdate(rowData.getAfterColumnsList());
                }
            }
        }
    }

    private void printColumn(List<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }

    /**
     * 更新或者添加觸發同步到redis
     * @param columns
     */
    private void insertOrUpdate (List<Column> columns) {
        if (columns.size() > 0) {
            JSONObject json = new JSONObject();
            for (Column column : columns) {
                json.put(column.getName(), column.getValue());
            }
            redisTemplate.opsForHash().put(KEY,columns.get(0).getValue(),json.toJSONString());
        }
    }

    /**
     * 刪除觸發同步到redis
     * @param columns
     */
    private void delete (List<Column> columns) {
        if (columns.size() > 0) {
            redisTemplate.opsForHash().delete(KEY, columns.get(0).getValue());
        }
        
    }
}

3、測試添加

數據庫插入一條：

insert into user values (1,'我是測試添加','男');

控制臺捕捉到信息：

我們看到redis已經有數據了，同步成功！

4、測試更新

更細我們剛剛添加的那條數據：

update user set name = '修改了' where id = 1;

控制臺捕捉到了更新信息：

redis也同步修改了！

5、測試刪除

我們先多添加幾條哈：

刪除id為1的那條數據：

delete from user where id = 1;

控制臺捕捉到了刪除信息：

redis也同步刪除了！

九、總結

這樣就實現了一個canal的應用場景，當然也可以把binlog的數據發送到MQ來！

責任編輯：姜華來源：小王博客基地

MySQL redis