使用TCP方式拉取Canal数据
1 Canal对接Kafka联调
1.1 配置修改
canal.properties

修改 zk:
canal.zkServers = 10.51.50.219:2181
instance.properties
开启配置项:
canal.mq.dynamicTopic 是 Canal 的 MQ 动态 Topic 配置项:
test_javaedge_01是kafka 的 topictest_db.users要监控的数据库、表- 当
test_db.users表发生变化时,Canal 将会把变化的数据推送到名为test_javaedge_01:test_db.users的 MQ Topic 中。
canal.mq.dynamicTopic=test_javaedge_01:test_db\\.users
开启一个消费者
[root@javaedge-kafka-dev bin]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_javaedge_01
datagrip 新增数据:

消费到该数据:

2 使用TCP方式拉取Canal数据
现在 serverMode 改回tcp。重启
javaedge@JavaEdgedeMac-mini deployer % jps
71002 CanalLauncher
javaedge@JavaEdgedeMac-mini deployer %
canal 同步程序
package com.javaedge.canal;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.common.base.CaseFormat;import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;public class CanalClientApp {public static void main(String[] args) throws Exception {CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("localhost", 11111),"example",null, null);while (true) {connector.connect();connector.subscribe("test_db.users");Message message = connector.get(100);List<CanalEntry.Entry> entries = message.getEntries();if (entries.size()>0) {for (CanalEntry.Entry entry : entries) {String tableName = entry.getHeader().getTableName();CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();CanalEntry.EventType eventType = rowChange.getEventType();if (eventType == CanalEntry.EventType.INSERT) {for (CanalEntry.RowData rowData : rowDatasList) {List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();HashMap<Object, Object> map = new HashMap<>();for (CanalEntry.Column column : afterColumnsList) {String key = CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, column.getName());map.put(key, column.getValue());}System.out.println("tableName=" + tableName + " map=" + JSON.toJSONString(map));}}}}}}
}
运行程序。操作 user 数据表,新增一行数据:

程序输出:

显然,后续不管你想把数据同步到哪儿去,都完全自由!
数据链路
MySQL -》canal server(tcp)-》canal client-》kafka。
相关文章:
使用TCP方式拉取Canal数据
1 Canal对接Kafka联调 1.1 配置修改 canal.properties 修改 zk: canal.zkServers 10.51.50.219:2181instance.properties 开启配置项: canal.mq.dynamicTopic 是 Canal 的 MQ 动态 Topic 配置项: test_javaedge_01 是kafka 的 topicte…...
Docker安装mysql实战说明
安装前准备 在安装MySQL之前,你需要确保已经正确安装和配置了Docker,可以通过以下命令检查Docker是否已正确安装: docker --version如果Docker已经成功安装,你将看到Docker的版本信息。 下载mysql的镜像 Docker Hub是一个存储…...
前端DOM操作精解:基础概念、方法与最佳实践
引言 本文将深入探讨前端开发中的DOM操作,包括基础概念、常用方法和最佳实践。通过清晰易懂的解释和实际案例分析,我们将一起了解如何最有效地使用DOM操作来提升前端应用的用户体验。 一、DOM操作入门 在深入探讨DOM操作之前,我们先要理解…...
python sorted函数详解2023.9.11
sorted函数详解 1. 输入和输出2. key传入函数 1. 输入和输出 help(sorted) Help on built-in function sorted in module builtins: sorted(iterable, /, *, keyNone, reverseFalse)Return a new list containing all items from the iterable in ascending order.A custom k…...
Spring Reactive:响应式编程与WebFlux的深度探索
🌷🍁 博主猫头虎(🐅🐾)带您 Go to New World✨🍁 🦄 博客首页——🐅🐾猫头虎的博客🎐 🐳 《面试题大全专栏》 🦕 文章图文…...
Qt应用开发(基础篇)——工具按钮类 QToolButton
一、前言 QToolButton类继承于QAbstractButton,该部件为命令或选项提供了一个快速访问按钮,通常用于QToolBar中。 按钮基类 QAbstractButton QToolButton是一个特殊的按钮,一般显示文本,只显示图标,结合toolBar使用。它…...
【数据结构面试题】栈与队列的相互实现
目录 1.队列实现栈 1.1创建栈 1.2判断是否为空 1.3入栈 1.4出栈 1.5获取栈顶元素 1.6完整代码 2. 用栈实现队列 2.1创建队列 2.2判断是否为空 2.3入队列 2.4出队列 2.5获取队头元素 2.6完整代码 1.队列实现栈 用队列实现栈https://leetcode.cn/problems/impleme…...
华为认证和红帽认证哪个比较好考呢
华为认证和红帽认证的考试难度、学习内容、适用范围等方面都有所不同,因此哪个比较好考要视具体情况而定: 考试难度:红帽认证的考试难度较高,需要考生具备较高的技术水平和实践经验;而华为认证则更注重基础知识的考察…...
[Java]_[中级]_[使用okhttp3和HttpClient代理访问外部网络]
场景 Java的http库常用的有HttpClient和Okhttp3, 如果公司有限制网络访问,需要代理才可以访问外网,那么如何使用代理Proxy? <dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient<…...
ubuntu 20.04 docker 安装 mysql
要在Ubuntu 20.04上安装Docker并运行MySQL容器,您可以按照以下步骤操作: 1.更新系统包列表: sudo apt update2.安装Docker: sudo apt install docker.io3.启动Docker服务并设置其开机自启动: sudo systemctl start…...
C++在C语言基础上的优化
目录 一、命名空间 1、命名空间的定义 2、命名空间的使用 二、输入&输出 三、缺省参数 1、缺省参数的概念 2、缺省参数的分类 四、函数重载 五、引用 1.引用的概念 2.引用的特性 3、引用和指针的区别 六、内联函数 七、基于范围的for循环 一、命名空间 命名空…...
分享一个python实验室设备预约管理系统 实验室设备维修系统源码 lw 调试
💕💕作者:计算机源码社 💕💕个人简介:本人七年开发经验,擅长Java、Python、PHP、.NET、微信小程序、爬虫、大数据等,大家有这一块的问题可以一起交流! 💕&…...
兵者多诡(HCTF2016)
环境:https://github.com/MartinxMax/CTFer_Zero_one 题目简介 解题过程 登录首页 提交png图片上传抓包,可以看到是向upload文件提交数据 在fp参数中尝试伪协议读取home.php文件 http://127.0.0.1:88/HCTF2016-LFI/home.php?fpphp://filter/readconvert.base64…...
【JAVA-Day04】Java关键字和示例:深入了解常用关键字的用法
Java关键字和示例:深入了解常用关键字的用法 摘要Java 关键字、标识符和命名规范一、Java 关键字常用关键字DEMO1. 示例代码使用 if 和 else 关键字:2. 示例代码使用 for 循环:3. 示例代码使用 switch 关键字:4. 示例代码使用 wh…...
Android请求网络报错:not permitted by network security policy
一、错误记录 https的接口请求正常的, 请求http的接口时报错:not permitted by network security policy 二、问题分析 原因: 由于 Android P(版本27以上) 限制了明文流量的网络请求,非加密的流量请求都会被系统禁止掉。如果当…...
python报错:ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1
python报错:ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1 问题分析 说明:requests包引入了urllib3,而新版本的urllib3 需要OpenSSL 1.1.1以上版本,否则报错: ImportError: urllib3 v2.0 only supports Ope…...
如何使用adb command来设置cpu频率和核数
透過ADB Shell設定CPU開核與freq的command與用法如下: # Disable PPM echo 0 > /proc/ppm/enabled # Enable PPM (Default) echo 1 > /proc/ppm/enabled echo 0 > /proc/ppm/enabled Fixed # Core for each cluster echo X Y > /proc/ppm/policy/ut_fix_core_num …...
236. 二叉树的最近公共祖先
236. 二叉树的最近公共祖先 题目-中等难度示例1. dfs 题目-中等难度 给定一个二叉树, 找到该树中两个指定节点的最近公共祖先。 百度百科中最近公共祖先的定义为:“对于有根树 T 的两个节点 p、q,最近公共祖先表示为一个节点 x,满足 x 是 p…...
Git常见问题:git pull 和 git pull --rebase二者区别
git pull 和 git pull --rebase 都是从远程仓库获取最新的更改并将其合并到本地分支。但它们之间的区别在于合并方式。以下是它们之间的主要区别: git pull: 当你执行 git pull 时,Git 会执行以下两个操作: git fetchÿ…...
关于HarmonyOS元服务的主题演讲与合作签约
一、感言 坚持中,总会有很多意想不到的收获。 前几次参与HDC时更多的是观众、开发者、专家的身份,以参观、学习、交流为主。 通过几年的努力,和HarmonyOS功能成长,在2023年的HDC大会中,有了我的演讲,并带领…...
【力扣数据库知识手册笔记】索引
索引 索引的优缺点 优点1. 通过创建唯一性索引,可以保证数据库表中每一行数据的唯一性。2. 可以加快数据的检索速度(创建索引的主要原因)。3. 可以加速表和表之间的连接,实现数据的参考完整性。4. 可以在查询过程中,…...
OkHttp 中实现断点续传 demo
在 OkHttp 中实现断点续传主要通过以下步骤完成,核心是利用 HTTP 协议的 Range 请求头指定下载范围: 实现原理 Range 请求头:向服务器请求文件的特定字节范围(如 Range: bytes1024-) 本地文件记录:保存已…...
从面试角度回答Android中ContentProvider启动原理
Android中ContentProvider原理的面试角度解析,分为已启动和未启动两种场景: 一、ContentProvider已启动的情况 1. 核心流程 触发条件:当其他组件(如Activity、Service)通过ContentR…...
ubuntu22.04有线网络无法连接,图标也没了
今天突然无法有线网络无法连接任何设备,并且图标都没了 错误案例 往上一顿搜索,试了很多博客都不行,比如 Ubuntu22.04右上角网络图标消失 最后解决的办法 下载网卡驱动,重新安装 操作步骤 查看自己网卡的型号 lspci | gre…...
如何配置一个sql server使得其它用户可以通过excel odbc获取数据
要让其他用户通过 Excel 使用 ODBC 连接到 SQL Server 获取数据,你需要完成以下配置步骤: ✅ 一、在 SQL Server 端配置(服务器设置) 1. 启用 TCP/IP 协议 打开 “SQL Server 配置管理器”。导航到:SQL Server 网络配…...
面试高频问题
文章目录 🚀 消息队列核心技术揭秘:从入门到秒杀面试官1️⃣ Kafka为何能"吞云吐雾"?性能背后的秘密1.1 顺序写入与零拷贝:性能的双引擎1.2 分区并行:数据的"八车道高速公路"1.3 页缓存与批量处理…...
在RK3588上搭建ROS1环境:创建节点与数据可视化实战指南
在RK3588上搭建ROS1环境:创建节点与数据可视化实战指南 背景介绍完整操作步骤1. 创建Docker容器环境2. 验证GUI显示功能3. 安装ROS Noetic4. 配置环境变量5. 创建ROS节点(小球运动模拟)6. 配置RVIZ默认视图7. 创建启动脚本8. 运行可视化系统效果展示与交互技术解析ROS节点通…...
拟合问题处理
在机器学习中,核心任务通常围绕模型训练和性能提升展开,但你提到的 “优化训练数据解决过拟合” 和 “提升泛化性能解决欠拟合” 需要结合更准确的概念进行梳理。以下是对机器学习核心任务的系统复习和修正: 一、机器学习的核心任务框架 机…...
Qwen系列之Qwen3解读:最强开源模型的细节拆解
文章目录 1.1分钟快览2.模型架构2.1.Dense模型2.2.MoE模型 3.预训练阶段3.1.数据3.2.训练3.3.评估 4.后训练阶段S1: 长链思维冷启动S2: 推理强化学习S3: 思考模式融合S4: 通用强化学习 5.全家桶中的小模型训练评估评估数据集评估细节评估效果弱智评估和民间Arena 分析展望 如果…...
【记录坑点问题】IDEA运行:maven-resources-production:XX: OOM: Java heap space
问题:IDEA出现maven-resources-production:operation-service: java.lang.OutOfMemoryError: Java heap space 解决方案:将编译的堆内存增加一点 位置:设置setting-》构建菜单build-》编译器Complier...
