当前位置：首页 > news >正文

Canal笔记:安装与整合Springboot模式Mysql同步Redis

news 文章来源：https://blog.csdn.net/Think_and_work/article/details/134812037 2024/9/20 8:52:42

官方文档

https://github.com/alibaba/canal

使用场景

学习一件东西前，要知道为什么使用它。

1、同步mysql数据到redis

常规情况下，产生数据的方法可能有很多地方，那么就需要在多个地方中，都去做mysql数据同步到redis的处理，相对麻烦很多。
可以使用canal，对mysql进行集中，统一的处理。

概述

canal [kə’næl]，译意为水道/管道/沟渠，主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费
当前的 canal 支持源端 MySQL 版本包括 5.1.x , 5.5.x , 5.6.x , 5.7.x , 8.0.x

原理

MySQL主备复制原理

MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)
MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)
MySQL slave 重放 relaylog 中事件，将数据变更反映它自己的数据

canal 工作原理

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议
MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )
canal 解析 binary log 对象(原始为 byte 流)

架构

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

安装和准备

Centos7安装Canal

1、Mysql配置

开启binlog日志

如果是使用Linux安装的话，则直接找my.cnf，直接修改内容即可

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

docker安装

1、安装my.cnf文件

[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复

2、修改docker-compose.yaml内容
配置挂载卷前面路径为my.cnf的路径，/etc/mysql/conf.d的路径

3、查询是否成功

show variables like "%server_id%";

show variables like 'log_bin';

获取bin_log当前位置

show master status;

获取后，记录下来，然后不要动数据库了

创建canal数据库用户
这里可以使用
mysql -uroot -p登录进入设置，
或者直接可视化页面

CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

2、Canal下载

下载
方式一：关注公众号 I am Walker 回复canal

方式二：https://github.com/alibaba/canal/releases?page=2

# 创建文件夹
mkdir /opt/env/canal
# 解压
tar -zxvf canal.deployer-1.1.4.tar.gz -C /opt/env/canal

3、配置文件修改

进入canal/conf/example/instance.properties
主要修改下列相关参数

# 数据库
canal.instance.master.address=127.0.0.1:3306
# bin log日志
canal.instance.master.journal.name=mysql-bin.000001
# bin log写入位置
canal.instance.master.position=157#数据库账号密码
canal.instance.dbUsername=canal
canal.instance.dbPassword=123456

之后进入 canal/bin 执行 ./startup.sh
查看是否启动

有CanalLauncher则代表ok，或者看日志也ok

场景

springboot整合

简单整合

1、依赖

<dependency><groupId>com.alibaba.otter</groupId><artifactId>canal.client</artifactId><version>1.1.4</version></dependency>

2、配置文件

canal:# 服务地址serverAddress: localhost# 端口serverPort: 11111# 订阅 库 表subscrie: ".*\\..*"# batchSize: 100# 实例instance:- example

subscrie配置


全库全表	
connector.subscribe(".*\\..*")
指定库全表	
connector.subscribe("test\\..*")
单表
connector.subscribe("test.user")
多规则组合使用
connector.subscribe("test\\..*,test2.user1,test3.user2")

3、properties类

package com.walker.mybatisplus.canal;import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;import java.util.List;@Data
@Component
@ConfigurationProperties(value = "canal")
public class CanalProperties {private String serverAddress;private Integer serverPort;private String subcribe;private Integer batchSize;private List<String> instance;}

4、监听类编写

package com.walker.mybatisplus.canal;import cn.hutool.core.collection.CollUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
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 com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;@Slf4j
@Component
public class CanalListener {@Autowiredprivate CanalProperties canalProperties;public static Map<String,Integer> NUM_MAP=new HashMap<>();/*** 可能会有多业务，不同的业务，应该有多个处理类,不要使用if等*/@PostConstructpublic void run() throws InterruptedException, InvalidProtocolBufferException {//创建Canal连接对象CanalConnector conn = CanalConnectors.newSingleConnector(new InetSocketAddress(canalProperties.getServerAddress(),canalProperties.getServerPort()),canalProperties.getInstance().get(0),null, null);while(true){//连接conn.connect();
//            监听的数据库和表conn.subscribe(canalProperties.getSubcribe());//回滚操作conn.rollback();//获取信息Message message = conn.getWithoutAck(canalProperties.getBatchSize());long id = message.getId();List<CanalEntry.Entry> entries = message.getEntries();if(id!=-1&&entries.size()>0){//处理数据messageProcess(entries);}else{//防止重复链接数据库Thread.sleep(1000);}//确认消费信息conn.ack(id);//释放连接conn.disconnect();}}private void messageProcess(List<CanalEntry.Entry> entries) throws InvalidProtocolBufferException {for (CanalEntry.Entry entry : entries) {log.info("接收Entry:{}", entry);CanalEntry.Header header = entry.getHeader();//数据库String schemaName = header.getSchemaName();//表名String tableName = header.getTableName();//事件类型CanalEntry.EventType eventType = header.getEventType();//这里可以对数据库和表进行一个重新判断 虽然在subscribe已经定义，但是一般可以配置一个库，然后表的可能可以是全部表
//            对库进行判断if(!"walker_share".equals(schemaName)){continue;}//对表进行判断//这里只是一个案例，如果是实际场景，可以使用工厂模式去编写，不然会有很多的ifif("dish".equals(tableName)){//获取修改数据List<CanalEntry.RowData> rowDataList = getRowDataList(entry);//新增if(eventType.getNumber()==CanalEntry.EventType.INSERT_VALUE){log.info("新增事件");if(CollUtil.isEmpty(rowDataList)) continue;//模拟场景：获取新增的数据，并存储到redis中，这里是直接存储到Map中for (CanalEntry.RowData rowData : rowDataList) {List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();for (CanalEntry.Column column : afterColumnsList) {//获取name的类型if("type".equals(column.getName())){//模拟redis  根据类型进行分类String key = column.getValue();NUM_MAP.put(key,NUM_MAP.getOrDefault(key,0)+1);log.info("NUM_MAP {}",NUM_MAP);continue;}}}}if(eventType.getNumber()==CanalEntry.EventType.UPDATE_VALUE){log.info("修改事件");}if(eventType.getNumber()==CanalEntry.EventType.DELETE_VALUE){log.info("删除事件");}}}}//获取row数据private List<CanalEntry.RowData> getRowDataList(CanalEntry.Entry entry) {CanalEntry.RowChange rowChange=null;try {//解析数据rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());} catch (InvalidProtocolBufferException e) {throw new RuntimeException("解析出现异常 data:" + entry.toString(), e);}List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();return rowDatasList;}
}