使用Canal和Kafka解決MySQL與緩存的數據一致性問題
1. 準備工作
1. 開啟并配置MySQL的 BinLog（MySQL 8.0 默認開啟）
修改配置：C:\ProgramData\MySQL\MySQL Server 8.0\my.ini
log-bin="HELONG-bin" binlog_format=ROW # 只能配置行模式， 因為 Cannal 不具備將SQL轉化成數據的能力 binlog-do-db=aicloud # 監(jiān)控 AI Cloud 項目
如果要同步多個項目：
binlog-do-db=aicloud binlog-do-db=aicloud2 binlog-do-db=aicloud3
2. 重啟MySQL服務
3. 賦值數據同步權限
CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; FLUSH PRIVILEGES;
4. 安裝并配置 Canal
下載地址：https://github.com/alibaba/canal/releases
① 修改canal.properties
canal.serverMode=kafka canal.mq.servers=127.0.0.1:9092
canal 監(jiān)控 binlog 日志，binlog 日志的傳輸默認使用 MySQL 的復制協議（基于 TCP/IP），
可以使用寫代碼的方式直接從 MySQL 服務器讀取數據，此處使用本地 kafka 進行存儲。
② 修改instance.properties
canal.instance.mysql.slaveId=100 # 大于 1 即可 canal.instance.master.address=127.0.0.1:3306 canal.mq.topic=ai-cloud-canal-to-kafka
slaveId 表示從節(jié)點 id，canal 的執(zhí)行原理就是偽裝成一個從庫去主庫同步數據
（主節(jié)點的 slaveId = 1）
address 配置連接本地的 MySQL
topic 配置數據發(fā)送到 Kafka 的某個主題下
5. 拷貝 Jar 包到 lib
將 canal 下 plugin 下的所有 jar 包拷貝到 lib 目錄下。
6. 刪除 bin 目錄下 startup.bat 里的參數
如果啟動時報錯：
Unrecognized VM option 'PermSize=128m'
Error: Could not create the Java Virtual Machine.
Error: A fatal exception has occurred. Program will exit.
刪除 -XX:PermSize=128m 參數即可。
7. 啟動 canal
打開 cmd ，cd 到 bin 目錄下，輸入 startup.bat 回車
2. 將需要緩存的數據存儲 Redis
此時我將這個查詢列表接口的數據，存儲在 Redis 中：
/**
* 獲取歷史聊天記錄(對話/繪圖)
*
* @param type
* @return {@link ResponseEntity }
*/
@RequestMapping("/list")
public ResponseEntity getHistoryList(Integer type, Integer model) {
String listCacheKey = RedisUtil.getListCacheKey(SecurityUtil.getCurrentUser().getUid(), model, type);
Object list = redisTemplate.opsForValue().get(listCacheKey);
if (ObjectUtil.isNull(list)) {
LambdaQueryWrapper<Answer> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(Answer::getUid, SecurityUtil.getCurrentUser().getUid());
queryWrapper.eq(Answer::getType, type);
queryWrapper.eq(Answer::getModel, model);
queryWrapper.orderByDesc(Answer::getAid);
List<Answer> answerList = answerService.list(queryWrapper);
List<Long> userIds = answerList.stream().map(Answer::getUid).collect(Collectors.toList());
Map<Long, User> userIdMap = userService.selectByIds(userIds).stream().collect(Collectors.toMap(User::getUid, Function.identity()));
List<AnswerVo> answerVoList = answerList.stream().map(answer -> AnswerVoUtil.getListAnswerVo(answer, userIdMap)).collect(Collectors.toList());
// 緩存 1 天
redisTemplate.opsForValue().set(listCacheKey, answerVoList, 1, TimeUnit.DAYS);
return ResponseEntity.success(answerVoList);
} else {
return ResponseEntity.success(list);
}
}/**
* 查詢列表存儲 Redis 緩存
*
* @param uid
* @param model
* @param type
* @return {@link String }
*/
public static String getListCacheKey(Long uid, Integer model, Integer type) {
return "LIST_CACHE_KEY_" + uid + "_" + model + "_" + type;
}3. 監(jiān)聽 Kafka Topic 中數據并刪除 Redis 緩存
首先對數據庫中需要緩存的數據進行一些修改操作：
此時，使用 kafka ui（下載地址劃到最底下），刷新 kafka 對應 topic 下的 message，就可以看到當前所作出的修改：
執(zhí)行修改操作：將 “如何學習Spring？？？”修改成 “如何學習Spring？？”
執(zhí)行刪除操作：
由此可見，對數據庫的每一個修改操作，都是對應固定格式的一個數據，所以可以監(jiān)聽對應的 topic 并針對 data 中的數據進行一個提取，得到一個 cacheKey，然后刪除對應的緩存，使得下一次的查詢去訪問數據庫，并同步緩存。
【代碼示例】
/**
* canal 監(jiān)控 binlog 日志，將修改的數據存儲 kafka topic 中
* 監(jiān)聽 kafka topic 中的數據
*
* @param data
* @param ack
* @throws JsonProcessingException
*/
@KafkaListener(topics = {KafkaConstant.CANAL_TOPIC})
public void canalListen(String data, Acknowledgment ack) throws JsonProcessingException {
HashMap<String, Object> map = objectMapper.readValue(data, HashMap.class);
if (map.isEmpty()) {
ack.acknowledge();
return;
}
// 匹配上對應的數據庫和數據表
if (KafkaConstant.TARGET_DATABASE.equals(map.get(KafkaConstant.DATABASE_KEY).toString()) &&
KafkaConstant.TARGET_TABLE.equals(map.get(KafkaConstant.TABLE_KEY).toString())) {
// 更新緩存
List<Map<String, Object>> list = (List<Map<String, Object>>) map.get(KafkaConstant.DATA_KEY);
if (!CollectionUtils.isEmpty(list)) {
for (Map<String, Object> answerMap : list) {
String answerListCacheKey = RedisUtil.getListCacheKey(
Long.valueOf(answerMap.get("uid").toString()),
Integer.parseInt(answerMap.get("model").toString()),
Integer.parseInt(answerMap.get("type").toString()));
// 刪除緩存，讓下一次查詢走數據庫，并同步緩存
redisTemplate.delete(answerListCacheKey);
}
}
}
// 手動確認應答
ack.acknowledge();
}/** * canal 同步數據到 kafka */ public static final String CANAL_TOPIC = "ai-cloud-canal-to-kafka"; /** * 數據庫，緩存數據一致性的 */ public static final String DATABASE_KEY = "database"; public static final String TABLE_KEY = "table"; public static final String DATA_KEY = "data"; public static final String TARGET_DATABASE = "aicloud"; public static final String TARGET_TABLE = "answer";
【補充】
kafka ui 下載地址：??????https://github.com/provectus/kafka-ui/tags
修改配置
kafka:
clusters:
- name: kafka3_cluster
bootstrapServers: 127.0.0.1:9092以上就是使用Canal和Kafka解決MySQL與緩存的數據一致性問題的詳細內容，更多關于MySQL與緩存的數據一致性的資料請關注腳本之家其它相關文章！
相關文章
mysql中insert與select的嵌套使用解決組合字段插入問題
本節(jié)主要介紹了mysql中insert與select的嵌套使用解決組合字段插入問題,需要的朋友可以參考下2014-07-07
MySQL8.0.28安裝教程詳細圖解(windows?64位)
如果電腦上已經有MySQL數據庫再進行重做往往會遇到問題,下面這篇文章主要給大家介紹了關于windows?64位系統下MySQL8.0.28安裝教程的詳細教程,文章通過圖文介紹的非常詳細,需要的朋友可以參考下2023-04-04
UPDATE JOIN是MySQL中一種結合UPDATE語句和JOIN操作的技術,本文主要介紹了MySQL中UPDATE JOIN語句的使用詳細,具有一定的參考價值,感興趣的可以了解一下2024-04-04
今天小編就為大家分享一篇關于MySQL不同表之前的字段復制，小編覺得內容挺不錯的，現在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-03-03
mysql中InnoDB默認的事務隔離級別為可重復讀(Repeated Read, RR),我們當下的所有介紹都是基于這個隔離級別為前提的,記錄鎖鎖定索引關聯的具體記錄,間隙鎖鎖定間隔,防止間隔中被其他事務插入,臨鍵鎖鎖定索引記錄+間隔,防止幻讀2023-12-12
大家好，本篇文章主要講的是mysql數據庫密碼忘記解決方法，感興趣的同學趕快來看一看吧，對你有幫助的話記得收藏一下，方便下次瀏覽2021-12-12
有時候,我們需要遷移整個數據庫,包括數據文件,本文將介紹如何通過復制MySQL的data文件來完成數據庫遷移,具有一定的參考價值,感興趣的可以了解一下2023-11-11
這篇文章主要介紹關于MySQL 大批量插入時如何過濾重復數據，比如線上庫有6個表存在重復數據，其中2個表比較大，96萬+和30萬+，因為之前處理過相同的問題，就直接拿來了上次的Python去重腳本，腳本很簡單，就是連接數據庫，查出來重復數據，循環(huán)刪除,需要的朋友可以參考下2021-09-09
mysql和oracle默認排序的方法 - 不指定order by
這篇文章主要介紹了mysql和oracle默認排序的方法 - 不指定order by。具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-07-07
這篇文章主要介紹了Mysql的增刪改查語句簡單實現的相關資料,需要的朋友可以參考下2017-04-04