前言
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这个文件：
# 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是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。
要想启动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
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的依赖：
<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);
}
}
}
}
}
测试
现在，运行项目:
图中队列名字可不要写错了， 运行成功后，我们向用户表中插入新的数据：
insert into user value(null ,'小黑','123456',20,'13812345678');控制台监听如下：
项目控制台监听如下：
Redis客户端数据如下：
至此，我们的整个应用就是成功的了，鼓掌👏 👏👏👏👏👏👏👏👏👏👏👏👏
从整体的监听上来看，我们可以得知database，table等参数，所以在使用中，可以根据这些参数灵活的进行使用，也可以创建多个队列和多个消费者。在本案例中，就可以创建一个新的队列和新的ES消费者来对ES进行数据的同步，所以，你明白ES的数据同步该怎么做了吗？
结语
本篇是对单节点的应用，但从中我们也能推断出一些关于集群的应用，不过大多数公司能搞个主从就已经是不错了，所以你能学会这个已经算是很有用了，关于ES，博主前面也讲过一些基础应用，创建一个工具类利用ES的数据模型进行存储就可以达到一个canal同时对Redis和ES的同步，如果担心出问题，可以把Canal搞成集群的形式，这个后续有时间博主再给大家做讲解。今天就到这里了，觉得不错就支持一下吧。
