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

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

news 2025/7/28 0:40:04

官方文档

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 中事件，将数据变更反映它自己的数据

image.png

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)

image.png

安装和准备

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的路径
image.png

3、查询是否成功

show variables like "%server_id%";

image.png

show variables like 'log_bin';

image.png

获取bin_log当前位置

show master status;

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

创建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
image.png

# 创建文件夹
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
查看是否启动
image.png
有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;}
}

相关类和配置

CanalConnector

image.png

CanalEntry

image.png

EntryType

image.png

Header

image.png

EventType

事件类型，可以根据事件类型去做不一样的操作
image.png

RowChange

image.png

获取数据

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();
log.info("rowDatasList:{}",rowDatasList);
RowData

image.png

CanalEntry.Column

属性
image.png

问题

IOException: caching_sha2_password Auth failed

因为mysql8.0.3后身份检验方式为caching_sha2_password，但canal使用的是mysql_native_password，因此需要设置检验方式（如果该版本之前的可跳过），否则会报错IOException: caching_sha2_password Auth failed

参考文档

Java开发 - Canal的基本用法_canal java-CSDN博客
15分钟学会Canal安装与部署-CSDN博客
SpringBoot整合Canal1.1.6并同步数据到Redis（超详细和很多踩坑点）_canal同步数据到redis-CSDN博客

http://www.lryc.cn/news/256016.html

相关文章：

  • C++的继承语法
  • C# .NET平台提取PDF表格数据，并转换为txt、CSV和Excel表格文件
  • spring boot学习第五篇:spring boot与JPA结合
  • 代理IP怎么使用？Mac苹果系统设置http代理IP教程
  • postgresql_conf中常用配置项
  • 使用MfgTool烧写前需准备的文件
  • SAP UI5 walkthrough step4 XML Views
  • Java 1对1
  • 云服务器Centos中安装Docker
  • 人工智能教程（三）：更多有用的 Python 库
  • 【带头学C++】----- 九、类和对象 ---- 9.10 C++设计模式之单例模式设计
  • Qt之QCache和QContiguousCache
  • Django讲课笔记01：初探Django框架
  • JS中的闭包
  • 深度学习在计算机视觉中的应用
  • 模板与泛型编程
  • 【Fastadmin】一个完整的轮播图功能示例
  • Ribbon 饥饿加载
  • 【AIGC】大语言模型的采样策略--temperature、top-k、top-p等
  • pip的基本命令和使用
  • RocksDB实现原理
  • mysql 链接超时的几个参数详解
  • [架构之路-259]：目标系统 - 设计方法 - 软件工程 - 软件设计 - 架构设计 - 面向服务的架构SOA与微服务架构（以服务为最小的构建单位）
  • 7z压缩成jar包
  • python-缩进式编码+算术运算符+定义与赋值代码示例
  • APM性能分享观看有感
  • 免费好用的API接口攻略
  • 五、C#笔记
  • Oracle的错误信息帮助：Error Help
  • 阿里云磁盘在线扩容