Docker安裝canal、mysql進行簡單測試與實現redis和mysql快取一致性
2022-07-08 12:07:41
一、簡介
canal [kə'næl]，譯意為水道/管道/溝渠，主要用途是基於 MySQL 資料庫增量紀錄檔解析，提供增量資料訂閱和消費。
早期阿里巴巴因為杭州和美國雙機房部署，存在跨機房同步的業務需求，實現方式主要是基於業務 trigger 獲取增量變更。從 2010 年開始，業務逐步嘗試資料庫紀錄檔解析獲取增量變更進行同步，由此衍生出了大量的資料庫增量訂閱和消費業務。
Canal 是用 Java 開發的基於資料庫增量紀錄檔解析，提供增量資料訂閱&消費的中介軟體。
目前，Canal 主要支援了 MySQL 的 Binlog 解析，解析完成後才利用 Canal Client 來處理獲得
的相關資料。（資料庫同步需要阿里的 Otter 中介軟體，基於 Canal）。
當前的 canal 支援源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x。
二、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地址
場景二：保證快取和資料庫一致性（我們今天要測試的）
場景三：實時資料分析
抓取業務表的新增變化資料，用於製作實時統計
五、安裝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_connect='SET collation_connection = utf8_unicode_ci'
init_connect='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進行整合
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來！
小編整理了一天，看到這裡給小編點個關注唄，謝謝大家的支援哦！！
有緣人才可以看得到的哦！！！