自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

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

作者：王振軍
數(shù)據(jù)庫 MySQL
Canal 是用 Java 開發(fā)的基于數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱&消費(fèi)的中間件。目前，Canal 主要支持了 MySQL 的 Binlog 解析，解析完成后才利用 Canal Client 來處理獲得 的相關(guān)數(shù)據(jù)。（數(shù)據(jù)庫同步需要阿里的 Otter 中間件，基于 Canal）。

一、簡介

canal [k?'n?l]，譯意為水道/管道/溝渠，主要用途是基于 MySQL 數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱和消費(fèi)。早期阿里巴巴因?yàn)楹贾莺兔绹p機(jī)房部署，存在跨機(jī)房同步的業(yè)務(wù)需求，實(shí)現(xiàn)方式主要是基于業(yè)務(wù) trigger 獲取增量變更。從 2010 年開始，業(yè)務(wù)逐步嘗試數(shù)據(jù)庫日志解析獲取增量變更進(jìn)行同步，由此衍生出了大量的數(shù)據(jù)庫增量訂閱和消費(fèi)業(yè)務(wù)。

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

當(dāng)前的 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 的二進(jìn)制日志可以說 MySQL 最重要的日志了，它記錄了所有的 DDL 和 DML(除 了數(shù)據(jù)查詢語句)語句，以事件形式記錄，還包含語句所執(zhí)行的消耗的時間，MySQL 的二進(jìn) 制日志是事務(wù)安全型的。一般來說開啟二進(jìn)制日志大概會有 1%的性能損耗。二進(jìn)制有兩個最重要的使用場景：

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

2、Binlog 的分類

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

  • statement：語句級，binlog 會記錄每次一執(zhí)行寫操作的語句。比如update user set create_date=now()優(yōu)點(diǎn)：節(jié)省空間。缺點(diǎn)：有可能造成數(shù)據(jù)不一致。
  • row：行級， binlog 會記錄每次操作后每行記錄的變化。優(yōu)點(diǎn)：保持?jǐn)?shù)據(jù)的絕對一致性 缺點(diǎn)：占用較大空間
  • mixed：statement 的升級版，一定程度上解決了，因?yàn)橐恍┣闆r而造成的 statement 模式不一致問題，默認(rèn)還是 statement，一些會產(chǎn)生不一致的情況還是會選擇row。

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

三、工作原理

1、MySQL主備復(fù)制原理

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

2、canal 工作原理

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

總結(jié)：

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

四、canal使用場景

場景一：阿里 Otter 中間件的一部分 Otter 是阿里用于進(jìn)行異地?cái)?shù)據(jù)庫之間的同步框架，Canal 是其中一部分。

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

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

場景三：實(shí)時數(shù)據(jù)分析。

抓取業(yè)務(wù)表的新增變化數(shù)據(jù)，用于制作實(shí)時統(tǒng)計(jì)。

五、安裝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 # 創(chuàng)建并進(jìn)入編輯

添加如下配置：

[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 重復(fù)
server-id=123456
binlog_format=row
# test數(shù)據(jù)庫開啟，不設(shè)置則所有庫開啟
binlog-do-db=test

3、重新啟動MySQL

docker restart mysql

4、創(chuàng)建用戶并賦權(quán)限

查看mysql的 id：

docker ps

進(jìn)入docker容器：

docker exec -it 7d /bin/bash

連接到mysql：

mysql -u root -p

創(chuàng)建用戶并賦予權(quán)限：

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

刷新：

flush privileges;

5、Win10連接mysql創(chuàng)建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、創(chuàng)建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

進(jìn)入容器：

docker exec -it 56 /bin/bash

切換目錄：

cd canal-server/conf/example

修改兩個地方：

「第一個是mysql的地址，第二個是我們創(chuàng)建數(shù)據(jù)庫名字（可以使用默認(rèn)帶的，就是全部的庫都進(jìn)行收集binlog日志）」

canal.instance.master.address=192.168.84.138:3306

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

3、查看日志

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

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發(fā)送到三大MQ中，或者tcp。

本次以tcp為準(zhǔn)測試，如果大家有需求可以進(jìn)行發(fā)送到MQ，往下滑都有對應(yīng)的配置！

七、簡單測試

1、新建springboot項(xiàng)目，導(dǎo)入依賴

<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關(guān)鍵字刪除了，方便和redis進(jìn)行整合。

例子地址：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() {
        // 創(chuàng)建鏈接,第一個參數(shù)是canal的ip，第二個參數(shù)是canal的端口號，
        // 第三個參數(shù)是canal虛擬的模塊名稱，canal是創(chuàng)建的數(shù)據(jù)庫賬號密碼
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.84.138",
                11111), "example", "canal", "canal");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            // 對應(yīng)上面的配置只對test庫進(jìn)行獲取binlog文件
            connector.subscribe("test\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 獲取指定數(shù)量的數(shù)據(jù)
                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); // 提交確認(rèn)
                // connector.rollback(batchId); // 處理失敗, 回滾數(shù)據(jù)
            }

            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、啟動項(xiàng)目

4、插入一條數(shù)據(jù)

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

總結(jié)： 我們測試是可以獲取到binlog日志的，下面我們進(jìn)入實(shí)戰(zhàn)：實(shí)現(xiàn)redis緩存同步

八、實(shí)戰(zhàn)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分鐘就會停止監(jiān)聽，大家可以按自己的來調(diào)整：

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() {
        // 創(chuàng)建鏈接,第一個參數(shù)是canal的ip，第二個參數(shù)是canal的端口號，
        // 第三個參數(shù)是canal虛擬的模塊名稱，canal是創(chuàng)建的數(shù)據(jù)庫賬號密碼
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("192.168.84.138",
                11111), "example", "canal", "canal");
        int batchSize = 1000;
        int emptyCount = 0;
        try {
            connector.connect();
            // 對應(yīng)上面的配置只對test庫進(jìn)行獲取binlog文件
            connector.subscribe("test\\..*");
            connector.rollback();
            int totalEmptyCount = 120;
            while (emptyCount < totalEmptyCount) {
                Message message = connector.getWithoutAck(batchSize); // 獲取指定數(shù)量的數(shù)據(jù)
                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); // 提交確認(rèn)
                // connector.rollback(batchId); // 處理失敗, 回滾數(shù)據(jù)
            }

            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());
        }
    }

    /**
     * 更新或者添加觸發(fā)同步到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());
        }
    }

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

3、測試添加

數(shù)據(jù)庫插入一條：

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

控制臺捕捉到信息：

我們看到redis已經(jīng)有數(shù)據(jù)了，同步成功！

4、測試更新

更細(xì)我們剛剛添加的那條數(shù)據(jù)：

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

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

redis也同步修改了！

5、測試刪除

我們先多添加幾條哈：

刪除id為1的那條數(shù)據(jù)：

delete from user where id = 1;

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

redis也同步刪除了！

九、總結(jié)

這樣就實(shí)現(xiàn)了一個canal的應(yīng)用場景，當(dāng)然也可以把binlog的數(shù)據(jù)發(fā)送到MQ來！

責(zé)任編輯：姜華 來源： 小王博客基地
MySQLredis
相關(guān)推薦

2022-02-17 21:04:27

數(shù)據(jù)庫MysqlRedis

2023-09-24 14:35:43

Redis數(shù)據(jù)庫

2022-12-14 08:23:30

2023-02-17 07:54:39

2022-12-27 08:56:28

2020-05-12 10:43:22

Redis緩存數(shù)據(jù)庫

2023-05-26 07:34:50

RedisMySQL緩存

2020-06-01 22:09:48

緩存緩存同步緩存誤用

2021-12-14 07:15:57

MySQLRedis數(shù)據(jù)

2022-09-15 10:37:46

MySQLRedis數(shù)據(jù)一致性

2024-05-28 00:50:00

RedisMySQL緩存

2017-06-27 09:40:28

MYSQL數(shù)據(jù)備份

2024-12-26 15:01:29

2024-04-11 13:45:14

Redis數(shù)據(jù)庫緩存

2017-07-25 14:38:56

數(shù)據(jù)庫一致性非鎖定讀一致性鎖定讀

2024-05-08 16:37:17

MySQLRedis數(shù)據(jù)庫

2024-04-23 08:22:58

Meta緩存系統(tǒng)

2024-10-28 12:41:25

2022-03-29 10:39:10

緩存數(shù)據(jù)庫數(shù)據(jù)

2024-01-15 10:38:20

多級緩存數(shù)據(jù)一致性分布式緩存
點(diǎn)贊
收藏

51CTO技術(shù)棧公眾號