使用 Canal 同步 Mysql 与 Redis 的数据
一、Canal 简介
要保证数据库和 redis 强一致性是不可能的，肯定有少许时间的不一致。canal 是阿里的一套组件，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。也就是用来监听 mysql master 发送的类似 binlog 的数据，然后让消息费去消费。其原理如下：
- canal 模拟 mysql slave 的交互协议，伪装自己为 mysql slave，向 mysql master 发送 dump 协议
- mysql master 收到 dump 请求，开始推送 binary log 给 slave (也就是canal)
- canal 解析 binary log 对象(原始为byte流)
二、MySql 环境搭建
1、MySql 开启 binlog
[1] 查看 mysql 是否开启了 binlog：
mysql > show variables like 'log_bin'
[2] 开启bin log
$ vim /etc/my.cnf
# binlog文件名
log-bin=mysql-bin
binlog_format=ROW
# mysql实例id 不能和 canal 的 slaveId 重复
server_id=121
# 需要同步的数据库名称
binlog-do-db=test_canal
# 忽略的数据库，建议填写
binlog-ignore-db=mysql
# 启动mysql时不启动grant-tables授权表
skip-grant-tables
##重启mysql
$ service mysqld restart
2、创建 canal 用户并授权
mysql> CREATE USER canal IDENTIFIED BY 'canal';
ERROR 1819 (HY000): Your password does not satisfy the current policy requirements
如果报错了，说明你的mysql不允许创建这么简单的用户名 canal，需要更改下策略，如下：
mysql> set global validate_password_policy=LOW;
mysql> set global validate_password_length=5;
给 canal 用户赋权限
mysql > GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
mysql > FLUSH PRIVILEGES;
3、为什么要创建canal用户？
有些用户没有REPLICATION SLAVE, REPLICATION CLIENT的权限，用这些用户连接canal时，无法获取到 binlog。这里的 canal 用户授权了全部权限，所以客户端可以从canal中获取binlog。
可使用 root 账户查看，如下 SQL 语句：
mysql > select * from mysql.user where user=’canal’ \G
若其查询结果列表中 Repl_slave_priv,Repl_client_priv 的值为 Y 则是正确的。
三、Canal 环境搭建
1、安装 canal server
canal 指的是 canal server，它会读取 mysql 的 binlog，解析后存储起来。其下载地址： https://github.com/alibaba/canal/releases
# mkdir canal
# cd canal
# wget https://github.com/alibaba/canal/releases/download/canal-1.1.5-alpha-2/canal.deployer-1.1.5-SNAPSHOT.tar.gz
# tar -zxvf canal.deployer-1.1.5-SNAPSHOT.tar.gz
# ./startup.sh
2、查看canal占用端口
# jps
7362 Jps
5322 CanalLauncher
# netstat -lnp |grep 5322
tcp 0 0 0.0.0.0:11110 0.0.0.0:* LISTEN 5322/java
tcp 0 0 0.0.0.0:11111 0.0.0.0:* LISTEN 5322/java
tcp 0 0 0.0.0.0:11112 0.0.0.0:* LISTEN 5322/java
3、配置
解压后得到四个目录bin、conf、lib、logs。基本上配置都在 canal.properties和instance.properties 两个配置文件中，canal.properties无需修改 ，instance.properties需要配置所同步的组件
instance.properties配置如下:
#################################################
## mysql serverId , v1.0.26+ will autoGen
#与mysql配置文件一致
canal.instance.mysql.slaveId=121
# enable gtid use true/false
canal.instance.gtidon=false
# position info
canal.instance.master.address=ip:3306
canal.instance.master.journal.name=
canal.instance.master.position=
canal.instance.master.timestamp=
canal.instance.master.gtid=
# rds oss binlog
canal.instance.rds.accesskey=
canal.instance.rds.secretkey=
canal.instance.rds.instanceId=
# table meta tsdb info
#暂时用不上设为false
canal.instance.tsdb.enable=false
canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:}
canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL;
#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb
canal.instance.tsdb.dbUsername=canal
canal.instance.tsdb.dbPassword=canal
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#canal.instance.standby.gtid=
# username/password
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.defaultDatabaseName=test_canal
canal.instance.connectionCharset=UTF-8
# table regex
canal.instance.filter.regex=.*\\..*
# table black regex
canal.instance.filter.black.regex=
#################################################
查看日志（logs/canal/canal.log 和logs/example/example.log）是否启动成功:
logs/canal/canal.log 日志文件中内容如下：the canal server is running now ......
logs/example/example.log 日志文件中内容如下：start successful....
四、实例
1、在 Java 项目中使用 canal 同步 Mysql 与 Redis 的数据
【1】导入 redis和canal 依赖
<!--Redis 依赖-->
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.9.0</version>
</dependency>
<!--Canal 依赖-->
<dependency>
<groupId>top.javatool</groupId>
<artifactId>canal-spring-boot-starter</artifactId>
<version>1.2.1-RELEASE</version>
</dependency>
【2】application.yml 配置设置连接
#canal配置信息
canal:
destination: example
server: 121.77.14.222:11111 #连接地址
【3】在要监听的表的实体类上加上注解
import static com.baomidou.mybatisplus.annotation.IdType.INPUT;
/**
* 演示如何加注解实体类
*
* @author common
* @date 2022/04/17
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@TableName("sks_product_info")
public class ProductInfoEntity implements Serializable {
private static final long serialVersionUID = 1L;
/**
* 1.必须加上@Id，要不然canal监控不到表
*/
@Id
@TableId()
private Long id;
/**
* 2.实体类字段名与数据库字段名不一致 ，需要加上@Column(name = "")注解
*/
@Column(name = "product_name")
private String productName;
@Column(name = "classify_id")
private Long classifyId;
private Integer state;
private Integer acura;
@Column(name = "view_image")
private String viewImage;
@Column(name = "default_spec_id")
private Long defaultSpecId;
}
【4】重写 EntryHandler<T> 接口
/**
* 1.首先加上 @CanalTable("需要监控的表名") 注解
* 2.实现 EntryHandler<需要监控的表对应的实体类> 借口
* 3.按 alt键+insert键，会得到三个方法 ，分别是 增 ，改 ，删
* 4.在这三个方法中分别写对应Redis 增，改， 删 方法 ，然后就可以了
* 5.当你用canal监控的mysql表发生增，改，删事件时，canal会自动同步Redis里面的数据
*
* @author common
* @date 2022/04/27
*/
@CanalTable("sks_product_carousel")
@Component
public class ProductCarouselHandle implements EntryHandler<ProductCarouselEntity> {
/**
* 插入
* 只要canal监控的mysql表，发生增加操作，就会执行此方法
* @param productCarouselEntity 演示实体类
*/
@Override
public void insert(ProductCarouselEntity productCarouselEntity) {
//这是我向Redis里面插入数据的方法(你们可以写自己的方法)
RedisUtil.hSetNotExist(RedisConfig.PRODUCT_CAROUSEL.getMsg(),productCarouselEntity.getId(),productCarouselEntity);
}
/**
* 更新
* 只要canal监控的mysql表，发生更新操作，就会执行此方法
* @param before 之前
* @param after 后
*/
@Override
public void update(ProductCarouselEntity before, ProductCarouselEntity after) {
//这是我向Redis里面更新数据的方法(你们可以写自己的方法)
RedisUtil.hSetNotExist(RedisConfig.PRODUCT_CAROUSEL.getMsg(),after.getId(),after);
}
/**
* 删除
* 只要canal监控的mysql表，发生删除操作，就会执行此方法
* @param productCarouselEntity 轮播图实体产品
*/
@Override
public void delete(ProductCarouselEntity productCarouselEntity) {
//这是我向Redis里面删除数据的方法(你们可以写自己的方法)
RedisUtil.hDel(RedisConfig.PRODUCT_CAROUSEL.getMsg(),productCarouselEntity.getId());
}
}
然后就可以实现 Mysql Redis 数据同步了
2、Java 代码验证 Canal 同步 Mysql 与 Redis 的数据
引入依赖同上
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
/**
* @author lgz
*/
public class RedisUtil {
private static Jedis jedis = null;
public static synchronized Jedis getJedis() {
if (jedis == null) {
jedis = new Jedis("192.168.188.128", 6379);
jedis.auth("redis1234");
}
return jedis;
}
public static boolean existKey(String key) {
return getJedis().exists(key);
}
public static void delKey(String key) {
getJedis().del(key);
}
public static String stringGet(String key) {
return getJedis().get(key);
}
public static String stringSet(String key, String value) {
return getJedis().set(key, value);
}
public static void hashSet(String key, String field, String value) {
getJedis().hset(key, field, value);
}
}
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import java.net.InetSocketAddress;
import java.util.List;
/**
* Created by zhangpan on 2019/1/4.
*/
public class CanalApplication {
public static void main(String args[]) {
CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("ip",
11111), "example", "", "");
int batchSize = 100;
try {
connector.connect();
connector.subscribe(".*\\..*");
connector.rollback();
while (true) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(batchSize);
long batchId = message.getId();
int size = message.getEntries().size();
System.out.println("batchId = " + batchId);
System.out.println("size = " + size);
if (batchId == -1 || size == 0) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
printEntry(message.getEntries());
}
// 提交确认
connector.ack(batchId);
// connector.rollback(batchId); // 处理失败, 回滚数据
}
} finally {
connector.disconnect();
}
}
private static void printEntry(List<Entry> entrys) {
for (Entry entry : entrys) {
if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
continue;
}
RowChange rowChage = null;
try {
rowChage = RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
EventType eventType = rowChage.getEventType();
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
eventType));
for (RowData rowData : rowChage.getRowDatasList()) {
if (eventType == EventType.DELETE) {
redisDelete(rowData.getBeforeColumnsList());
} else if (eventType == EventType.INSERT) {
redisInsert(rowData.getAfterColumnsList());
} else {
System.out.println("-------> before");
printColumn(rowData.getBeforeColumnsList());
System.out.println("-------> after");
redisUpdate(rowData.getAfterColumnsList());
}
}
}
}
private static void printColumn(List<Column> columns) {
for (Column column : columns) {
System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated());
}
}
private static void redisInsert(List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
RedisUtil.stringSet("user:" + columns.get(0).getValue(), json.toJSONString());
}
}
private static void redisUpdate(List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
RedisUtil.stringSet("user:" + columns.get(0).getValue(), json.toJSONString());
}
}
private static void redisDelete(List<Column> columns) {
JSONObject json = new JSONObject();
for (Column column : columns) {
json.put(column.getName(), column.getValue());
}
if (columns.size() > 0) {
RedisUtil.delKey("user:" + columns.get(0).getValue());
}
}
}
在test_canal新建user表，
CREATE TABLE `user` (
`id` int(11) NOT NULL,
`name` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
添加数据可以在控制台看到消息：
INSERT INTOuser(id,name) VALUES (1,'zhangsan');
控制台消息如下：
size = 3
================> binlog[mysql-bin.000005:5001] , name[test_canal,user] , eventType : INSERT