Docker安裝canal、mysql進行簡單測試與實現redis和mysql快取一致性
一、簡介
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來！
小編整理了一天，看到這裡給小編點個關注唄，謝謝大家的支援哦！！
有緣人才可以看得到的哦！！！
