• Java开发 - Canal进阶之和Redis的数据同步

    前言

    Canal在数据同步中是非常常见的，一般我们会用它来做MySQL和Redis之间、MySQL和ES之间的数据同步，否则就是手动通过代码进行同步，造成代码耦合度高的问题，这并不是我们愿意看见的，今天这篇博客博主将给大家演示Canal的数据同步做法，敲小黑板了啊，实战中基本也是这么做的，有需要的小伙伴可以仔细研究。

    前文回顾

    前文中，我们讲解了Canal的基本用法，用来监听数据库的变化，对数据库和Canal的一些基础配置都在这里：
    Java开发 - Canal的基本用法

    如果你对Canal还不了解，那么你有必要先去看这篇博客，再来学习这里的实战内容。 这很有必要，当然如果你对MySQL主从也不了解的话，那么你也有必要去看上一篇博客，里面有提到MySQL主从的博客地址，也可以前往博主的博客首页去向下翻找。同时，还有Redis主从，Redis集群的博客，你想知道的，博主都有，自行去查找。

    基础架构

    所以，本文中我们要使用到的技术就是以上几种了，而MySQL前文中已经配置过，此处不再重复，可自行前往查看。Canal虽然也配置过，但要做到和Kafka配合还需要额外的配置，而Kafka的使用必然是脱离不开Zookeeper的，这个自不必多说，最后就是同步了，还有点小期待呢，下面我们就一起动动小手开始进行今天的学习吧。 

    Canal配置

    Canal的下载使用前文中有部分，希望大家在看此篇博客前先去查看上一篇，在上一篇中，我们仅对conf/example/instance.properties文件进行了配置，在配合MQ的时候，还需要对conf/canal.properties文件进行配置，不仅如此，instance.properties中也要补充一些配置。

    下面，我们先来配置canal.properties这个文件：

    1. # tcp, kafka, RocketMQ 这里选择kafka模式
    2. canal.serverMode = kafka
    3. # 解析器的线程数，默认是注释的，不打开可能会出现阻塞或者不进行解析的情况
    4. canal.instance.parser.parallelThreadSize = 16
    5. # MQ的服务地址，这里是kafka对应的地址和端口
    6. canal.mq.servers = 127.0.0.1:9092
    7. # 在conf目录下要有example同名的目录，可以配置多个，这也是上一篇中配置的文件生效的关键
    8. canal.destinations = example

    配置instance.properties文件，再原先配置的基础上增加MQ相关的配置：

    1. #MQ队列名称
    2. canal.mq.topic=canal_topic
    3. #单队列模式的分区下标
    4. canal.mq.partition=0

     此时，名为canal_topic的队列我们还没有创建，等下开启Kafka之后，我们就创建。

    Zookeeper、Kafka、Redis、Canal启动

    Zookeeper&Kafka启动

    ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。

    要想启动Kafka，就要先启动Zookeeper，电脑终端进入Kafka的bin文件目录下：

    1. # 进入Kafka文件夹
    2. cd Desktop/JAVATOOL/kafka/kafka_2.13-2.4.1/bin
    3. # 启动Zookeeper服务
    4. ./zookeeper-server-start.sh ../config/zookeeper.properties 
    5. # 启动Kafka服务
    6. ./kafka-server-start.sh ../config/server.properties

    然后你就看到了一堆的输出......

    如果需要关闭，命令如下：

    使用命令进行监听消费：

    需要先关闭Kafka，再关闭zookeeper，否则Kafka会一直报断开连接的错。 

    文件路径写你自己的，不要写博主的，两个服务启动用两个窗口来运行命令，运行期间窗口不要关闭。

    Windows电脑需要进入到kafka/bin/windows目录下，接着输入以下命令：

    启动zookeeper：

    zookeeper-server-start.bat ..\..\config\zookeeper.properties

    启动kafka： 

    kafka-server-start.bat ..\..\config\server.properties

     我们需要先把前面配置Kafka未创建的队列先创建一下，首先进入Kafka的bin目录下，然后执行以下命令：

    ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic canal_topic

    Redis启动

    找到Redis所在的目录，将redis-server直接拖到终端即可启动：

    进入Redis安装目录：

    cd /Users/Codeliu/Desktop/JAVATOOL/redis/src

    执行以下命令：

    redis-server redis.conf

      

    通过命令可以查看Redis运行状态：

    可以看到Redis已经在6379端口运行。 

    Canal启动

    和Redis一样，将canal下bin文件夹内的startup.sh文件拖动到终端即可启动：

    /Users/Codeliu/Desktop/JAVATOOL/canal/bin/startup.sh

    前文中我们也多有演示，要想停止，只需要把stop.sh拖进去回车就可以了。 

    简单测试一下

    现在，我们依赖的服务都已启动，我们简单通过命令来测试一下，使用命令进行监听消费：

    ./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --from-beginning --topic canal_topic

    我们向MySQL插入数据：

    insert into user value(null ,'小花','123456',20,'13812345678');

     看看控制台会有什么输出：

    哇，这就是我们刚刚插入的数据，到此，起码表明我们的配置都是成功的。

    注意：因为是本地配置，一个服务一个控制台，不要共用，看看博主的控制台：

    虽然乱了些，但是生产环境......好吧，博主也不敢保证生产环境能好好的，主要还是看你用多少台服务器，肯定比本地强。 

    项目阶段

    下面，我们进入项目阶段，在上一篇的项目基础上继续搞。

    添加依赖

    主要是Kafka和Redis的依赖：

    1. <dependency>
    2.     <groupId>org.springframework.kafka</groupId>
    3.     <artifactId>spring-kafka</artifactId>
    4. </dependency>
    5. <dependency>
    6.     <groupId>org.springframework.boot</groupId>
    7.     <artifactId>spring-boot-starter-data-redis</artifactId>
    8. </dependency>

    版本号大家要选择对应服务的版本号，博主这里不再给，以免误导大家。 

    添加配置

    1. spring:
    2.   redis:
    3.     host: 127.0.0.1
    4.     port: 6379
    5.     database: 0
    6.   kafka:
    7.     # Kafka服务地址
    8.     bootstrap-servers: 127.0.0.1:9092
    9.     consumer:
    10.       # 指定一个默认的组名
    11.       group-id: consumer-group1
    12.       #序列化反序列化
    13.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    14.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    15.     producer:
    16.       key-serializer: org.apache.kafka.common.serialization.StringDeserializer
    17.       value-serializer: org.apache.kafka.common.serialization.StringDeserializer
    18.       # 批量抓取
    19.       batch-size: 65536
    20.       # 缓存容量
    21.       buffer-memory: 524288
    22.  
    23. canal:
    24.   serverAddress: 127.0.0.1
    25.   serverPort: 11111
    26.   instance: #多个instance
    27.     - example

    创建Redis工具类

    1. package com.codingfire.canal.Client;
    2.  
    3. import org.springframework.data.redis.core.StringRedisTemplate;
    4. import org.springframework.stereotype.Component;
    5.  
    6. import javax.annotation.Resource;
    7. import java.util.concurrent.TimeUnit;
    8.  
    9. @Component
    10. public class RedisClient {
    11.  
    12.     /**
    13.      * 获取redis模版
    14.      */
    15.     @Resource
    16.     private StringRedisTemplate stringRedisTemplate;
    17.  
    18.     /**
    19.      * 设置redis的key-value
    20.      */
    21.     public void setString(String key, String value) {
    22.         setString(key, value, null);
    23.     }
    24.  
    25.     /**
    26.      * 设置redis的key-value，带过期时间
    27.      */
    28.     public void setString(String key, String value, Long timeOut) {
    29.         stringRedisTemplate.opsForValue().set(key, value);
    30.         if (timeOut != null) {
    31.             stringRedisTemplate.expire(key, timeOut, TimeUnit.SECONDS);
    32.         }
    33.     }
    34.  
    35.     /**
    36.      * 获取redis中key对应的值
    37.      */
    38.     public String getString(String key) {
    39.         return stringRedisTemplate.opsForValue().get(key);
    40.     }
    41.  
    42.     /**
    43.      * 删除redis中key对应的值
    44.      */
    45.     public Boolean deleteKey(String key) {
    46.         return stringRedisTemplate.delete(key);
    47.     }
    48. }

    这是一个比较通用的redis的存储工具类，很多佬都用这个类，没啥特色，直接贴就行。 

    创建数据模型

    创建Canal抓取数据的实体类：

    1. package com.codingfire.canal.entity;
    2.  
    3. import java.util.List;
    4.  
    5. public class CanalEntity {
    6.     //数据
    7.     private List<UserType> data;
    8.     //数据库名称
    9.     private String database;
    10.     private long es;
    11.     //递增，从1开始
    12.     private int id;
    13.     //是否是DDL语句
    14.     private boolean isDdl;
    15.     //表结构的字段类型
    16.     private SqlType sqlType;
    17.     //UPDATE语句，旧数据
    18.     private String old;
    19.     //主键名称
    20.     private List<String> pkNames;
    21.     //sql语句
    22.     private String sql;
    23.     private UserType userType;
    24.     //表名
    25.     private String table;
    26.     private long ts;
    27.     //(新增)INSERT、(更新)UPDATE、(删除)DELETE、(删除表)ERASE等等
    28.     private String type;
    29.  
    30. }

    创建用户信息表实体类：

    1. package com.codingfire.canal.entity;
    2.  
    3. public class UserType {
    4.     private Long id;
    5.     private String username;
    6.     private String password;
    7.     private int age;
    8.     private String phone;
    9. }

    创建MQ的消费者

    1. package com.codingfire.canal.consumer;
    2.  
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.codingfire.canal.Client.RedisClient;
    5. import com.codingfire.canal.entity.CanalEntity;
    6. import com.codingfire.canal.entity.UserType;
    7. import org.apache.kafka.clients.consumer.ConsumerRecord;
    8. import org.slf4j.Logger;
    9. import org.slf4j.LoggerFactory;
    10. import org.springframework.kafka.annotation.KafkaListener;
    11. import org.springframework.stereotype.Component;
    12.  
    13. import javax.annotation.Resource;
    14. import java.util.List;
    15.  
    16. @Component
    17. public class CanalConsumer {
    18.     //日志记录
    19.     private static Logger log = LoggerFactory.getLogger(CanalConsumer.class);
    20.     //redis操作工具类
    21.     @Resource
    22.     private RedisClient redisClient;
    23.     //监听的队列名称为：canal_topic
    24.     @KafkaListener(topics = "canal_topic")
    25.     public void receive(ConsumerRecord<?, ?> consumer) {
    26.         String value = (String) consumer.value();
    27.         log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}", consumer.topic(), consumer.key(),consumer.partition(), consumer.offset(), value);
    28.         //转换为javaBean
    29.         CanalEntity canalEntity = JSONObject.parseObject(value, CanalEntity.class);
    30.         //获取是否是DDL语句
    31.         boolean isDdl = canalEntity.isDdl();
    32.         //获取类型
    33.         String type = canalEntity.getType();
    34.         //不是DDL语句
    35.         if (!isDdl) {
    36.             List<UserType> userTypeList = canalEntity.getData();
    37.             //过期时间
    38.             long TIME_OUT = 180L;
    39.             if ("INSERT".equals(type)) {
    40.                 //新增语句
    41.                 for (UserType userType : userTypeList) {
    42.                     //此处id需要注意，由于是演示项目，博主用的是自增id，实际应用中可能会用复杂的id规则
    43.                     String id = userType.getId().toString();
    44.                     //新增到redis中,过期时间3分钟
    45.                     redisClient.setString(id, JSONObject.toJSONString(userType), TIME_OUT);
    46.                 }
    47.             } else if ("UPDATE".equals(type)) {
    48.                 //更新语句
    49.                 for (UserType userType : userTypeList) {
    50.                     //此处id需要注意，由于是演示项目，博主用的是自增id，实际应用中可能会用复杂的id规则
    51.                     String id = userType.getId().toString();
    52.                     //更新到redis中,过期时间是3分钟
    53.                     redisClient.setString(id, JSONObject.toJSONString(userType), TIME_OUT);
    54.                 }
    55.             } else {
    56.                 //删除语句
    57.                 for (UserType userType : userTypeList) {
    58.                     //此处id需要注意，由于是演示项目，博主用的是自增id，实际应用中可能会用复杂的id规则
    59.                     String id = userType.getId().toString();
    60.                     //从redis中删除
    61.                     redisClient.deleteKey(id);
    62.                 }
    63.             }
    64.         }
    65.     }
    66. }
    67.

    测试

    现在，运行项目:

    图中队列名字可不要写错了， 运行成功后，我们向用户表中插入新的数据：

    insert into user value(null ,'小黑','123456',20,'13812345678');

    控制台监听如下：

    项目控制台监听如下：

    Redis客户端数据如下：

    至此，我们的整个应用就是成功的了，鼓掌👏 👏👏👏👏👏👏👏👏👏👏👏👏

    从整体的监听上来看，我们可以得知database，table等参数，所以在使用中，可以根据这些参数灵活的进行使用，也可以创建多个队列和多个消费者。在本案例中，就可以创建一个新的队列和新的ES消费者来对ES进行数据的同步，所以，你明白ES的数据同步该怎么做了吗？

    结语

    本篇是对单节点的应用，但从中我们也能推断出一些关于集群的应用，不过大多数公司能搞个主从就已经是不错了，所以你能学会这个已经算是很有用了，关于ES，博主前面也讲过一些基础应用，创建一个工具类利用ES的数据模型进行存储就可以达到一个canal同时对Redis和ES的同步，如果担心出问题，可以把Canal搞成集群的形式，这个后续有时间博主再给大家做讲解。今天就到这里了，觉得不错就支持一下吧。

  • 相关阅读:
    在若依Ruoyi-Vue中集成mybatisplus实现mybatis增强
    HACKTHEBOX——Bank
    6. 用Rust手把手编写一个wmproxy(代理，内网穿透等), 通讯协议源码解读篇
    报表工具应该具备哪些特性？_光点科技
    Codeforces Round 895 (Div. 3)
    大模型微调方法
    labelImg
    海量数据处理
    拿到这份Java面试文档“狂刷”3周，成功拿到京东的offer
    学习笔记-微信支付
  • 原文地址：https://blog.csdn.net/CodingFire/article/details/131474696