• （高阶）Redis 7 第13讲 数据双写一致性 canal篇

     面试题

    问题答案
    如何保证mysql改动后,立即同步到Rediscanal

     

    简介

    https://github.com/alibaba/canal/wikiicon-default.png?t=N7T8https://github.com/alibaba/canal/wiki

     基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费 

     业务

    1. 数据库镜像
    2. 数据库实时备份
    3. 多级索引 (卖家和买家各自分库索引)
    4. search build
    5. 业务cache刷新
    6. 价格变化等重要业务消息

    下载

    官网https://github.com/alibaba/canal/releases/tag/canal-1.1.6icon-default.png?t=N7T8https://github.com/alibaba/canal/releases/tag/canal-1.1.6
    百度网盘链接：https://pan.baidu.com/s/1Hs7JieAZA_q4lmvIdJZgFw?pwd=aqi2 
    提取码：aqi2 

     Mysql 主从复制原理

    Canal原理

    1. canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送 dump 协议
    2. MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
    3. canal 解析 binary log 对象(原始为 byte 流)

     案例

    mysql 环境 

    1. # 查看mysql 版本
    2. SELECT VERSION();
    3.  
    4. # 查看当前主机的二进制日志
    5. SHOW MASTER status;
    6.  
    7. # 查看binlog 开启状态
    8.  
    9. SHOW VARIABLES LIKE 'log_bin'

     

     my.ini 配置

    1. # 在mysqld中加入一下内容
    2. [mysqld]
    3. log-bin=mysql-bin #开启 binlog
    4. binlog-format=ROW #选择 ROW 模式
    5. server_id=1 #配置MySQL replaction需要定义，不要和canal的 slaveId重复

     

     

     重启Mysql

     

    创建canal用户并授权

    1. DROP USER IF EXISTS 'canal'@'%';
    2. CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
    3.  
    4. GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
    5.  
    6. FLUSH PRIVILEGES;
    7.  
    8. SELECT * FROM mysql.`user`

    canal安装配置

    上传安装包

    解压安装包

    tar -zxvf canal.deployer-1.1.6.tar.gz

     配置文件地址

    修改配置

    启动canal

     

    启动成功

     如果出现如下错误

    Caused by: java.io.IOException: caching_sha2_password Auth failed
    com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector.negotiate(MysqlConnector.java:260)

     

    1. # 修改加密方式
    2. select host,user,plugin from mysql.user ;
    3. ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'canal';
    4.

    创建测试表

    1. CREATE TABLE `t_user` (
    2.   `id` int NOT NULL AUTO_INCREMENT,
    3.   `userName` varchar(255) NOT NULL,
    4.   PRIMARY KEY (`id`)
    5. ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

     创建Redis工具类

    1. package com.mco.utils;
    2.  
    3. import cn.hutool.core.util.RandomUtil;
    4. import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
    5. import org.slf4j.Logger;
    6. import org.slf4j.LoggerFactory;
    7. import redis.clients.jedis.GeoCoordinate;
    8. import redis.clients.jedis.Jedis;
    9. import redis.clients.jedis.JedisPool;
    10. import redis.clients.jedis.JedisPoolConfig;
    11. import redis.clients.jedis.args.GeoUnit;
    12.  
    13. import java.time.Duration;
    14. import java.util.HashMap;
    15. import java.util.Map;
    16. import java.util.Objects;
    17.  
    18. /**
    19.  * @author ：liao.wei
    20.  * @date ：2023/9/18 21:15
    21.  * @package : com.mco.utils
    22.  */
    23. public class RedisUtils {
    24. 	private static Logger logger = LoggerFactory.getLogger(JedisPoolUtil.class);
    25. 	
    26. 	public static final String REDIS_IP_ADDR = "120.77.64.190";
    27. 	public static final String REDIS_PWD = "111111";
    28. 	
    29. 	public static JedisPool jedisPool;
    30. 	
    31. 	static {
    32. 		JedisPoolConfig poolConfig = new JedisPoolConfig();
    33. 		poolConfig.setMaxIdle(8);
    34. 		poolConfig.setMinIdle(2);
    35. 		poolConfig.setMaxWait(Duration.ofSeconds(30000));
    36. 		jedisPool = new JedisPool(poolConfig, REDIS_IP_ADDR, 6379, 10000, REDIS_PWD);
    37. 	}
    38. 	
    39. 	public static Jedis getJedis() throws Exception {
    40. 		if (null != jedisPool) {
    41. 			return jedisPool.getResource();
    42. 		}
    43. 		throw new Exception("Jedispool is not ok");
    44. 	}
    45. }

    Canal 业务类

    1. public class RedisCanalClient {
    2. 	public static final Integer _60SECONDS = 60;
    3. 	public static final String  CANAL_IP_ADDR = "192.168.1.11";
    4. 	
    5. 	private static void redisInsert(List columns)
    6. 	{
    7. 		JSONObject jsonObject = new JSONObject();
    8. 		for (Column column : columns)
    9. 		{
    10. 			System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    11. 			jsonObject.put(column.getName(),column.getValue());
    12. 		}
    13. 		if(columns.size() > 0)
    14. 		{
    15. 			try(Jedis jedis = RedisUtils.getJedis())
    16. 			{
    17. 				jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
    18. 			}catch (Exception e){
    19. 				e.printStackTrace();
    20. 			}
    21. 		}
    22. 	}
    23. 	
    24. 	
    25. 	private static void redisDelete(List columns)
    26. 	{
    27. 		JSONObject jsonObject = new JSONObject();
    28. 		for (Column column : columns)
    29. 		{
    30. 			jsonObject.put(column.getName(),column.getValue());
    31. 		}
    32. 		if(columns.size() > 0)
    33. 		{
    34. 			try(Jedis jedis = RedisUtils.getJedis())
    35. 			{
    36. 				jedis.del(columns.get(0).getValue());
    37. 			}catch (Exception e){
    38. 				e.printStackTrace();
    39. 			}
    40. 		}
    41. 	}
    42. 	
    43. 	private static void redisUpdate(List columns)
    44. 	{
    45. 		JSONObject jsonObject = new JSONObject();
    46. 		for (Column column : columns)
    47. 		{
    48. 			System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
    49. 			jsonObject.put(column.getName(),column.getValue());
    50. 		}
    51. 		if(columns.size() > 0)
    52. 		{
    53. 			try(Jedis jedis = RedisUtils.getJedis())
    54. 			{
    55. 				jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
    56. 				System.out.println("---------update after: "+jedis.get(columns.get(0).getValue()));
    57. 			}catch (Exception e){
    58. 				e.printStackTrace();
    59. 			}
    60. 		}
    61. 	}
    62. 	
    63. 	public static void printEntry(List entrys)
    64. 	{
    65. 		for (Entry entry : entrys) {
    66. 			if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
    67. 				continue;
    68. 			}
    69. 			
    70. 			RowChange rowChage = null;
    71. 			try {
    72. 				//获取变更的row数据
    73. 				rowChage = RowChange.parseFrom(entry.getStoreValue());
    74. 			} catch (Exception e) {
    75. 				throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(),e);
    76. 			}
    77. 			//获取变动类型
    78. 			EventType eventType = rowChage.getEventType();
    79. 			System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
    80. 					entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
    81. 					entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));
    82. 			
    83. 			for (RowData rowData : rowChage.getRowDatasList()) {
    84. 				if (eventType == EventType.INSERT) {
    85. 					redisInsert(rowData.getAfterColumnsList());
    86. 				} else if (eventType == EventType.DELETE) {
    87. 					redisDelete(rowData.getBeforeColumnsList());
    88. 				} else {//EventType.UPDATE
    89. 					redisUpdate(rowData.getAfterColumnsList());
    90. 				}
    91. 			}
    92. 		}
    93. 	}
    94. 	
    95. 	
    96. 	public static void main(String[] args)
    97. 	{
    98. 		System.out.println("---------O(∩_∩)O哈哈~ initCanal() main方法-----------");
    99. 		
    100. 		//=================================
    101. 		// 创建链接canal服务端
    102. 		CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(CANAL_IP_ADDR,
    103. 				11111), "example", "", "");  // 这里用户名和密码如果在这写了，会覆盖canal配置文件的账号密码，如果不填从配置文件中读
    104. 		int batchSize = 1000;
    105. 		//空闲空转计数器
    106. 		int emptyCount = 0;
    107. 		System.out.println("---------------------canal init OK，开始监听mysql变化------");
    108. 		try {
    109. 			connector.connect();
    110. 			//connector.subscribe(".*\\..*");
    111. 			connector.subscribe("test.t_user");   // 设置监听哪个表
    112. 			connector.rollback();
    113. 			int totalEmptyCount = 10 * _60SECONDS;
    114. 			while (emptyCount < totalEmptyCount) {
    115. 				System.out.println("我是canal，每秒一次正在监听:"+ UUID.randomUUID().toString());
    116. 				Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
    117. 				long batchId = message.getId();
    118. 				int size = message.getEntries().size();
    119. 				if (batchId == -1 || size == 0) {
    120. 					emptyCount++;
    121. 					try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
    122. 				} else {
    123. 					//计数器重新置零
    124. 					emptyCount = 0;
    125. 					printEntry(message.getEntries());
    126. 				}
    127. 				connector.ack(batchId); // 提交确认
    128. 				// connector.rollback(batchId); // 处理失败, 回滚数据
    129. 			}
    130. 			System.out.println("已经监听了"+totalEmptyCount+"秒，无任何消息，请重启重试......");
    131. 		} finally {
    132. 			connector.disconnect();
    133. 		}
    134. 	}
    135. }

     说明：

            

    CANAL_IP_ADDR：canal 服务部署ip

    InetSocketAddress: 端口可从canal.log 中查看

    启动main方法

    在数据库中新增一条数据

     查看Canal客户端监听

     查看Redis数据

    POM引入

    1.     
    2.         UTF-8
    3.         1.8
    4.         1.8
    5.         4.12
    6.         1.2.17
    7.         1.16.18
    8.         5.1.47
    9.         1.1.16
    10.         4.1.5
    11.         1.3.0
    12.     
    13.     
    14.         
    15.             redis.clients
    16.             jedis
    17.             4.3.1
    18.         
    19.         
    20.         
    21.             com.alibaba.otter
    22.             canal.client
    23.             1.1.0
    24.         
    25.         
    26.         
    27.             org.springframework.boot
    28.             spring-boot-starter-web
    29.         
    30.         
    31.             org.springframework.boot
    32.             spring-boot-starter-actuator
    33.         
    34.         
    35.         
    36.             org.springframework.boot
    37.             spring-boot-starter-data-redis
    38.         
    39.         
    40.             org.apache.commons
    41.             commons-pool2
    42.         
    43.         
    44.         
    45.             org.springframework.boot
    46.             spring-boot-starter-aop
    47.         
    48.         
    49.             org.aspectj
    50.             aspectjweaver
    51.         
    52.         
    53.         
    54.             mysql
    55.             mysql-connector-java
    56.             5.1.47
    57.         
    58.         
    59.         
    60.             com.alibaba
    61.             druid-spring-boot-starter
    62.             1.1.10
    63.         
    64.         
    65.             com.alibaba
    66.             druid
    67.             ${druid.version}
    68.         
    69.         
    70.         
    71.             org.mybatis.spring.boot
    72.             mybatis-spring-boot-starter
    73.             ${mybatis.spring.boot.version}
    74.         
    75.         
    76.         
    77.             cn.hutool
    78.             hutool-all
    79.             5.2.3
    80.         
    81.         
    82.             junit
    83.             junit
    84.             ${junit.version}
    85.         
    86.         
    87.             org.springframework.boot
    88.             spring-boot-starter-test
    89.             test
    90.         
    91.         
    92.             log4j
    93.             log4j
    94.             ${log4j.version}
    95.         
    96.         
    97.             org.projectlombok
    98.             lombok
    99.             ${lombok.version}
    100.             true
    101.         
    102.         
    103.         
    104.             javax.persistence
    105.             persistence-api
    106.             1.0.2
    107.         
    108.         
    109.         
    110.             tk.mybatis
    111.             mapper
    112.             ${mapper.version}
    113.         
    114.         
    115.             org.springframework.boot
    116.             spring-boot-autoconfigure
    117.         
    118.         
    119.             org.apache.commons
    120.             commons-pool2
    121.             2.11.1
    122.         
    123.         
    124.             com.baomidou
    125.             mybatis-plus-boot-starter
    126.             3.4.1
    127.         
    128.

    connector.subscribe过滤规则

    源码地址icon-default.png?t=N7T8https://gitee.com/UniQue006/redis_example.git 

    1. 🌹 以上分享 Redis 数据一致性 canal应用，如有问题请指教。
    2.  
    3. 🌹🌹 如你对技术也感兴趣，欢迎交流。
    4.  
    5. 🌹🌹🌹  如有需要，请👍点赞💖收藏🐱‍🏍分享
  • 相关阅读:
    【QT】QtCreator卸载与安装（非正常状态）
    正则表达式实战：最新豆瓣top250爬虫超详细教程
    Go语言实践模式 - 函数选项模式(Functional Options Pattern)
    [autojs]用户界面GUI编程
    一,安卓aosp源码编译环境搭建
    技术与安全的交织
    常见漏洞修复方案
    SpringMVC
    竞赛 基于深度学习的视频多目标跟踪实现
    redis
  • 原文地址：https://blog.csdn.net/qq_32662595/article/details/132993624