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

Canal+Kafka实现MySQL与Redis数据同步(二)

Canal+Kafka实现MySQL与Redis数据同步(二)

创建MQ消费者进行同步

在application.yml配置文件加上kafka的配置信息:

spring:kafka:# Kafka服务地址bootstrap-servers: 127.0.0.1:9092consumer:# 指定一个默认的组名group-id: consumer-group1#序列化反序列化key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializerproducer:key-serializer: org.apache.kafka.common.serialization.StringDeserializervalue-serializer: org.apache.kafka.common.serialization.StringDeserializer# 批量抓取batch-size: 65536# 缓存容量buffer-memory: 524288

根据上面Kafka消费命令那里,我们知道了json数据的结构,可以创建一个CanalBean对象进行接收:

public class CanalBean {//数据private List<TbCommodityInfo> data;//数据库名称private String database;private long es;//递增,从1开始private int id;//是否是DDL语句private boolean isDdl;//表结构的字段类型private MysqlType mysqlType;//UPDATE语句,旧数据private String old;//主键名称private List<String> pkNames;//sql语句private String sql;private SqlType sqlType;//表名private String table;private long ts;//(新增)INSERT、(更新)UPDATE、(删除)DELETE、(删除表)ERASE等等private String type;//getter、setter方法
}
public class MysqlType {private String id;private String commodity_name;private String commodity_price;private String number;private String description;//getter、setter方法
}
public class SqlType {private int id;private int commodity_name;private int commodity_price;private int number;private int description;
}

最后就可以创建一个消费者CanalConsumer进行消费:

@Component
public class CanalConsumer {//日志记录private static Logger log = LoggerFactory.getLogger(CanalConsumer.class);//redis操作工具类@Resourceprivate RedisClient redisClient;//监听的队列名称为:canaltopic@KafkaListener(topics = "canaltopic")public void receive(ConsumerRecord<?, ?> consumer) {String value = (String) consumer.value();log.info("topic名称:{},key:{},分区位置:{},下标:{},value:{}", consumer.topic(), consumer.key(),consumer.partition(), consumer.offset(), value);//转换为javaBeanCanalBean canalBean = JSONObject.parseObject(value, CanalBean.class);//获取是否是DDL语句boolean isDdl = canalBean.getIsDdl();//获取类型String type = canalBean.getType();//不是DDL语句if (!isDdl) {List<TbCommodityInfo> tbCommodityInfos = canalBean.getData();//过期时间long TIME_OUT = 600L;if ("INSERT".equals(type)) {//新增语句for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {String id = tbCommodityInfo.getId();//新增到redis中,过期时间是10分钟redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);}} else if ("UPDATE".equals(type)) {//更新语句for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {String id = tbCommodityInfo.getId();//更新到redis中,过期时间是10分钟redisClient.setString(id, JSONObject.toJSONString(tbCommodityInfo), TIME_OUT);}} else {//删除语句for (TbCommodityInfo tbCommodityInfo : tbCommodityInfos) {String id = tbCommodityInfo.getId();//从redis中删除redisClient.deleteKey(id);}}}}
}

测试MySQL与Redis同步

mysql对应的表结构如下:

CREATE TABLE `tb_commodity_info` (`id` varchar(32) NOT NULL,`commodity_name` varchar(512) DEFAULT NULL COMMENT '商品名称',`commodity_price` varchar(36) DEFAULT '0' COMMENT '商品价格',`number` int(10) DEFAULT '0' COMMENT '商品数量',`description` varchar(2048) DEFAULT '' COMMENT '商品描述',PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='商品信息表';

首先在MySQL创建表。然后启动项目,接着新增一条数据:

INSERT INTO `canaldb`.`tb_commodity_info` (`id`, `commodity_name`, `commodity_price`, `number`, `description`) VALUES ('3e71a81fd80711eaaed600163e046cc3', '叉包', '3.99', '3', '大叉包,老喜欢');

tb_commodity_info表查到新增的数据:

img

Redis也查到了对应的数据,证明同步成功!

img

如果更新呢?试一下Update语句:

UPDATE `canaldb`.`tb_commodity_info` SET `commodity_name`='青菜包',`description`='便宜的青菜包' WHERE `id`='3e71a81fd80711eaaed600163e046cc3';

img

img

没有问题!

总结

canal的缺点:

  1. canal只能同步增量数据。
  2. 不是实时同步,是准实时同步。
  3. 存在一些bug,不过社区活跃度较高,对于提出的bug能及时修复。
  4. MQ顺序性问题。
    网的回答,大家参考一下
    img

尽管有一些缺点,毕竟没有一样技术(产品)是完美的,合适最重要。

相关文章:

Canal+Kafka实现MySQL与Redis数据同步(二)

CanalKafka实现MySQL与Redis数据同步&#xff08;二&#xff09; 创建MQ消费者进行同步 在application.yml配置文件加上kafka的配置信息&#xff1a; spring:kafka:# Kafka服务地址bootstrap-servers: 127.0.0.1:9092consumer:# 指定一个默认的组名group-id: consumer-group…...

NOIP2023模拟19联测40 诡异键盘

题目大意 有一个键盘&#xff0c;上面有 n 1 n1 n1个按键&#xff0c;按下按键 1 ≤ i ≤ n 1\leq i\leq n 1≤i≤n会打印出字符串 S i S_i Si​&#xff0c;按下按键 n 1 n1 n1会删掉结尾的 K K K个字符&#xff0c;如果不足 K K K个字符则全部删完&#xff0c;问打印出 S …...

算法设计与分析 | 众数问题(c语言)

题目 所谓众数&#xff0c;就是对于给定的含有N个元素的多重集合&#xff0c;每个元素在S中出现次数最多的成为该元素的重数&#xff0c; 多重集合S重的重数最大的元素成为众数。例如&#xff1a;S{1,2,2,2,3,5}&#xff0c;则多重集S的众数是2&#xff0c;其重数为3。 现在你…...

sql server外键设置

SQL Server外键设置 简介 在关系型数据库中&#xff0c;外键是一种约束&#xff0c;用于确保数据的完整性和一致性。外键约束定义了一个表中的列与另一个表中的列之间的关系&#xff0c;它可以用来保证数据的一致性、防止数据的破坏和数据冗余。在SQL Server中&#xff0c;我们…...

R语言实现多变量孟德尔随机化分析(1)

多变量孟德尔随机化分析调整了潜在混杂因素的影响。 1、调整哪些因素&#xff1f;参考以往文献。可以分别调整&#xff0c;也可以一起调整。 2、解决了什么问题&#xff1f;某个暴露相关的SNP&#xff0c;往往与某个或者某几个混杂因素相关。可以控制混杂偏倚。 3、如何解释…...

搭建 AI 图像生成器 (SAAS) php laravel

今天来搭一套&#xff0c;AI 图像生成器 是基于 Openai DALLE 2 和 Openai DALLE 3 以及 Stability AI 和稳定扩散 API 构建的脚本&#xff0c;为用户提供了使用简单的提示和大小生成独特自定义图像的可能性。在这个平台上&#xff0c;创意得以快速、高效地实现&#xff0c;借助…...

Maven引用本地jar包

先上命令: ​ mvn install:install-file -Dfile..\.m2\repository\jl1.0.1.jar -DgroupId"com.liz.local" -DartifactId"jl" -Dversion"1.0.1" -Dpackagingjar​ 参数注释: -Dfile: jar 包路径&#xff08;建议放在 meven 的 repository&…...

一起学docker系列之五docker的常用命令--操作容器的命令

目录 前言1 启动容器2 查看容器3 退出容器4 启动已经停止的容器5 重启容器6 停止容器7 删除已经停止的容器8 启动容器说明和举例9 查看容器日志10 查看容器内运行的进程11 查看容器内部细节12 进入正在运行的容器并进行交互13 导入和导出容器结语 前言 当涉及到容器化技术&…...

WPF打开对话框选择文件、选择文件夹

在WPF中实现文件的打开和选择&#xff0c;可以通过使用Microsoft.Win32.OpenFileDialog类来完成。这是一个通用的对话框组件&#xff0c;允许用户在本地文件系统中浏览和选择文件。这个组件属于WPF的一部分&#xff0c;因此不需要引用额外的库。 以下是一个如何使用OpenFileDi…...

nginx学习(3)

Nginx 负载均衡 实战案例 实现效果 浏览器地址栏输入地址 http://172.31.0.99/oa/a.html&#xff0c;负载均衡效果&#xff0c;平均 8083 和 8084 端口中 一、配置 1、先创建2个文件夹&#xff0c;并将apache-tomcat-8.5.87解压到tomcat8083和tomcat8084中 &#xff08;或…...

【系统架构设计】计算机公共基础知识: 4 数据库系统

目录 一 数据库模式 二 分布式数据库 三 索引和视图 四 数据库设计 五 关系代数...

主键问题以及分布式 id

分布式 id 需要处理的问题主要是同一时间在多台机器中保证生成的 id 唯一&#xff0c;为了这么做我们可以这么做&#xff1a; 分布式 id 生成策略 先说几个已经被淘汰的策略引出分布式 id 的问题 1&#xff0c;UUID&#xff1a;UUID 随机并且唯一&#xff0c;在单一的数据库…...

ReentranReadWriteLock 使用案例

ReentranReadWriteLock使用案例 /*** ReentranReadWriteLock 使用案例* 读线程共享* 写线程互斥*/ public class ReentrantReadWriteLockExample {private String news;private ReentrantReadWriteLock lock new ReentrantReadWriteLock();public String readNews() {lock.re…...

“我们把最扎心的话,说给了自己最亲近的人” 何解?| IDCF

引子 我们把最好的一面给了陌生人&#xff0c;却把最扎心的话&#xff0c;说给了自己最亲近的人。 我们往往会对关心自己的人发脾气&#xff0c;很多时候意图是好的&#xff0c;表达方式却简单粗暴&#xff0c;结果自然不必多言。你认为自己给的是反馈和建议&#xff0c;对方…...

MongoDB之索引和聚合

文章目录 一、索引1、说明2、原理3、相关操作3.1、创建索引3.2、查看集合索引3.3、查看集合索引大小3.4、删除集合所有索引&#xff08;不包含_id索引&#xff09;3.5、删除集合指定索引 4、复合索引 二、聚合1、说明2、使用 总结 一、索引 1、说明 索引通常能够极大的提高查…...

【GEE】基于GEE进行非监督学习

1 简介与摘要 之前写了多季节叠加的监督学习&#xff0c;所以这次简单写一个非监督学习吧。。 这次为了简单明了&#xff0c;就不整那么多虚的了&#xff0c;在这里我不叠图层了&#xff0c;有需要的可以参考前一篇博客自己添加输入的图层。 2 制作输入影像 首先&#xff0c…...

多视图聚类的论文阅读(一)

当聚类的方式使用的是某一类预定义好的相似性度量时&#xff0c; 会出现如下情况&#xff1a; 数据聚类方面取得了成功&#xff0c;但它们通常依赖于预定义的相似性度量&#xff0c;而这些度量受原始方法的影响:当输入维数相对较高时&#xff0c;往往是无效的。 1. Deep Mult…...

K-Means算法进行分类

已知数据集D中有9个数据点&#xff0c;分别是&#xff08;1,2&#xff09;&#xff0c;(2,3), (2,1), (3,1),(2,4),(3,5),(4,3),(1,5),(4,2)。采用K-Means算法进行聚类&#xff0c;k2&#xff0c;设初始中心点为&#xff08;1.1,2.2&#xff09;&#xff0c;&#xff08;2.3,3.…...

深度学习交通车辆流量分析 - 目标检测与跟踪 - python opencv 计算机竞赛

文章目录 0 前言1 课题背景2 实现效果3 DeepSORT车辆跟踪3.1 Deep SORT多目标跟踪算法3.2 算法流程 4 YOLOV5算法4.1 网络架构图4.2 输入端4.3 基准网络4.4 Neck网络4.5 Head输出层 5 最后 0 前言 &#x1f525; 优质竞赛项目系列&#xff0c;今天要分享的是 &#x1f6a9; *…...

网络协议入门 笔记一

一、服务器和客户端及java的概念 JVM (Java Virtual Machine) : Java虚拟机&#xff0c;Java的跨平台:一次编译&#xff0c;到处运行&#xff0c;编译生成跟平台无关的字节码文件 (class文件)&#xff0c;由对应平台的JVM解析字节码为机器指令 (010101)。 如下图所示&#xff0…...

MPNet:旋转机械轻量化故障诊断模型详解python代码复现

目录 一、问题背景与挑战 二、MPNet核心架构 2.1 多分支特征融合模块(MBFM) 2.2 残差注意力金字塔模块(RAPM) 2.2.1 空间金字塔注意力(SPA) 2.2.2 金字塔残差块(PRBlock) 2.3 分类器设计 三、关键技术突破 3.1 多尺度特征融合 3.2 轻量化设计策略 3.3 抗噪声…...

【Oracle APEX开发小技巧12】

有如下需求&#xff1a; 有一个问题反馈页面&#xff0c;要实现在apex页面展示能直观看到反馈时间超过7天未处理的数据&#xff0c;方便管理员及时处理反馈。 我的方法&#xff1a;直接将逻辑写在SQL中&#xff0c;这样可以直接在页面展示 完整代码&#xff1a; SELECTSF.FE…...

OkHttp 中实现断点续传 demo

在 OkHttp 中实现断点续传主要通过以下步骤完成&#xff0c;核心是利用 HTTP 协议的 Range 请求头指定下载范围&#xff1a; 实现原理 Range 请求头&#xff1a;向服务器请求文件的特定字节范围&#xff08;如 Range: bytes1024-&#xff09; 本地文件记录&#xff1a;保存已…...

解决本地部署 SmolVLM2 大语言模型运行 flash-attn 报错

出现的问题 安装 flash-attn 会一直卡在 build 那一步或者运行报错 解决办法 是因为你安装的 flash-attn 版本没有对应上&#xff0c;所以报错&#xff0c;到 https://github.com/Dao-AILab/flash-attention/releases 下载对应版本&#xff0c;cu、torch、cp 的版本一定要对…...

微信小程序云开发平台MySQL的连接方式

注&#xff1a;微信小程序云开发平台指的是腾讯云开发 先给结论&#xff1a;微信小程序云开发平台的MySQL&#xff0c;无法通过获取数据库连接信息的方式进行连接&#xff0c;连接只能通过云开发的SDK连接&#xff0c;具体要参考官方文档&#xff1a; 为什么&#xff1f; 因为…...

Java编程之桥接模式

定义 桥接模式&#xff08;Bridge Pattern&#xff09;属于结构型设计模式&#xff0c;它的核心意图是将抽象部分与实现部分分离&#xff0c;使它们可以独立地变化。这种模式通过组合关系来替代继承关系&#xff0c;从而降低了抽象和实现这两个可变维度之间的耦合度。 用例子…...

MFC 抛体运动模拟:常见问题解决与界面美化

在 MFC 中开发抛体运动模拟程序时,我们常遇到 轨迹残留、无效刷新、视觉单调、物理逻辑瑕疵 等问题。本文将针对这些痛点,详细解析原因并提供解决方案,同时兼顾界面美化,让模拟效果更专业、更高效。 问题一:历史轨迹与小球残影残留 现象 小球运动后,历史位置的 “残影”…...

MySQL 索引底层结构揭秘:B-Tree 与 B+Tree 的区别与应用

文章目录 一、背景知识&#xff1a;什么是 B-Tree 和 BTree&#xff1f; B-Tree&#xff08;平衡多路查找树&#xff09; BTree&#xff08;B-Tree 的变种&#xff09; 二、结构对比&#xff1a;一张图看懂 三、为什么 MySQL InnoDB 选择 BTree&#xff1f; 1. 范围查询更快 2…...

手机平板能效生态设计指令EU 2023/1670标准解读

手机平板能效生态设计指令EU 2023/1670标准解读 以下是针对欧盟《手机和平板电脑生态设计法规》(EU) 2023/1670 的核心解读&#xff0c;综合法规核心要求、最新修正及企业合规要点&#xff1a; 一、法规背景与目标 生效与强制时间 发布于2023年8月31日&#xff08;OJ公报&…...

Spring AI Chat Memory 实战指南:Local 与 JDBC 存储集成

一个面向 Java 开发者的 Sring-Ai 示例工程项目&#xff0c;该项目是一个 Spring AI 快速入门的样例工程项目&#xff0c;旨在通过一些小的案例展示 Spring AI 框架的核心功能和使用方法。 项目采用模块化设计&#xff0c;每个模块都专注于特定的功能领域&#xff0c;便于学习和…...