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

使用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 的 topic
  • test_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&#xff1a; canal.zkServers 10.51.50.219:2181instance.properties 开启配置项&#xff1a; canal.mq.dynamicTopic 是 Canal 的 MQ 动态 Topic 配置项&#xff1a; test_javaedge_01 是kafka 的 topicte…...

Docker安装mysql实战说明

安装前准备 在安装MySQL之前&#xff0c;你需要确保已经正确安装和配置了Docker&#xff0c;可以通过以下命令检查Docker是否已正确安装&#xff1a; docker --version如果Docker已经成功安装&#xff0c;你将看到Docker的版本信息。 下载mysql的镜像 Docker Hub是一个存储…...

前端DOM操作精解:基础概念、方法与最佳实践

引言 本文将深入探讨前端开发中的DOM操作&#xff0c;包括基础概念、常用方法和最佳实践。通过清晰易懂的解释和实际案例分析&#xff0c;我们将一起了解如何最有效地使用DOM操作来提升前端应用的用户体验。 一、DOM操作入门 在深入探讨DOM操作之前&#xff0c;我们先要理解…...

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的深度探索

&#x1f337;&#x1f341; 博主猫头虎&#xff08;&#x1f405;&#x1f43e;&#xff09;带您 Go to New World✨&#x1f341; &#x1f984; 博客首页——&#x1f405;&#x1f43e;猫头虎的博客&#x1f390; &#x1f433; 《面试题大全专栏》 &#x1f995; 文章图文…...

Qt应用开发(基础篇)——工具按钮类 QToolButton

一、前言 QToolButton类继承于QAbstractButton&#xff0c;该部件为命令或选项提供了一个快速访问按钮&#xff0c;通常用于QToolBar中。 按钮基类 QAbstractButton QToolButton是一个特殊的按钮&#xff0c;一般显示文本&#xff0c;只显示图标&#xff0c;结合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…...

华为认证和红帽认证哪个比较好考呢

华为认证和红帽认证的考试难度、学习内容、适用范围等方面都有所不同&#xff0c;因此哪个比较好考要视具体情况而定&#xff1a; 考试难度&#xff1a;红帽认证的考试难度较高&#xff0c;需要考生具备较高的技术水平和实践经验&#xff1b;而华为认证则更注重基础知识的考察…...

[Java]_[中级]_[使用okhttp3和HttpClient代理访问外部网络]

场景 Java的http库常用的有HttpClient和Okhttp3, 如果公司有限制网络访问&#xff0c;需要代理才可以访问外网&#xff0c;那么如何使用代理Proxy&#xff1f; <dependency><groupId>org.apache.httpcomponents</groupId><artifactId>httpclient<…...

ubuntu 20.04 docker 安装 mysql

要在Ubuntu 20.04上安装Docker并运行MySQL容器&#xff0c;您可以按照以下步骤操作&#xff1a; 1.更新系统包列表&#xff1a; sudo apt update2.安装Docker&#xff1a; sudo apt install docker.io3.启动Docker服务并设置其开机自启动&#xff1a; sudo systemctl start…...

C++在C语言基础上的优化

目录 一、命名空间 1、命名空间的定义 2、命名空间的使用 二、输入&输出 三、缺省参数 1、缺省参数的概念 2、缺省参数的分类 四、函数重载 五、引用 1.引用的概念 2.引用的特性 3、引用和指针的区别 六、内联函数 七、基于范围的for循环 一、命名空间 命名空…...

分享一个python实验室设备预约管理系统 实验室设备维修系统源码 lw 调试

&#x1f495;&#x1f495;作者&#xff1a;计算机源码社 &#x1f495;&#x1f495;个人简介&#xff1a;本人七年开发经验&#xff0c;擅长Java、Python、PHP、.NET、微信小程序、爬虫、大数据等&#xff0c;大家有这一块的问题可以一起交流&#xff01; &#x1f495;&…...

兵者多诡(HCTF2016)

环境:https://github.com/MartinxMax/CTFer_Zero_one 题目简介 解题过程 登录首页 提交png图片上传抓包&#xff0c;可以看到是向upload文件提交数据 在fp参数中尝试伪协议读取home.php文件 http://127.0.0.1:88/HCTF2016-LFI/home.php?fpphp://filter/readconvert.base64…...

【JAVA-Day04】Java关键字和示例:深入了解常用关键字的用法

Java关键字和示例&#xff1a;深入了解常用关键字的用法 摘要Java 关键字、标识符和命名规范一、Java 关键字常用关键字DEMO1. 示例代码使用 if 和 else 关键字&#xff1a;2. 示例代码使用 for 循环&#xff1a;3. 示例代码使用 switch 关键字&#xff1a;4. 示例代码使用 wh…...

Android请求网络报错:not permitted by network security policy

一、错误记录 https的接口请求正常的&#xff0c; 请求http的接口时报错&#xff1a;not permitted by network security policy 二、问题分析 原因&#xff1a; 由于 Android P(版本27以上) 限制了明文流量的网络请求&#xff0c;非加密的流量请求都会被系统禁止掉。如果当…...

python报错:ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1

python报错&#xff1a;ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1 问题分析 说明&#xff1a;requests包引入了urllib3&#xff0c;而新版本的urllib3 需要OpenSSL 1.1.1以上版本&#xff0c;否则报错&#xff1a; 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 题目-中等难度 给定一个二叉树, 找到该树中两个指定节点的最近公共祖先。 百度百科中最近公共祖先的定义为&#xff1a;“对于有根树 T 的两个节点 p、q&#xff0c;最近公共祖先表示为一个节点 x&#xff0c;满足 x 是 p…...

Git常见问题:git pull 和 git pull --rebase二者区别

git pull 和 git pull --rebase 都是从远程仓库获取最新的更改并将其合并到本地分支。但它们之间的区别在于合并方式。以下是它们之间的主要区别&#xff1a; git pull&#xff1a; 当你执行 git pull 时&#xff0c;Git 会执行以下两个操作&#xff1a; git fetch&#xff…...

关于HarmonyOS元服务的主题演讲与合作签约

一、感言 坚持中&#xff0c;总会有很多意想不到的收获。 前几次参与HDC时更多的是观众、开发者、专家的身份&#xff0c;以参观、学习、交流为主。 通过几年的努力&#xff0c;和HarmonyOS功能成长&#xff0c;在2023年的HDC大会中&#xff0c;有了我的演讲&#xff0c;并带领…...

java_网络服务相关_gateway_nacos_feign区别联系

1. spring-cloud-starter-gateway 作用&#xff1a;作为微服务架构的网关&#xff0c;统一入口&#xff0c;处理所有外部请求。 核心能力&#xff1a; 路由转发&#xff08;基于路径、服务名等&#xff09;过滤器&#xff08;鉴权、限流、日志、Header 处理&#xff09;支持负…...

Python:操作 Excel 折叠

💖亲爱的技术爱好者们,热烈欢迎来到 Kant2048 的博客!我是 Thomas Kant,很开心能在CSDN上与你们相遇~💖 本博客的精华专栏: 【自动化测试】 【测试经验】 【人工智能】 【Python】 Python 操作 Excel 系列 读取单元格数据按行写入设置行高和列宽自动调整行高和列宽水平…...

云启出海,智联未来|阿里云网络「企业出海」系列客户沙龙上海站圆满落地

借阿里云中企出海大会的东风&#xff0c;以**「云启出海&#xff0c;智联未来&#xff5c;打造安全可靠的出海云网络引擎」为主题的阿里云企业出海客户沙龙云网络&安全专场于5.28日下午在上海顺利举办&#xff0c;现场吸引了来自携程、小红书、米哈游、哔哩哔哩、波克城市、…...

Rapidio门铃消息FIFO溢出机制

关于RapidIO门铃消息FIFO的溢出机制及其与中断抖动的关系&#xff0c;以下是深入解析&#xff1a; 门铃FIFO溢出的本质 在RapidIO系统中&#xff0c;门铃消息FIFO是硬件控制器内部的缓冲区&#xff0c;用于临时存储接收到的门铃消息&#xff08;Doorbell Message&#xff09;。…...

初学 pytest 记录

安装 pip install pytest用例可以是函数也可以是类中的方法 def test_func():print()class TestAdd: # def __init__(self): 在 pytest 中不可以使用__init__方法 # self.cc 12345 pytest.mark.api def test_str(self):res add(1, 2)assert res 12def test_int(self):r…...

Python Ovito统计金刚石结构数量

大家好,我是小马老师。 本文介绍python ovito方法统计金刚石结构的方法。 Ovito Identify diamond structure命令可以识别和统计金刚石结构,但是无法直接输出结构的变化情况。 本文使用python调用ovito包的方法,可以持续统计各步的金刚石结构,具体代码如下: from ovito…...

wpf在image控件上快速显示内存图像

wpf在image控件上快速显示内存图像https://www.cnblogs.com/haodafeng/p/10431387.html 如果你在寻找能够快速在image控件刷新大图像&#xff08;比如分辨率3000*3000的图像&#xff09;的办法&#xff0c;尤其是想把内存中的裸数据&#xff08;只有图像的数据&#xff0c;不包…...

CVPR2025重磅突破:AnomalyAny框架实现单样本生成逼真异常数据,破解视觉检测瓶颈!

本文介绍了一种名为AnomalyAny的创新框架&#xff0c;该方法利用Stable Diffusion的强大生成能力&#xff0c;仅需单个正常样本和文本描述&#xff0c;即可生成逼真且多样化的异常样本&#xff0c;有效解决了视觉异常检测中异常样本稀缺的难题&#xff0c;为工业质检、医疗影像…...

系统掌握PyTorch:图解张量、Autograd、DataLoader、nn.Module与实战模型

本文较长&#xff0c;建议点赞收藏&#xff0c;以免遗失。更多AI大模型应用开发学习视频及资料&#xff0c;尽在聚客AI学院。 本文通过代码驱动的方式&#xff0c;系统讲解PyTorch核心概念和实战技巧&#xff0c;涵盖张量操作、自动微分、数据加载、模型构建和训练全流程&#…...

用鸿蒙HarmonyOS5实现中国象棋小游戏的过程

下面是一个基于鸿蒙OS (HarmonyOS) 的中国象棋小游戏的实现代码。这个实现使用Java语言和鸿蒙的Ability框架。 1. 项目结构 /src/main/java/com/example/chinesechess/├── MainAbilitySlice.java // 主界面逻辑├── ChessView.java // 游戏视图和逻辑├──…...