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

柔情只为你懂 2024-05-31 04:28 130阅读 0赞

## 前言 ##

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

## 前文回顾 ##

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

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

## 基础架构 ##

![3ccaa04b5b3f4566b6d84292ebc166f0.png][]

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

## Canal配置 ##

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

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

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

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

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

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

## Zookeeper、Kafka、Redis、Canal启动 ##

### Zookeeper&Kafka启动 ###

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

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

# 进入Kafka文件夹
    cd Desktop/JAVATOOL/kafka/kafka_2.13-2.4.1/bin
    # 启动Zookeeper服务
    ./zookeeper-server-start.sh ../config/zookeeper.properties 
    # 启动Kafka服务
    ./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

![90e5563c490a4d4fad0729e82a292842.png][]

### Redis启动 ###

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

进入Redis安装目录：

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

执行以下命令：

redis-server redis.conf

![878b5c3f8d4b452691acc70a95965c6d.png][]

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

![d1ab250ca2c742d6afd8796d6b4de934.png][]

可以看到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');

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

![0b69653480fc42d0833a2ae689de72e5.png][]

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

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

![c123e72e8bc04f20b41c6625598d34fc.png][]

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

## 项目阶段 ##

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

## 添加依赖 ##

主要是Kafka和Redis的依赖：

<dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

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

### 添加配置 ###

spring:
      redis:
        host: 127.0.0.1
        port: 6379
        database: 0
      kafka:
        # Kafka服务地址
        bootstrap-servers: 127.0.0.1:9092
        consumer:
          # 指定一个默认的组名
          group-id: consumer-group1
          #序列化反序列化
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        producer:
          key-serializer: org.apache.kafka.common.serialization.StringDeserializer
          value-serializer: org.apache.kafka.common.serialization.StringDeserializer
          # 批量抓取
          batch-size: 65536
          # 缓存容量
          buffer-memory: 524288
    
    canal:
      serverAddress: 127.0.0.1
      serverPort: 11111
      instance: #多个instance
        - example

### 创建Redis工具类 ###

package com.codingfire.canal.Client;
    
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.concurrent.TimeUnit;
    
    @Component
    public class RedisClient {
    
        /**
         * 获取redis模版
         */
        @Resource
        private StringRedisTemplate stringRedisTemplate;
    
        /**
         * 设置redis的key-value
         */
        public void setString(String key, String value) {
            setString(key, value, null);
        }
    
        /**
         * 设置redis的key-value，带过期时间
         */
        public void setString(String key, String value, Long timeOut) {
            stringRedisTemplate.opsForValue().set(key, value);
            if (timeOut != null) {
                stringRedisTemplate.expire(key, timeOut, TimeUnit.SECONDS);
            }
        }
    
        /**
         * 获取redis中key对应的值
         */
        public String getString(String key) {
            return stringRedisTemplate.opsForValue().get(key);
        }
    
        /**
         * 删除redis中key对应的值
         */
        public Boolean deleteKey(String key) {
            return stringRedisTemplate.delete(key);
        }
    }

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

### 创建数据模型 ###

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

package com.codingfire.canal.entity;
    
    import java.util.List;
    
    public class CanalEntity {
        //数据
        private List<UserType> data;
        //数据库名称
        private String database;
        private long es;
        //递增，从1开始
        private int id;
        //是否是DDL语句
        private boolean isDdl;
        //表结构的字段类型
        private SqlType sqlType;
        //UPDATE语句，旧数据
        private String old;
        //主键名称
        private List<String> pkNames;
        //sql语句
        private String sql;
        private UserType userType;
        //表名
        private String table;
        private long ts;
        //(新增)INSERT、(更新)UPDATE、(删除)DELETE、(删除表)ERASE等等
        private String type;
    
    }

创建用户信息表实体类：

package com.codingfire.canal.entity;
    
    public class UserType {
        private Long id;
        private String username;
        private String password;
        private int age;
        private String phone;
    }

### 创建MQ的消费者 ###

package com.codingfire.canal.consumer;
    
    import com.alibaba.fastjson.JSONObject;
    import com.codingfire.canal.Client.RedisClient;
    import com.codingfire.canal.entity.CanalEntity;
    import com.codingfire.canal.entity.UserType;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.Resource;
    import java.util.List;
    
    @Component
    public class CanalConsumer {
        //日志记录
        private static Logger log = LoggerFactory.getLogger(CanalConsumer.class);
        //redis操作工具类
        @Resource
        private RedisClient redisClient;
        //监听的队列名称为：canal_topic
        @KafkaListener(topics = "canal_topic")
        public void receive(ConsumerRecord<?, ?> consumer) {
            String value = (String) consumer.value();
            log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}", consumer.topic(), consumer.key(),consumer.partition(), consumer.offset(), value);
            //转换为javaBean
            CanalEntity canalEntity = JSONObject.parseObject(value, CanalEntity.class);
            //获取是否是DDL语句
            boolean isDdl = canalEntity.isDdl();
            //获取类型
            String type = canalEntity.getType();
            //不是DDL语句
            if (!isDdl) {
                List<UserType> userTypeList = canalEntity.getData();
                //过期时间
                long TIME_OUT = 180L;
                if ("INSERT".equals(type)) {
                    //新增语句
                    for (UserType userType : userTypeList) {
                        //此处id需要注意，由于是演示项目，博主用的是自增id，实际应用中可能会用复杂的id规则
                        String id = userType.getId().toString();
                        //新增到redis中,过期时间3分钟
                        redisClient.setString(id, JSONObject.toJSONString(userType), TIME_OUT);
                    }
                } else if ("UPDATE".equals(type)) {
                    //更新语句
                    for (UserType userType : userTypeList) {
                        //此处id需要注意，由于是演示项目，博主用的是自增id，实际应用中可能会用复杂的id规则
                        String id = userType.getId().toString();
                        //更新到redis中,过期时间是3分钟
                        redisClient.setString(id, JSONObject.toJSONString(userType), TIME_OUT);
                    }
                } else {
                    //删除语句
                    for (UserType userType : userTypeList) {
                        //此处id需要注意，由于是演示项目，博主用的是自增id，实际应用中可能会用复杂的id规则
                        String id = userType.getId().toString();
                        //从redis中删除
                        redisClient.deleteKey(id);
                    }
                }
            }
        }
    }

### 测试 ###

现在，运行项目:

![63c385d7dffa4dbe9aeaea648de0805a.png][]

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

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

控制台监听如下：

![1ace10e29efb457cba0136114dbcb6c3.png][]

项目控制台监听如下：

![120c3e5206004798af57398821472a33.png][]

Redis客户端数据如下：

![181fbf50eacd4b3881ea8b28e5e70c29.png][]

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

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

## 结语 ##

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

[Java_ - Canal]: https://blog.csdn.net/CodingFire/article/details/131420343?spm=1001.2014.3001.5501
[3ccaa04b5b3f4566b6d84292ebc166f0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/555cfcc5f3914083992e94cff6052ba0.png
[Link 1]: https://baike.baidu.com/item/%E5%BC%80%E6%94%BE%E6%BA%90%E7%A0%81/7176422?fromModule=lemma_inlink
[Link 2]: https://baike.baidu.com/item/%E5%88%86%E5%B8%83%E5%BC%8F%E5%BA%94%E7%94%A8%E7%A8%8B%E5%BA%8F/9854429?fromModule=lemma_inlink
[Google]: https://baike.baidu.com/item/Google/86964?fromModule=lemma_inlink
[Hadoop]: https://baike.baidu.com/item/Hadoop/3526507?fromModule=lemma_inlink
[Hbase]: https://baike.baidu.com/item/Hbase/7670213?fromModule=lemma_inlink
[Link 3]: https://baike.baidu.com/item/%E5%9F%9F%E5%90%8D%E6%9C%8D%E5%8A%A1/3352191?fromModule=lemma_inlink
[90e5563c490a4d4fad0729e82a292842.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/427a7a3c9f9a482a9177460ffdb68e81.png
[878b5c3f8d4b452691acc70a95965c6d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/323faec4feb648828825372e7513dd0b.png
[d1ab250ca2c742d6afd8796d6b4de934.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/604fe4eff1ba4e119ed3e55c6b845e1b.png
[0b69653480fc42d0833a2ae689de72e5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/bde1db7a3b91410faa3b461e55c97f50.png
[c123e72e8bc04f20b41c6625598d34fc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/ea5a669bc36542718014944b8901f246.png
[63c385d7dffa4dbe9aeaea648de0805a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/f4c5887c32e64d778669262e79a20742.png
[1ace10e29efb457cba0136114dbcb6c3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/fd306e442e914bde949fae71ef44e908.png
[120c3e5206004798af57398821472a33.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/b4596bed74b249ef853cf15b86650289.png
[181fbf50eacd4b3881ea8b28e5e70c29.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/31/615af7a2ec7a46f1ae163e2be19b10a6.png