mysql清除数据库缓存 基于binlog+Canal+Redis保姆级教程，帮助您实现数据一致性落地的全过程！

01

MySQL主备复制原理

uMySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)

uMySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)

uMySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

02

MySQL binlog配置

mysql>show variables like 'log_bin';

+—————+——-+

| Variable_name | Value |

+—————+——-+

|log_bin | OFF|

+—————+——-+

1 row in set (0.00 sec)

vi /etc/my.cnf

server_id=1 #mysql实例id,不能canal的slaveId重复

log-bin=/var/lib/mysql/mysql-bin #binlog文件名

binlog_format=ROW #选择row模式

binlog-do-db=timo #监控的数据库

service mysql restart

mysql>show variables like 'log_bin';

+—————+——-+

| Variable_name | Value |

+—————+——-+

|log_bin | ON

+—————+——-+

1 row in set (0.00 sec)

cd /var/lib/mysql

CREATE USER canal IDENTIFIED BY 'BmBfM123456@';GRANT SELECT,INSERT,UPDATE,DELETE REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';FLUSH PRIVILEGES;ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY 'BmBfM123456@';

03

什么是Canal

canal 是阿里巴巴的一个开源项目，基于java实现，整体已经在很多大型的互联网项目生产环境中使用，包括阿里、美团等都有广泛的应用，是一个非常成熟的数据库同步方案，基础的使用只需要进行简单的配置即可。canal是通过模拟成为mysql 的slave的方式， 监听mysql 的binlog日志来获取数据，binlog设置为row模式以后，不仅能获取到执行的每一个增删改的脚本，同时还能获取到修改前和修改后的数据，基于这个特性，canal就能高性能的获取到mysql数据数据的变更。

04

Canal工作原理

uCanal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议

uMySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )

uCanal 解析 binary log 对象(原始为 byte 流)

05

安装和使用Canal

Canal Server下载

官方文档：

项目地址：

下载地址：

进入下载地址，选择canal.deployer-1.1.5.tar.gz

mkdir /data/canal

cd /data/canal

tar -zxvf /opt/canal.deployer-1.1.5.tar.gz -C ./

解压后的目录如下：

- bin   # 运维脚本- conf  # 配置文件  canal_local.properties  # canal本地配置，一般不需要动  canal.properties        # canal服务配置  logback.xml             # logback日志配置  metrics                 # 度量统计配置  spring                  # spring-实例配置，主要和binlog位置计算、一些策略配置相关，可以在canal.properties选用其中的任意一个配置文件  example                 # 实例配置文件夹，一般认为单个数据库对应一个独立的实例配置文件夹    instance.properties   # 实例配置，一般指单个数据库的配置- lib   # 服务依赖包- logs  # 日志文件输出目录

修改内容如下：

在开发和测试环境建议把logback.xml的日志级别修改为DEBUG方便定位问题。

这里需要关注canal.properties和instance.properties两个配置文件。canal.properties文件中，

需要修改：去掉canal.instance.parser.parallelThreadSize = 16这个配置项的「注释」，也就是启用此配置项，和实例解析器的线程数相关，不配置会表现为阻塞或者不进行解析。

canal.serverMode配置项指定为kafka，可选值有tcp、kafka和rocketmq（master分支或者最新的的v1.1.5-alpha-1版本，可以选用rabbitmq），默认是kafka。canal.mq.servers配置需要指定为Kafka服务或者集群Broker的地址，这里配置为127.0.0.1:9092。

canal.mq.servers在不同的canal.serverMode有不同的意义。

kafka模式下，指Kafka服务或者集群Broker的地址，也就是bootstrap.servers

rocketmq模式下，指NameServer列表

rabbitmq模式下，指RabbitMQ服务的Host和Port

instance.properties一般指一个数据库实例的配置，Canal架构支持一个Canal服务实例，处理多个数据库实例的binlog异步解析。

instance.properties需要修改的配置项主要包括：

canal.instance.mysql.slaveId需要配置一个和Master节点的服务ID完全不同的值，这里笔者配置为123456。

配置数据源实例，包括地址、用户、密码和目标数据库：

canal.instance.master.address，这里指定为127.0.0.1:3306。canal.instance.dbUsername，这里指定为canal。canal.instance.dbPassword，这里指定为BmBfM123456@。

新增canal.instance.defaultDatabaseName，这里指定为test（需要在MySQL中建立一个test数据库，见前面的流程）。

Kafka相关配置，这里暂时使用静态topic和单个partition：

canal.mq.topic，这里指定为test，「也就是解析完的binlog结构化数据会发送到Kafka的命名为test的topic中」。canal.mq.partition，这里指定为0。

配置工作做好之后，可以启动Canal服务：sh /data/canal/bin/startup.sh # 查看服务日志tail -100f /data/canal/logs/canal/canal.log# 查看实例日志  -- 一般情况下，关注实例日志即可tail -100f /data/canal/logs/example/example.log

06

Spring Boot整合canal实现数据同步

6.1 依赖

    top.javatool    canal-spring-boot-starter    1.2.1-RELEASE

canal:  # canal实例名称，要跟canal-server运行时设置destination一致  destination: example  # canal地址  server: 172.16.123.12:11111

/** * 获取canal监听mysql的binlog的数据 */// 指定要监听的表@CanalTable("order")@Component@Slf4jpublic class MyItemHandler implements EntryHandler {    @Autowired    private RedisTemplate redisTemplate;    // 监听新增数据，插入redis    @Override    public void insert(Order order) {        log.info("监听新增数据{}，插入redis",order);        redisTemplate.opsForValue().set("order:"+order.getOrder_id(), String.valueOf(order));    }    // 监听修改数据，更新redis，更新本地缓存    @Override    public void update(Order before, Order after) {        log.info("监听修改数据{}，更新redis，更新本地缓存",after);        redisTemplate.delete("order:"+before.getOrder_id());        redisTemplate.opsForValue().set("order:"+after.getOrder_id(),String.valueOf(after));    }    // 监听删除数据，删除本地缓存，删除redis数据    @Override    public void delete(Order order) {        log.info("监听删除数据，删除本地缓存，删除redis数据",order);        redisTemplate.delete("order:"+order.getOrder_id());    }}