使用canal解决Mysql和Redis数据同步(TCP)

工作原理分析

我们在面试的时候常常听面试官问这么一个问题：你们的Mysql和Redis怎么做数据同步的，根据不同的业务场景又很多方案，你可能会说先写库再删缓存，或者延迟双删或其他方案。今天我要给大家分享的就是比较成熟的方案-使用Canal实现Mysql和Redis数据的同步。

我不知道你是否了解Mysql主从，根据2/8原则，80%的性能问题都在读上面，当我们数据库的读并发较大的时候，我们可以使用Mysql主从来分担读的压力。它的原理是所有的写操作在主库上，读操作在从库上，当然主库也可以承担读请求，而从库的数据则通过主库复制而来，Mysql自带主从复制的功能。如下图

主从复制步骤：

将Master的binary-log日志文件打开，mysql会把所有的DDL,DML,TCL写入BinaryLog日志文件中
Master会生成一个 log dump 线程，用来给从库的 i/o线程传binlog
从库的i/o线程去请求主库的binlog，并将得到的binlog日志写到中继日志（relaylog）中
从库的sql线程，会读取relaylog文件中的日志，并解析成具体操作，通过主从的操作一致，而达到最终数据一致

而Canal的原理就是伪装成Slave从Binlog中复制SQL语句或者数据。

Mysql和Redis数据同步方案

根据上面所说，我们就可以通过Canal去自动同步数据库的binlog数据日志文件，然后再把数据同步到Redis，从而达到Mysql和Redis自动同步的功能。很遗憾的是Canal没办法直接把数据库同步到Redis，它支持的是组件有： mysql、Kafka、ElasticSearch、Hbase、RocketMQ等

当然 canal 特别设计了 client-server 模式，交互协议使用 protobuf 3.0 , client 端可采用不同语言实现不同的消费逻辑

canal java 客户端: https://github.com/alibaba/canal/wiki/ClientExample
canal c# 客户端: https://github.com/dotnetcore/CanalSharp
canal go客户端: https://github.com/CanalClient/canal-go
canal Python客户端: https://github.com/haozi3156666/canal-python

canal 作为 MySQL binlog 增量获取和解析工具，可将数据通过TCP协议将数据同步到canal-client也就是我们的应用中，因此我们可以使用下面这种方案来同步数据

首选需要开启Mysql的bin-log
然后需要安装canal-server伪装成slave同步mysql中的数据
编写canal-client客户端监听canal-server，把数据从canal-server中同步过来
然后把拿到的数据写入Redis即可

开启Mysql bin-log日志

找到Mysql安装目录中的my.ini 配置文件，我以mysql 5.5为例,在 mysqld 下做如下配置

[mysqld]
#开启bInlog
log-bin=mysql-bin
#给mysql服务指定一个唯一的ID
server-id=1
#以数据的方式写binlog日志 ：statement 是记录SQL，row是记录数据
binlog-format=ROW
#同步的数据库名
#binlog-do-db=canaldb
#忽略的表
binlog-ignore-db=mysql
# 启动mysql时不启动grant-tables授权表
skip-grant-tables

修改好之后，重启Mysql服务。注意：我这里指定了需要同步的数据库为canaldb，所以需要创建一个数据库，同时创建了一个employee表作为演示

然后创建一个用户提供给canal来链接Mysql做数据同步

flush privileges;
#创建用户cannal
CREATE USER canal IDENTIFIED BY 'canal';
#把所有权限赋予canal，密码也是canal
GRANT ALL PRIVILEGES ON canaldb.user TO 'canal'@'%' identified by "canal";
//GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%' identified by "canal";
#刷新权限
flush privileges;

到这，Mysql部分就搞定了

安装Canal

去官网下载 Canal : https://github.com/alibaba/canal/releases ，我使用的是canal.deployer-1.1.5.tar.gz版本

下载好之后解压，目录结构如下

接下来修改instance 配置文件： conf/example/instance.properties

#  按需修改成自己的数据库信息
#################################################
...
#我的端口是3307
canal.instance.master.address=192.168.1.20:3307
# username/password,数据库的用户名和密码
...
#刚才开通的mysql的账户密码
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
...
# 同步的表的规则
# table regex
# 同步所有表
#canal.instance.filter.regex=.*\\..*
# 同步多个表，用逗号隔开
canal.instance.filter.regex=canaldb.employee,canaldb.dept
#################################################...省略...

这里注意如下几个东西，其他的不用管

master.address ：Mysql的地址，我的端口是3307，默认是3306
dbUsername ：上面开通的Mysql用户
dbPassword ：密码
ccanal.instance.filter.regex ：要同步的表，多个表用逗号隔开

接着修改canal 配置文件 conf/canal.properties

# ...
# 可选项: tcp(默认), kafka, RocketMQ
# 这里使用tcp , 还支持kafka和rocketmq
canal.serverMode = tcp
...省略...

这里需要注意 : canal.serverMode = tcp：我这里以tcp为例,指的是以tcp协议把数据同步数据，而不是同步到mq

配置好之后，找到 canal 安装目录下 bin目录下的 startup.bat 双击启动，linux上启动：startup.sh

编写canal-client

接下来我们需要在项目中整合canal-client来同步canal-server中的数据，然后写入Redis

第一步：导入如下依赖，我这里使用了 canal-spring-boot-starter 来整合canal-client

org.springframework.bootspring-boot-starter-parent2.2.5.RELEASEorg.projectlomboklomboktrueorg.springframework.bootspring-boot-starter-weborg.springframework.bootspring-boot-starter-testorg.springframework.bootspring-boot-starter-data-redistop.javatoolcanal-spring-boot-starter1.2.1-RELEASEcom.alibabafastjson1.2.50

第二步：在yaml配置canal地址，以及Redis相关参数

canal:server: 127.0.0.1:11111 #canal的地址destination: example #默认的数据同步的目的地
spring:redis:host: 127.0.0.1password: 123456

编写启动类

@SpringBootApplication
public class CanalApplication {public static void main(String[] args) {SpringApplication.run(CanalApplication.class,args);}
}

第三步：对Redis做配置，实现自动序列化

//缓存的配置
@Configuration
public class RedisConfig {@Resourceprivate RedisConnectionFactory factory;//使用JSON进行序列化@Beanpublic RedisTemplate redisTemplate() {RedisTemplate redisTemplate = new RedisTemplate<>();redisTemplate.setConnectionFactory(factory);//JSON格式序列化GenericFastJsonRedisSerializer serializer = new GenericFastJsonRedisSerializer();//key的序列化redisTemplate.setKeySerializer(serializer);//value的序列化redisTemplate.setValueSerializer(serializer);//hash结构key的虚拟化redisTemplate.setHashKeySerializer(new StringRedisSerializer());//hash结构value的虚拟化redisTemplate.setHashValueSerializer(serializer);return redisTemplate;}}

第四步：编写实体类，对应要同步的数据库的表

@Data
public class Employee {private Long id;private String username;
}

第五步：编写数据同步处理器，canal-client提供了EntryHandler，该handler中提供了insert,delete,update方法，当监听到某张表的相关操作后，会回调对应的方法把数据传递进来，我们就可以拿到数据往Redis同步了。

@CanalTable("employee")
@Component
@Slf4j
public class EmployeeHandler implements EntryHandler {//把数据往Redis同步@Autowiredprivate RedisTemplate redisTemplate;@Overridepublic void insert(Employee employee) {redisTemplate.opsForValue().set("EMP:"+employee.getId(),employee);}@Overridepublic void delete(Employee employee) {redisTemplate.delete("EMP:"+employee.getId());}@Overridepublic void update(Employee before, Employee after) {redisTemplate.opsForValue().set("EMP:"+after.getId(),after);}
}