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

redis高级 使用canal进行mysql和redis的双写一致应用篇

作者：菜鸟追梦旅行 | 2024-06-16 18:33:08

赞

踩

redis高级 使用canal进行mysql和redis的双写一致应用篇

前言

我们昨天谈论了对应的redis和mysql进行双写一致的理论篇

我们说了五种更新策略和查看的策略

更新策略可以使用

1.先更新数据库再更新redis  (高并发可能导致数据不一致)

2.先更新redis再更新数据库  (高并发可能导致数据不一致)

上述建议加上双检加锁策略来保证mysql的负载没那么高

3.停机更新  (业务允许可以使用)

4.先删除redis再更新数据库  ----延迟双删策略(但是业务时间大概率不好估量)

5.先更新数据库再删除redis      使用canal保证数据一致

--------------------

前面我们说了canal主要是通过二进制binlog的监听来对应进行一个增量的监控

下面我们来玩一玩对应的canal

这里的canal我们可以就当做一个从机,用来做对应的同步操作

下载以及前置工作

 下载地址 : Release v1.1.6 · alibaba/canal · GitHub

对mysql进行配置,新加一个canal用户

使用以下sql即可(注:mysql5和8不同)

  1. //mysql5.7   开通用户权限
  2.  
  3. DROP USER IF EXISTS 'canal'@'%';
  4. CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';  
  5. GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' IDENTIFIED BY 'canal';  
  6. FLUSH PRIVILEGES;
  7.  
  8. SELECT * FROM mysql.user;
  9.  
  10.  
  11.

只要结果是这样就代表是ok的

下载之后我们放在linux 下的 /mycanal文件夹下

使用 tar -zxvf 进行对应的解压

之后我们需要修改redis下对应的配置文件  

我们只需要修改 /mycanal/conf/example/instance.properties

修改为对应的mysql的ip和端口即可     

然后我们需要配置对应的my.ini开启对应的binlog二进制文件

  1. log-bin=mysql-bin #开启 binlog
  2. binlog-format=ROW #选择 ROW 模式
  3. server_id=1    #配置MySQL replaction需要定义，不要和canal的 slaveId重复

然后我们重启mysql即可,查看binlog是否开启

开启即可

然后我们去对应的bin目录下开启对应的脚本

对应的stop.sh就是停止对应的应用

然后我们可以去对应日志文件查看是否启动成功

分别查看logs下面文件夹的canal.log和example.log即可

然后我们创建对应的数据库和测试数据表

  1. 先创建数据库
  2.  
  3. 然后执行以下语句
  4. CREATE TABLE `t_user` (
  5.  
  6.   `id` bigint(20) NOT NULL AUTO_INCREMENT,
  7.  
  8.   `userName` varchar(100) NOT NULL,
  9.  
  10.   PRIMARY KEY (`id`)
  11.  
  12. ) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8mb4

springboot整合

我们先创建一个项目

然后先配置对应的pom文件

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  3.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  4.     <modelVersion>4.0.0</modelVersion>
  5.  
  6.     <groupId>com.atguigu.canal</groupId>
  7.     <artifactId>canal_demo02</artifactId>
  8.     <version>1.0-SNAPSHOT</version>
  9.  
  10.     <parent>
  11.         <groupId>org.springframework.boot</groupId>
  12.         <artifactId>spring-boot-starter-parent</artifactId>
  13.         <version>2.5.14</version>
  14.         <relativePath/>
  15.     </parent>
  16.  
  17.     <properties>
  18.         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  19.         <maven.compiler.source>1.8</maven.compiler.source>
  20.         <maven.compiler.target>1.8</maven.compiler.target>
  21.         <junit.version>4.12</junit.version>
  22.         <log4j.version>1.2.17</log4j.version>
  23.         <lombok.version>1.16.18</lombok.version>
  24.         <mysql.version>5.1.47</mysql.version>
  25.         <druid.version>1.1.16</druid.version>
  26.         <mapper.version>4.1.5</mapper.version>
  27.         <mybatis.spring.boot.version>1.3.0</mybatis.spring.boot.version>
  28.     </properties>
  29.  
  30.     <dependencies>
  31.         <!--canal-->
  32.         <dependency>
  33.             <groupId>com.alibaba.otter</groupId>
  34.             <artifactId>canal.client</artifactId>
  35.             <version>1.1.0</version>
  36.         </dependency>
  37.         <!--SpringBoot通用依赖模块-->
  38.         <dependency>
  39.             <groupId>org.springframework.boot</groupId>
  40.             <artifactId>spring-boot-starter-web</artifactId>
  41.         </dependency>
  42.         <dependency>
  43.             <groupId>org.springframework.boot</groupId>
  44.             <artifactId>spring-boot-starter-actuator</artifactId>
  45.         </dependency>
  46.         <!--swagger2-->
  47.         <dependency>
  48.             <groupId>io.springfox</groupId>
  49.             <artifactId>springfox-swagger2</artifactId>
  50.             <version>2.9.2</version>
  51.         </dependency>
  52.         <dependency>
  53.             <groupId>io.springfox</groupId>
  54.             <artifactId>springfox-swagger-ui</artifactId>
  55.             <version>2.9.2</version>
  56.         </dependency>
  57.         <!--SpringBoot与Redis整合依赖-->
  58.         <dependency>
  59.             <groupId>org.springframework.boot</groupId>
  60.             <artifactId>spring-boot-starter-data-redis</artifactId>
  61.         </dependency>
  62.         <dependency>
  63.             <groupId>org.apache.commons</groupId>
  64.             <artifactId>commons-pool2</artifactId>
  65.         </dependency>
  66.         <!--SpringBoot与AOP-->
  67.         <dependency>
  68.             <groupId>org.springframework.boot</groupId>
  69.             <artifactId>spring-boot-starter-aop</artifactId>
  70.         </dependency>
  71.         <dependency>
  72.             <groupId>org.aspectj</groupId>
  73.             <artifactId>aspectjweaver</artifactId>
  74.         </dependency>
  75.         <!--Mysql数据库驱动-->
  76.         <dependency>
  77.             <groupId>mysql</groupId>
  78.             <artifactId>mysql-connector-java</artifactId>
  79.             <version>5.1.47</version>
  80.         </dependency>
  81.         <!--SpringBoot集成druid连接池-->
  82.         <dependency>
  83.             <groupId>com.alibaba</groupId>
  84.             <artifactId>druid-spring-boot-starter</artifactId>
  85.             <version>1.1.10</version>
  86.         </dependency>
  87.         <dependency>
  88.             <groupId>com.alibaba</groupId>
  89.             <artifactId>druid</artifactId>
  90.             <version>${druid.version}</version>
  91.         </dependency>
  92.         <!--mybatis和springboot整合-->
  93.         <dependency>
  94.             <groupId>org.mybatis.spring.boot</groupId>
  95.             <artifactId>mybatis-spring-boot-starter</artifactId>
  96.             <version>${mybatis.spring.boot.version}</version>
  97.         </dependency>
  98.         <!--通用基础配置junit/devtools/test/log4j/lombok/hutool-->
  99.         <!--hutool-->
  100.         <dependency>
  101.             <groupId>cn.hutool</groupId>
  102.             <artifactId>hutool-all</artifactId>
  103.             <version>5.2.3</version>
  104.         </dependency>
  105.         <dependency>
  106.             <groupId>junit</groupId>
  107.             <artifactId>junit</artifactId>
  108.             <version>${junit.version}</version>
  109.         </dependency>
  110.         <dependency>
  111.             <groupId>org.springframework.boot</groupId>
  112.             <artifactId>spring-boot-starter-test</artifactId>
  113.             <scope>test</scope>
  114.         </dependency>
  115.         <dependency>
  116.             <groupId>log4j</groupId>
  117.             <artifactId>log4j</artifactId>
  118.             <version>${log4j.version}</version>
  119.         </dependency>
  120.         <dependency>
  121.             <groupId>org.projectlombok</groupId>
  122.             <artifactId>lombok</artifactId>
  123.             <version>${lombok.version}</version>
  124.             <optional>true</optional>
  125.         </dependency>
  126.         <!--persistence-->
  127.         <dependency>
  128.             <groupId>javax.persistence</groupId>
  129.             <artifactId>persistence-api</artifactId>
  130.             <version>1.0.2</version>
  131.         </dependency>
  132.         <!--通用Mapper-->
  133.         <dependency>
  134.             <groupId>tk.mybatis</groupId>
  135.             <artifactId>mapper</artifactId>
  136.             <version>${mapper.version}</version>
  137.         </dependency>
  138.         <dependency>
  139.             <groupId>org.springframework.boot</groupId>
  140.             <artifactId>spring-boot-autoconfigure</artifactId>
  141.         </dependency>
  142.         <dependency>
  143.             <groupId>redis.clients</groupId>
  144.             <artifactId>jedis</artifactId>
  145.             <version>3.8.0</version>
  146.         </dependency>
  147.     </dependencies>
  148.  
  149.     <build>
  150.         <plugins>
  151.             <plugin>
  152.                 <groupId>org.springframework.boot</groupId>
  153.                 <artifactId>spring-boot-maven-plugin</artifactId>
  154.             </plugin>
  155.         </plugins>
  156.     </build>
  157.  
  158. </project>

然后是对应的properties文件,记得按照自己的设置修改

记得这里修改数据库的名字以及密码

  1. server.port=5555
  2.  
  3. # ========================alibaba.druid=====================
  4. spring.datasource.type=com.alibaba.druid.pool.DruidDataSource
  5. spring.datasource.driver-class-name=com.mysql.jdbc.Driver
  6. spring.datasource.url=jdbc:mysql://localhost:13306/canal_test?useUnicode=true&characterEncoding=utf-8&useSSL=false
  7. spring.datasource.username=root
  8. spring.datasource.password=abc123
  9. spring.datasource.druid.test-while-idle=false

接着主启动类这里用不到,无需配置

下面写业务类

首先先搞连接池

注意修改对应的配置密码以及redis的IP地址

  1. public class RedisUtils
  2. {
  3.     public static final String  REDIS_IP_ADDR = "192.168.188.136";
  4.     public static final String  REDIS_pwd = "abc123";
  5.     public static JedisPool jedisPool;
  6.  
  7.     static {
  8.         JedisPoolConfig jedisPoolConfig=new JedisPoolConfig();
  9.         jedisPoolConfig.setMaxTotal(20);
  10.         jedisPoolConfig.setMaxIdle(10);
  11.         jedisPool=new JedisPool(jedisPoolConfig,REDIS_IP_ADDR,6379,10000,REDIS_pwd);
  12.     }
  13.  
  14.     public static Jedis getJedis() throws Exception {
  15.         if(null!=jedisPool){
  16.             return jedisPool.getResource();
  17.         }
  18.         throw new Exception("Jedispool is not ok");
  19.     }
  20.  
  21. }

最后就是业务类了

  1. public class RedisCanalClientExample
  2. {
  3.     public static final Integer _60SECONDS = 60;
  4.     public static final String  REDIS_IP_ADDR = "192.168.111.185";
  5.  
  6.     private static void redisInsert(List<Column> columns)
  7.     {
  8.         JSONObject jsonObject = new JSONObject();
  9.         for (Column column : columns)
  10.         {
  11.             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
  12.             jsonObject.put(column.getName(),column.getValue());
  13.         }
  14.         if(columns.size() > 0)
  15.         {
  16.             try(Jedis jedis = RedisUtils.getJedis())
  17.             {
  18.                 jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
  19.             }catch (Exception e){
  20.                 e.printStackTrace();
  21.             }
  22.         }
  23.     }
  24.  
  25.  
  26.     private static void redisDelete(List<Column> columns)
  27.     {
  28.         JSONObject jsonObject = new JSONObject();
  29.         for (Column column : columns)
  30.         {
  31.             jsonObject.put(column.getName(),column.getValue());
  32.         }
  33.         if(columns.size() > 0)
  34.         {
  35.             try(Jedis jedis = RedisUtils.getJedis())
  36.             {
  37.                 jedis.del(columns.get(0).getValue());
  38.             }catch (Exception e){
  39.                 e.printStackTrace();
  40.             }
  41.         }
  42.     }
  43.  
  44.     private static void redisUpdate(List<Column> columns)
  45.     {
  46.         JSONObject jsonObject = new JSONObject();
  47.         for (Column column : columns)
  48.         {
  49.             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
  50.             jsonObject.put(column.getName(),column.getValue());
  51.         }
  52.         if(columns.size() > 0)
  53.         {
  54.             try(Jedis jedis = RedisUtils.getJedis())
  55.             {
  56.                 jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
  57.                 System.out.println("---------update after: "+jedis.get(columns.get(0).getValue()));
  58.             }catch (Exception e){
  59.                 e.printStackTrace();
  60.             }
  61.         }
  62.     }
  63.  
  64.     public static void printEntry(List<Entry> entrys) {
  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(REDIS_IP_ADDR,
  103.                 11111), "example", "", "");
  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("bigdata.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. }

此时我们修改mysql,redis就会自动做同步,原因就是我们在这里打日志的时候也进行了redis的回写操作,大家可以在业务类中很容易的发现

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

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

  

闽ICP备14008679号