devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：菜鸟追梦旅行 | 2024-02-29 10:54:58

赞

踩

（高阶）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<Column> 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<Column> 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<Column> 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<Entry> 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("================&gt; 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.     <properties>
  2.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  3.         <maven.compiler.source>1.8</maven.compiler.source>
  4.         <maven.compiler.target>1.8</maven.compiler.target>
  5.         <junit.version>4.12</junit.version>
  6.         <log4j.version>1.2.17</log4j.version>
  7.         <lombok.version>1.16.18</lombok.version>
  8.         <mysql.version>5.1.47</mysql.version>
  9.         <druid.version>1.1.16</druid.version>
  10.         <mapper.version>4.1.5</mapper.version>
  11.         <mybatis.spring.boot.version>1.3.0</mybatis.spring.boot.version>
  12.     </properties>
  13.     <dependencies>
  14.         <dependency>
  15.             <groupId>redis.clients</groupId>
  16.             <artifactId>jedis</artifactId>
  17.             <version>4.3.1</version>
  18.         </dependency>
  19.         <!--canal-->
  20.         <dependency>
  21.             <groupId>com.alibaba.otter</groupId>
  22.             <artifactId>canal.client</artifactId>
  23.             <version>1.1.0</version>
  24.         </dependency>
  25.         <!--SpringBoot通用依赖模块-->
  26.         <dependency>
  27.             <groupId>org.springframework.boot</groupId>
  28.             <artifactId>spring-boot-starter-web</artifactId>
  29.         </dependency>
  30.         <dependency>
  31.             <groupId>org.springframework.boot</groupId>
  32.             <artifactId>spring-boot-starter-actuator</artifactId>
  33.         </dependency>
  34.         <!--SpringBoot与Redis整合依赖-->
  35.         <dependency>
  36.             <groupId>org.springframework.boot</groupId>
  37.             <artifactId>spring-boot-starter-data-redis</artifactId>
  38.         </dependency>
  39.         <dependency>
  40.             <groupId>org.apache.commons</groupId>
  41.             <artifactId>commons-pool2</artifactId>
  42.         </dependency>
  43.         <!--SpringBoot与AOP-->
  44.         <dependency>
  45.             <groupId>org.springframework.boot</groupId>
  46.             <artifactId>spring-boot-starter-aop</artifactId>
  47.         </dependency>
  48.         <dependency>
  49.             <groupId>org.aspectj</groupId>
  50.             <artifactId>aspectjweaver</artifactId>
  51.         </dependency>
  52.         <!--Mysql数据库驱动-->
  53.         <dependency>
  54.             <groupId>mysql</groupId>
  55.             <artifactId>mysql-connector-java</artifactId>
  56.             <version>5.1.47</version>
  57.         </dependency>
  58.         <!--SpringBoot集成druid连接池-->
  59.         <dependency>
  60.             <groupId>com.alibaba</groupId>
  61.             <artifactId>druid-spring-boot-starter</artifactId>
  62.             <version>1.1.10</version>
  63.         </dependency>
  64.         <dependency>
  65.             <groupId>com.alibaba</groupId>
  66.             <artifactId>druid</artifactId>
  67.             <version>${druid.version}</version>
  68.         </dependency>
  69.         <!--mybatis和springboot整合-->
  70.         <dependency>
  71.             <groupId>org.mybatis.spring.boot</groupId>
  72.             <artifactId>mybatis-spring-boot-starter</artifactId>
  73.             <version>${mybatis.spring.boot.version}</version>
  74.         </dependency>
  75.         <!--hutool-->
  76.         <dependency>
  77.             <groupId>cn.hutool</groupId>
  78.             <artifactId>hutool-all</artifactId>
  79.             <version>5.2.3</version>
  80.         </dependency>
  81.         <dependency>
  82.             <groupId>junit</groupId>
  83.             <artifactId>junit</artifactId>
  84.             <version>${junit.version}</version>
  85.         </dependency>
  86.         <dependency>
  87.             <groupId>org.springframework.boot</groupId>
  88.             <artifactId>spring-boot-starter-test</artifactId>
  89.             <scope>test</scope>
  90.         </dependency>
  91.         <dependency>
  92.             <groupId>log4j</groupId>
  93.             <artifactId>log4j</artifactId>
  94.             <version>${log4j.version}</version>
  95.         </dependency>
  96.         <dependency>
  97.             <groupId>org.projectlombok</groupId>
  98.             <artifactId>lombok</artifactId>
  99.             <version>${lombok.version}</version>
  100.             <optional>true</optional>
  101.         </dependency>
  102.         <!--persistence-->
  103.         <dependency>
  104.             <groupId>javax.persistence</groupId>
  105.             <artifactId>persistence-api</artifactId>
  106.             <version>1.0.2</version>
  107.         </dependency>
  108.         <!--通用Mapper-->
  109.         <dependency>
  110.             <groupId>tk.mybatis</groupId>
  111.             <artifactId>mapper</artifactId>
  112.             <version>${mapper.version}</version>
  113.         </dependency>
  114.         <dependency>
  115.             <groupId>org.springframework.boot</groupId>
  116.             <artifactId>spring-boot-autoconfigure</artifactId>
  117.         </dependency>
  118.         <dependency>
  119.             <groupId>org.apache.commons</groupId>
  120.             <artifactId>commons-pool2</artifactId>
  121.             <version>2.11.1</version>
  122.         </dependency>
  123.         <dependency>
  124.             <groupId>com.baomidou</groupId>
  125.             <artifactId>mybatis-plus-boot-starter</artifactId>
  126.             <version>3.4.1</version>
  127.         </dependency>
  128.     </dependencies>

connector.subscribe过滤规则

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

  1. 声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/菜鸟追梦旅行/article/detail/165197?site
    推荐阅读
    相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。