Canal框架 Redis和MySQL数据一致性的解决方案阿里中间件Canal-Toy模板网

这篇具有很好参考价值的文章主要介绍了Canal框架 Redis和MySQL数据一致性的解决方案阿里中间件Canal。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

MySQL与Redis缓存的同步的两种方案
方案2：解析MySQL的binlog 实现，将数据库中的数据同步到Redis。数据库每一步操作都会写入binlog，通过订阅监听binlog实时感知数据变化，根据数据变化情况删除reids并添加重试机制。
引入消息队列：若Redis删除失败，则将Redis key放入消息队列，消费端监听消息队列并删除Redis直至删除成功；
如果你的项目对缓存的要求是强一致性的，那么请不要使用缓存。

Canal的工作原理

1、MySQL主从复制过程
    1)  Master  主库将改变记录，写入二进制日志（Binary Log）中;
    2)  Slave 从库向 MySQL 发送dump协议，将 Master主库的 binary log events 进行拷贝到它的中继日志(relay log) 
    3） Slave 从库读取并重做中继日志中的事件，将改变的数据同步到自己的数据库。
    
2、Canal的工作原理 (ROW级别)
    -- $ 将自身伪装成 Slave ， 假装从 Master  复制数据。
    -- $ Otter是阿里用于进行异地数据库之间的同步框架，Canal是其中一部分。
    
https://blog.csdn.net/weixin_55769150/article/details/128997455

同步效果

1）已验证：仅支持增量同步，不支持全量已有数据同步。这点，canal的初衷订位就是“阿里巴巴mysql数据库binlog的增量订阅&消费组件”。

2）已验证：由于采用了binlog机制，Mysql中的新增、更新、删除操作，对应的Elasticsearch都能实时新增、更新、删除。

3）推荐使用场景
canal适用于对于Mysql和Elasticsearch数据实时增、删、改要求高的业务场景。
实时场景要求不高的业务场景，logstash_input_jdbc也能满足。

技术整体思路

mysql binlog增量订阅消费+消息队列+增量数据更新到redis
1）读redis：热数据基本都在redis
2）写mysql:增删改都是操作mysql
3）更新redis数据：mysq的数据操作binlog，来更新到redis

mysql binlog 配置

查询

show variables like 'log_%' -- 开启 Binlog 写入功能，开启bin-log权限
SHOW VARIABLES like "%binlog_format%"; -- 配置 binlog-format 为 ROW 模式
SHOW VARIABLES like "%server_id%"; -- 主数据库的唯一值

show master status; --  查询此主库的状态
show slave status; --查询从库的状态
show binary logs; -- 查看当前所有binlog的日志存储
show binlog events in 'mysql-bin.000001'; -- 查看当前已经消费到了什么位置

binlog配置

# windows下mysql的配置文件是my.ini。一般放到安装的根目录
# Linux下MySQL的配置文件是my.cnf，一般会放在/etc/my.cnf，/etc/mysql/my.cnf

[mysqld]
# *** Replication related settings ***

#在复制方面的改进就是引进了新的复制技术：基于行的复制。　　　　　
binlog-format=ROW

#开启二进制日志功能，binlog数据位置
log-bin="mysql-binlog"

#服务端ID，用来高可用时做区分　　　
server_id=100

#二进制日志自动删除的天数。默认值为0,表示“没有自动删除”。启动时和二进制日志循环时可能删除。
#expire-logs-days=2  mysql8.x中准备作废
binlog_expire_logs_seconds=86400

# 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复
server_id=100

创建Mysql ，canal，用户

#创建用户
CREATE USER canal IDENTIFIED BY 'canal';
#赋权
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';    
#刷新
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;    
FLUSH PRIVILEGES;

注意：如果使用的是阿里云的 RDS for MySQL。已经是默认打开binlog了，并且账号默认具有 binlog dump 权限, 不需要任何权限或者 binlog设置,可以直接跳过这一步
https://blog.csdn.net/zcl111/article/details/119670138

canal 下载地址

https://github.com/alibaba/canal/releases

cd /usr/local
mkdir canal
tar -zxvf canal.deployer-1.1.6.tar.gz -C /usr/local/canal/

canal.properties

这个文件是 canal 的基本通用配置，canal 端口号默认就是 11111

vim /usr/local/canal/canal.properties

修改 instance.properties

进入example目录，修改instance.properties配置文件
vi /usr/local/canal/example/instance.properties

canal.instance.mysql.slaveId=5
canal.instance.master.address=rm-wz90xxxu7.mysql.rds.aliyuncs.com:3306
canal.instance.dbUsername=root
canal.instance.dbPassword=123456

注：不配置canal.instance.tsdb相关属性。

s

常见的匹配规则：

所有表：.* or .\…
canal schema下所有表： canal\…*
canal下的以canal打头的表：canal.canal.*
canal schema下的一张表：canal.test1
多个规则组合使用：canal\…*,mysql.test1,mysql.test2 (逗号分隔)

链接：https://blog.csdn.net/yy139926/article/details/127768446

Canal架构

server 代表一个 canal 运行实例，对应于一个 jvm
instance 对应于一个数据队列（1个 canal server 对应 n个 instance )
instance 下的子模块：
eventParser: 数据源接入，模拟 slave 协议和 master 进行交互，协议解析
eventSink: Parser 和 Store 链接器，进行数据过滤，加工，分发的工作
eventStore: 数据存储
metaManager: 增量订阅 & 消费信息管理器
链接：https://blog.csdn.net/yy139926/article/details/127768446

SpringBoot整合Canal(同步MySQL到Redis)

<dependency>
    <groupId>top.javatool</groupId>
    <artifactId>canal-spring-boot-starter</artifactId>
    <version>1.2.1-RELEASE</version>
</dependency>

application.yml

canal:
  server: localhost:11111
  destination: example

屏蔽日志

logging:
  level:
    tracer: trace # 开启trace级别日志,控制台可以打印es全部请求信息及DSL语句,为了避免重复,开启此项配置后,可以将EE的print-dsl设置为false.
    #top.javatool.canal.client: warn  #禁止AbstractCanalClient 打印常規日志 获取消息 {}

添加Handler 消息监听器

import com.java.pojo.SysDictData;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;

@CanalTable("sys_dict_data") //要监听的表
@Component
public class CacalHandlerService implements EntryHandler<SysDictData> {

    @Override
    public void insert(SysDictData t) { log.info("canal insert event." + t); }

    @Override
    public void update(SysDictData before, SysDictData after) { log.info("canal update event." + after); }

    @Override
    public void delete(SysDictData t) { log.info("canal delete event." + t); }
}