Debezium嵌入式连接postgresql封装服务

文章目录

      • 1.项目结构：
      • 2.依赖：
      • 3.application.properties
      • 4.DebeziumConnectorConfig类
      • 5.TableEnum类
      • 6.TableHandler接口（表处理抽象）
      • 7.DefaultTableHandler默认实现类
      • 8.UserTableHandler处理类
      • 9.TableHandlerFactory工厂
      • 10.DebeziumListener 监听事件
      • 11.测试

环境：JDK8，Debezium1.94，postgresql12

1.项目结构：

在这里插入图片描述

2.依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><artifactId>spring-boot-starter-parent</artifactId><groupId>org.springframework.boot</groupId><version>2.7.18</version></parent><groupId>com.linging</groupId><artifactId>springboot-debezium-server</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><debezium.version>1.9.4.Final</debezium.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>io.debezium</groupId><artifactId>debezium-embedded</artifactId><version>${debezium.version}</version><exclusions><exclusion><artifactId>slf4j-reload4j</artifactId><groupId>org.slf4j</groupId></exclusion></exclusions></dependency><dependency><groupId>io.debezium</groupId><artifactId>debezium-connector-postgres</artifactId><version>${debezium.version}</version></dependency><dependency><groupId>io.debezium</groupId><artifactId>debezium-api</artifactId><version>${debezium.version}</version></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId></dependency><dependency><groupId>com.alibaba.fastjson2</groupId><artifactId>fastjson2</artifactId><version>2.0.43</version></dependency></dependencies></project>

3.application.properties

# Debezium Configuration
#连接器基本信息
#指定 Debezium 连接器的名称,唯一标识，用于在 Kafka Connect 中区分不同的连接器。
debezium.name=my-postgres-connector
#指定连接器的类名，这里是连接postgresql
debezium.connector.class=io.debezium.connector.postgresql.PostgresConnector#偏移量相关
#指定偏移量存储的实现类，这里使用的是文件存储，将偏移量存储在本地文件中。
debezium.offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
#处理数据的偏移量存储路径
debezium.offset.storage.file.filename=C:\\Users\\Linging\\Desktop\\debezinum\\offsets_0.dat
#处理数据的偏移量提交时间间隔，单位毫秒，设置为 0 表示每次处理完一批记录后立即提交偏移量，这可以减少数据丢失的风险，但会增加系统开销
#以上一次提交时间开始计算
debezium.offset.flush.interval.ms=10000#数据库连接信息
debezium.database.hostname=192.168.159.103
debezium.database.port=15432
debezium.database.user=postgres
debezium.database.password=123456
#捕获变更的数据库名称
debezium.database.dbname=db_test
#指定逻辑服务器的唯一标识，用于区分不同的数据库实例
debezium.database.server.id=postgresql_0
#指定逻辑服务器的名称，用于在 Kafka 主题中区分不同的数据库实例
debezium.database.server.name=customer_postgres_db_server#数据库历史记录
#指定数据库模式记录的实现类，这里使用的是文件存储，将数据库历史记录存储在本地文件中
debezium.database.history=io.debezium.relational.history.FileDatabaseHistory
#指定数据库模式记录文件的路径
debezium.database.history.file.filename=C:\\Users\\Linging\\Desktop\\debezinum\\history_0.dat#表和字段过滤
#debezium.table.include.list=public.user
#debezium.column.include.list=public.user.id,public.user.name#其他配置
#指定是否自动创建 PostgreSQL 的逻辑复制槽，设置为 filtered 表示只捕获配置中指定的表和字段的变更
debezium.publication.autocreate.mode=filtered
#指定 PostgreSQL 的逻辑复制插件名称，pgoutput是 PostgreSQL 的逻辑复制插件名称，用于捕获变更
debezium.plugin.name=pgoutput
#指定逻辑复制槽的名称
debezium.slot.name=dbz_customerdb_listener
#不执行初始快照，直接捕获变更数据，取值：never、initial、when_needed
debezium.snapshot.mode=never
#批量提交条数
debezium.max.batch.size=100logging.level.root=INFO
logging.level.io.debezium.postgres.BinlogReader=INFO
logging.level.io.davidarhcanjo=DEBUG
logging.level.io.debezium=INFO

4.DebeziumConnectorConfig类

@Configuration
public class DebeziumConnectorConfig {@Beanpublic Properties customerConnector(Environment env) {Properties props = new Properties();props.setProperty("name", env.getProperty("debezium.name"));props.setProperty("connector.class", env.getProperty("debezium.connector.class"));props.setProperty("offset.storage", env.getProperty("debezium.offset.storage"));props.setProperty("offset.storage.file.filename", env.getProperty("debezium.offset.storage.file.filename"));props.setProperty("offset.flush.interval.ms", env.getProperty("debezium.offset.flush.interval.ms"));props.setProperty("database.hostname", env.getProperty("debezium.database.hostname"));props.setProperty("database.port", env.getProperty("debezium.database.port"));props.setProperty("database.user", env.getProperty("debezium.database.user"));props.setProperty("database.password", env.getProperty("debezium.database.password"));props.setProperty("database.dbname", env.getProperty("debezium.database.dbname"));props.setProperty("database.server.id", env.getProperty("debezium.database.server.id"));props.setProperty("database.server.name", env.getProperty("debezium.database.server.name"));props.setProperty("database.history", env.getProperty("debezium.database.history"));props.setProperty("database.history.file.filename", env.getProperty("debezium.database.history.file.filename"));props.setProperty("table.include.list", TableEnum.getTableNames()); //表名props.setProperty("column.include.list", TableEnum.getColumns()); // 表中得哪些字段props.setProperty("publication.autocreate.mode", env.getProperty("debezium.publication.autocreate.mode"));props.setProperty("plugin.name", env.getProperty("debezium.plugin.name"));props.setProperty("slot.name", env.getProperty("debezium.slot.name"));props.setProperty("snapshot.mode", env.getProperty("debezium.snapshot.mode"));props.setProperty("max.batch.size", env.getProperty("debezium.max.batch.size"));return props;}
}

5.TableEnum类

package com.linging.enums;import java.util.Arrays;
import java.util.stream.Collectors;/*** 监听的表及字段配置* @author Linging* @version 1.0.0* @since 1.0*/
public enum TableEnum {DEFAULT("default", "defaultTableHandler", null),USER("public.user", "userTableHandler", "public.user.id,public.user.name"),;// 表名称private final String tableName;// 表处理类的名称private final String handlerName;// 表的字段名称，多个用逗号隔开public final String columnName;TableEnum(String tableName, String handlerName, String columnName) {this.tableName = tableName;this.handlerName = handlerName;this.columnName = columnName;}public String getTableName() {return tableName;}public String getHandlerName() {return handlerName;}public String getColumnName() {return columnName;}public static String getTableNames(){return Arrays.stream(TableEnum.values()).map(TableEnum::getTableName).filter(name -> !"default".equals(name)).distinct().collect(Collectors.joining(","));}public static String getColumns(){return Arrays.stream(TableEnum.values()).filter(e -> !"default".equals(e.getTableName()) && e.getColumnName() != null).map(TableEnum::getColumnName).distinct().collect(Collectors.joining(","));}
}

6.TableHandler接口（表处理抽象）

public interface TableHandler {void handle(SourceRecord sourceRecord);void handleBatch(List<RecordChangeEvent<SourceRecord>> recordChangeEvents, DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer);
}

7.DefaultTableHandler默认实现类

/*** 默认处理类*/
@Component("defaultTableHandler")
public class DefaultTableHandler implements TableHandler {private static final Logger log = LoggerFactory.getLogger(DefaultTableHandler.class);@Overridepublic void handle(SourceRecord sourceRecord) {log.info("Handling default table: {}", sourceRecord.topic());log.info("Key = {}, Value = {}", sourceRecord.key(), sourceRecord.value());}@Overridepublic void handleBatch(List<RecordChangeEvent<SourceRecord>> recordChangeEvents,DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer) {log.info("Handling batch default table: {}", recordChangeEvents.size());}
}

8.UserTableHandler处理类

/*** user表变更处理类*/
@Component("userTableHandler")
public class UserTableHandler implements TableHandler {private static final Logger log = LoggerFactory.getLogger(UserTableHandler.class);@Overridepublic void handle(SourceRecord sourceRecord) {log.info("Handling user table: {}", sourceRecord.topic());log.info("Key = {}, Value = {}", sourceRecord.key(), sourceRecord.value());// 添加具体的处理逻辑Struct sourceRecordChangeValue= (Struct) sourceRecord.value();if (sourceRecordChangeValue != null) {Envelope.Operation operation = Envelope.Operation.forCode((String) sourceRecordChangeValue.get(OPERATION));// 处理非读操作if(operation != Envelope.Operation.READ) {String record = operation == Envelope.Operation.DELETE ? BEFORE : AFTER;Struct struct = (Struct) sourceRecordChangeValue.get(record);Map<String, Object> payload = struct.schema().fields().stream().map(Field::name).filter(fieldName -> struct.get(fieldName) != null).map(fieldName -> Pair.of(fieldName, struct.get(fieldName))).collect(toMap(Pair::getKey, Pair::getValue));// TODO 处理逻辑(保存数据库，发送MQ等操作，需要保证幂等)log.info("Updated Data: {} with Operation: {}", payload, operation.name());}}}@Overridepublic void handleBatch(List<RecordChangeEvent<SourceRecord>> recordChangeEvents,DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer) {for (RecordChangeEvent<SourceRecord> recordChangeEvent : recordChangeEvents) {try {SourceRecord sourceRecord = recordChangeEvent.record();// TODO 处理逻辑(保存数据库，发送MQ等操作，需要保证幂等)this.handle(sourceRecord);// 标记已处理committer.markProcessed(recordChangeEvent);} catch (InterruptedException e) {log.error("处理异常：", e);}}}
}

9.TableHandlerFactory工厂

@Component
public class TableHandlerFactory implements ApplicationContextAware {@Value("${debezium.database.server.name}")private String prefixServerName;private ApplicationContext context;private final Map<String, TableHandler> handlers = new HashMap<>();@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.context = applicationContext;}@PostConstructpublic void init(){String name = null;for (TableEnum tableEnum : TableEnum.values()) {if(TableEnum.DEFAULT.equals(tableEnum)){name = tableEnum.getTableName();}else{name = getTableName(tableEnum.getTableName());}handlers.putIfAbsent(name,context.getBean(tableEnum.getHandlerName(), TableHandler.class));}}public TableHandler getHandler(String tableName) {return handlers.getOrDefault(tableName, handlers.get(TableEnum.DEFAULT.getTableName()));}public String getTableName(String name){return prefixServerName + "." + name;}}

10.DebeziumListener 监听事件

@Component
public class DebeziumListener {private static final Logger log = LoggerFactory.getLogger(DebeziumListener.class);private final ExecutorService executor = Executors.newSingleThreadExecutor();private final DebeziumEngine<RecordChangeEvent<SourceRecord>> debeziumEngine;private final TableHandlerFactory tableHandlerFactory;@Autowiredpublic DebeziumListener(Properties customerConnector, TableHandlerFactory tableHandlerFactory) {this.debeziumEngine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class)).using(customerConnector).using(OffsetCommitPolicy.periodic(customerConnector)).notifying(this::handleChangeEventBatch).build();this.tableHandlerFactory = tableHandlerFactory;}/*** 批量记录处理* @param recordChangeEvents* @param committer*/private void handleChangeEventBatch(List<RecordChangeEvent<SourceRecord>> recordChangeEvents,DebeziumEngine.RecordCommitter<RecordChangeEvent<SourceRecord>> committer){// 根据表分组Map<String, List<RecordChangeEvent<SourceRecord>>> tableName2List = recordChangeEvents.stream().collect(Collectors.groupingBy(event -> event.record().topic()));tableName2List.forEach((tableName, recordChangeEventList) -> {TableHandler handler = tableHandlerFactory.getHandler(tableName);handler.handleBatch(recordChangeEventList, committer);});try {// 触发提交策略committer.markBatchFinished();} catch (InterruptedException e) {log.error("提交异常：", e);}}/*** 单条记录处理* @param sourceRecordChangeEvent*/private void handleChangeEvent(RecordChangeEvent<SourceRecord> sourceRecordChangeEvent) {SourceRecord sourceRecord = sourceRecordChangeEvent.record();log.info("Key = {}, Value = {}", sourceRecord.key(), sourceRecord.value());String tableName = sourceRecord.topic();// 获取对应的表处理类TableHandler handler = tableHandlerFactory.getHandler(tableName);handler.handle(sourceRecord);}@PostConstructprivate void start() {this.executor.execute(debeziumEngine);}@PreDestroyprivate void stop() {if (this.debeziumEngine != null) {try {this.debeziumEngine.close();} catch (IOException e) {log.error("关闭debeziumEngine异常：", e);}}this.executor.shutdown();}
}

11.测试

启动服务，修改数据库user表数据：
在这里插入图片描述
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/bicheng/75881.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

ER-图，详情和画法

ER-图，详情和画法

一、E-R图的核心元素 1.实体 表示现实中对象或概念&#xff0c;用矩形表示 示例&#xff1a;用户、老师、学生 2.属性 描述实体的特征&#xff0c;用椭圆表示。 分为主键&#xff08;用户id&#xff09; 和非主键&#xff08;用户昵称&#xff09; 3.关系 表示实体间的…
阅读更多...
Windows Flip PDF Plus Corporate PDF翻页工具

Windows Flip PDF Plus Corporate PDF翻页工具

软件介绍 Flip PDF Plus Corporate是一款功能强大的PDF翻页工具&#xff0c;也被称为名编辑电子杂志大师。这款软件能够迅速将PDF文件转换为具有翻页动画效果的电子书&#xff0c;同时保留原始的超链接和书签。无论是相册、视频、音频&#xff0c;还是Flash、视频和链接&#…
阅读更多...
Linux文件系统中的Page Cache和内存管理中的Page之间的关系

Linux文件系统中的Page Cache和内存管理中的Page之间的关系

Linux文件系统中的Page Cache和内存管理中的Page之间有密切的关联&#xff0c;两者在底层机制上紧密结合&#xff0c;共同实现高效的内存和文件系统管理。以下是它们的关系和关键点&#xff1a; 核心关系 Page Cache的底层是内存Page Page Cache是由内存管理中的物理内存页&…
阅读更多...
每日一个小病毒（C++）EnumChildWindows+shellcode

每日一个小病毒（C++）EnumChildWindows+shellcode

这里写目录标题 1. `EnumChildWindows` 的基本用法2. 如何利用 `EnumChildWindows` 执行 Shellcode？关键点：完整 Shellcode 执行示例3. 为什么 `EnumChildWindows` 能执行 Shellcode？4. 防御方法5. 总结EnumChildWindows 是 Windows API 中的一个函数，通常用于枚举所有子窗…
阅读更多...
AI爬虫？爬！

AI爬虫？爬！

1.你是否还在为大模型的key而感到忧伤和囊中羞涩&#xff0c;openrouter.ai&#xff0c;目前可免费白嫖多个大模型&#xff0c;代码如下 from openai import OpenAIclient OpenAI(base_url"https://openrouter.ai/api/v1",api_key"", )completion clien…
阅读更多...
洛谷题单3-P5720 【深基4.例4】一尺之棰-python-流程图重构

洛谷题单3-P5720 【深基4.例4】一尺之棰-python-流程图重构

题目描述 《庄子》中说到&#xff0c;“一尺之棰&#xff0c;日取其半&#xff0c;万世不竭”。第一天有一根长度为 a a a 的木棍&#xff0c;从第二天开始&#xff0c;每天都要将这根木棍锯掉一半&#xff08;每次除 2 2 2&#xff0c;向下取整&#xff09;。第几天的时候木…
阅读更多...
c++中的auto关键字

c++中的auto关键字

在 C 中&#xff0c;auto 是一个类型推断关键字&#xff08;C11 引入&#xff09;&#xff0c;允许编译器根据变量的初始化表达式自动推导其类型。它极大地简化了代码编写&#xff0c;尤其在涉及复杂类型或模板的场景中。以下是 auto 的详细说明&#xff1a; 1. 基本用法 1.1 …
阅读更多...
开发指南111-关闭所有打开的子窗口

开发指南111-关闭所有打开的子窗口

门户系统是通过window.open通过单点登录的模式打开子系统的&#xff0c;这就要求门户系统退出时&#xff0c;关闭所有打开的子系统。 平台处理这一问题的核心原理如下&#xff1a; 主窗口定义&#xff1a; allChildWindows:[], //所有子窗口 pushChildWindow(childWindow){ …
阅读更多...
Kotlin语言进阶：协程、Flow、Channel详解（二）

Kotlin语言进阶：协程、Flow、Channel详解（二）

Kotlin语言进阶：协程、Flow、Channel详解（二） 一、Flow基础 1.1 什么是Flow Flow是Kotlin提供的用于处理异步数据流的解决方案，它建立在协程之上，具有以下特点： 冷流特性：只有在收集时才会开始发射数据背压处理：自动处理生产者和消费者速度不匹配的问题组合操作：提…
阅读更多...
mysql中my.cnf权限不能过大。否则无法生效

mysql中my.cnf权限不能过大。否则无法生效

mysql 报错 World-writable config file ‘/etc/my.cnf‘ is ignored. /etc/my.cnf 配置文件, 或着docker 挂载的配置文件(宿主机中的配置文件),权限过大 如是二进制启动 chmod 644 /etc/my.cnf 如是docker启动 chmod 644 /opt/docker-data/mysql/conf/my.cnf 重启服务,就可…
阅读更多...
Spring 中的 @Autowired 和 @Resource

Spring 中的 @Autowired 和 @Resource

&#x1f9e9; 一、Autowired 和 Resource 的基本作用 注解来源作用AutowiredSpring 提供&#xff08;org.springframework.beans.factory.annotation.Autowired&#xff09;按类型 自动注入ResourceJDK 提供&#xff08;javax.annotation.Resource&#xff09;默认按名称 注入…
阅读更多...
anomalib—2—输入图像大小调整

anomalib—2—输入图像大小调整

三个地方 第一&#xff1a;在定义model时&#xff0c;要在pre_processor里面去定义一个前处理&#xff0c;前处理就一个功能&#xff0c;定义图像的大小 pre_processor0 Patchcore.configure_pre_processor( image_size (128, 128)) model Patchcore( backbone"wide_r…
阅读更多...
JSX、支持HTML标签、Ref的使用、虚拟DOM的使用

JSX、支持HTML标签、Ref的使用、虚拟DOM的使用

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》、《前端求职突破计划》 &#x1f35a; 蓝桥云课签约作者、…
阅读更多...
JVM 学习计划表（2025 版）

JVM 学习计划表（2025 版）

JVM 学习计划表&#xff08;2025 版&#xff09; &#x1f4da; 基础阶段&#xff08;2 周&#xff09; 1. JVM 核心概念 ​JVM 作用与体系结构 理解 JVM 在 Java 跨平台运行中的核心作用&#xff0c;掌握类加载子系统、运行时数据区、执行引擎的交互流程​内存结构与数据存…
阅读更多...
进程内存分布--之理论知识

进程内存分布--之理论知识

一个由C/C编译的程序占用的内存分为以下几个部分 &#xff1a; 1、栈区&#xff08;stack&#xff09;&#xff1a;由编译器自动分配释放 &#xff0c;存放函数调用函数的参数值&#xff0c;局部变量的值等。其操作方式类似于数据结构中的栈。 2、堆区&#xff08;heap&#xf…
阅读更多...
WEB安全--内网渗透--LMNTLM基础

WEB安全--内网渗透--LMNTLM基础

一、前言 LM Hash和NTLM Hash是Windows系统中的两种加密算法&#xff0c;不过LM Hash加密算法存在缺陷&#xff0c;在Windows Vista 和 Windows Server 2008开始&#xff0c;默认情况下只存储NTLM Hash&#xff0c;LM Hash将不再存在。所以我们会着重分析NTLM Hash。 在我们内…
阅读更多...
时尚优雅奢华品牌包装徽标设计衬线英文字体安装包 Kagea – Luxury Women Ligature Font

时尚优雅奢华品牌包装徽标设计衬线英文字体安装包 Kagea – Luxury Women Ligature Font

Kagea 是一种复杂的衬线字体&#xff0c;有常规和压缩两种样式&#xff0c;具有 50 连字和替代字体&#xff0c;并支持多种语言。其精致的比例和别致的字体使其成为高端品牌、编辑布局、高档包装、时尚品牌和奢侈品设计的理想选择。 包含内容&#xff1a; TTF、OTF 和 WOFF 格…
阅读更多...
cut命令用法

cut命令用法

cut 是 Linux/Unix 系统中一个用于按列提取文本内容的命令&#xff0c;常用于处理结构化文本&#xff08;如 CSV、日志、配置文件等&#xff09;。它通过分隔符、字符位置或字节位置来切割文本&#xff0c;提取指定部分。 核心功能 按字段&#xff08;列&#xff09;提取&#…
阅读更多...
美国mlb与韩国mlb的关系·棒球9号位

美国mlb与韩国mlb的关系·棒球9号位

MLB&#xff08;Major League Baseball&#xff0c;美国职业棒球大联盟&#xff09;作为全球最高水平的职业棒球联赛&#xff0c;与韩国市场流行的“MLB”时尚品牌之间存在着授权合作关系&#xff0c;但两者在业务范畴和品牌定位上存在显著差异。 一、品牌授权背景&#xff1a;…
阅读更多...
从吉卜力漫画到艺术创造：GPT-4o多种风格绘图Prompt大全

从吉卜力漫画到艺术创造：GPT-4o多种风格绘图Prompt大全

在3月底&#xff0c;GPT-4o掀起了一阵吉卜力绘图浪潮&#xff0c;大家纷纷输入一张图片&#xff0c;让4o模型进行风格化迁移&#xff0c;其中吉卜力风格的漫画在社交媒体上最为火热。在大家争议4o的训练数据是否侵权和4o背后的技术原理的时候&#xff0c;我们先来玩一玩&#x…
阅读更多...
最新文章

Copyright @ 2022~2023