Debezium日常分享系列之：定制Debezium 信号发送和通知

Debezium日常分享系列之：定制Debezium 信号发送和通知

  • 一、自定义信号和通知通道
  • 二、结论

Debezium 2.3 在信号和通知功能方面引入了新的改进。除了 Debezium 提供的预定义信号和通知通道之外，您还可以设置新的信号和通知通道。此功能使用户能够自定义系统以满足他们的独特需求，并将其与现有基础设施或第三方解决方案相结合。它通过精确捕获和传达信号事件并通过首选渠道触发通知，实现对数据变化的有效监控和主动响应。

Debezium日常分享系列之：Debezium 信号发送和通知 - 第 1 部分

一、自定义信号和通知通道

在 Debezium 中，可以自定义信号和通知通道以满足特定要求。例如，我们可以通过为信号和通知创建 HTTP 通道来实现自定义。此 HTTP 通道从 http 端点接收信号，并且可以在信号传送后将通知发送回端点。

让我们探索一个示例，演示如何使用 Debezium Postgres 连接器、发送信号的模拟服务器以及通过 http 端点接收通知的 Postbin 来创建和利用 HTTP 信号和通知通道。

设置 HTTP 信号通道：

  • 将 Debezium Postgres 连接器配置为在发生相关数据库更改时接收信号。
  • 设置服务以使用 HTTP 通道向 Debezium 发送信号。该服务可以是数据库、第三方应用程序或任何其他可以发送 http 请求的系统。在此示例中，我们将使用模拟服务器向 Debezium 发送信号。 Mock Server 是一个可用于模拟 http 请求和响应的服务。
  • 配置模拟服务器以使用适当的 HTTP 方法（例如 POST）通过 http 端点发送信号。
  • 根据需要自定义 HTTP 通道设置以定义 http 端点 URL、身份验证、标头和任何其他参数。

设置 HTTP 通知通道：

  • 一旦 Debezium 接收并处理信号，它就可以触发向 http 端点发布通知。在此示例中，我们将使用 HTTP 通道将通知发送到 Postbin bin。 Postbin是一个可以用来接收http请求并查看请求详细信息的服务。
  • 自定义通知的 HTTP 通道设置，在 Postbin 中创建 bin，并根据需要定义 http 端点 URL、身份验证、标头和任何其他参数。
  • 使用适当的 HTTP 方法（例如 POST）将通知事件转发到 http 端点，即 Postbin bin。可以根据需要自定义通知负载。

博客文章中此示例的完整源代码在 Debezium 示例存储库的 http-signal-notification 目录下提供。

创建一个 java 项目来构建 HTTP 信号和通知通道。运行以下命令使用 Maven 创建一个新的 java 项目：

mvn archetype:generate
    -DgroupId=io.debezium.examples
    -DartifactId=http-signaling-notification

将以下依赖项添加到 Debezium 版本（2.3 及更高版本）的 pom.xml 文件中：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-core</artifactId>
    <version>2.3.0.Final</version>
</dependency>

要使用模拟服务器接收信号，请创建定义模拟服务器服务的 Docker Compose 文件。模拟服务器服务的配置如下：

services:
  mockServer:
    image: mockserver/mockserver:latest
    ports:
      - 1080:1080
    environment:
      - MOCKSERVER_WATCH_INITIALIZATION_JSON=true
      - MOCKSERVER_INITIALIZATION_JSON_PATH=/config/initializerJson.json
    volumes:
        - ./initializerJson.json:/config/initializerJson.json

设置环境变量 MOCKSERVER_WATCH_INITIALIZATION_JSON 和 MOCKSERVER_INITIALIZATION_JSON_PATH 以使模拟服务器能够监视初始化 JSON 文件中的更改并指定其路径。包含信号的http请求和响应信息的initializerJson.json文件被安装到模拟服务器容器中。

initializerJson.json 文件定义了一个对路径 /api/signal 的模拟 http 请求，其中查询字符串参数 code=10969。当模拟服务器收到此请求时，它将使用包含 id、类型和数据的 JSON 正文进行响应。响应的状态码为200，表示响应成功。 initializerJson.json文件的定义如下：

[
  {
    "httpRequest" : {
      "method" : "GET",
      "path" : "/api/signal",
      "queryStringParameters" : {
        "code" : ["10969"]
      }
    },
    "httpResponse" : {
      "body": "{\"id\":\"924e3ff8-2245-43ca-ba77-2af9af02fa07\",\"type\":\"log\",\"data\":{\"message\": \"Signal message received from http endpoint.\"}}",
      "statusCode": 200
    }
  }
]
  • id ：标识信号实例的任意唯一字符串。
  • type ：要发送的信号类型。在此示例中，类型为日志，它请求连接器将条目添加到连接器的日志文件中。处理信号后，连接器会在日志中打印指定的消息。
  • data ：传递给信号事件的 JSON 格式的参数。在此示例中，消息参数被传递给信号事件。

通过实现SignalChannelReader接口创建HTTP信号通道，如下所示：

public class HttpSignalChannel implements SignalChannelReader {
    private static final Logger LOGGER = LoggerFactory.getLogger(HttpSignalChannel.class);

    public static final String CHANNEL_NAME = "http";
    private static final List<SignalRecord> SIGNALS = new ArrayList<>();
    public CommonConnectorConfig connectorConfig;

        @Override
    public String name() { (1)
        return CHANNEL_NAME;
    }

    @Override
    public void init(CommonConnectorConfig connectorConfig) { (2)
        this.connectorConfig = connectorConfig;
    }

    @Override
    public List<SignalRecord> read() { (3)
        try {
            String requestUrl = "http://mockServer:1080/api/signal?code=10969";

            // send http request to the mock server
            HttpClient httpClient = HttpClient.newHttpClient();
            HttpRequest request = HttpRequest.newBuilder()
                    .uri(URI.create(requestUrl))
                    .GET()
                    .header("Content-Type", "application/json")
                    .build();

            // read the response
            HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
           if (response.statusCode() == 200) {
               ObjectMapper mapper = new ObjectMapper();
               String responseBody = response.body();

               // parse the response body
               JsonNode signalJson = mapper.readTree(responseBody);
               Map<String, Object> additionalData = signalJson.has("additionalData") ? mapper.convertValue(signalJson.get("additionalData"), new TypeReference<>() {}) : new HashMap<>();
               String id = signalJson.get("id").asText();
               String type = signalJson.get("type").asText();
               String data = signalJson.get("data").toString();
               SignalRecord signal = new SignalRecord(id, type, data, additionalData);

               LOGGER.info("Recorded signal event '{}' ", signal);

               // process the signal
               SIGNALS.add(signal);
                } else {
                    LOGGER.warn("Error while reading signaling events from endpoint: {}", response.statusCode());
                }
            } catch (IOException | InterruptedException e) {
                LOGGER.warn("Exception while preparing to process the signal '{}' from the endpoint", e.getMessage());
                e.printStackTrace();
            }
        return SIGNALS;
        }

    @Override
    public void close() { (4)
       SIGNALS.clear();
    }
}
  1. name() 方法返回信号通道的名称。要使 Debezium 能够使用通道，请在连接器的 signal.enabled.channels 属性中指定名称 http。
  2. init() 方法可用于初始化 http 通道所需的特定配置、变量或连接。
  3. read() 方法从 http 端点读取信号并返回将由 Debezium 连接器处理的 SignalRecord 对象列表。
  4. close() 方法关闭所有分配的资源。

通过实现NotificationChannel接口创建通知通道，如下所示：

public class HttpNotificationChannel implements NotificationChannel {
    private static final Logger LOGGER = LoggerFactory.getLogger(HttpNotificationChannel.class);

    public static final String CHANNEL_NAME = "http";
    private static final String NOTIFICATION_PREFIX = "[HTTP NOTIFICATION SERVICE]";

    @Override
    public String name() { (1)
        return CHANNEL_NAME;
    }

    @Override
    public void init(CommonConnectorConfig config) { (2)
        // custom configuration
    }

    @Override
    public void send(Notification notification) { (3)
        LOGGER.info(String.format("%s Sending notification to http channel", NOTIFICATION_PREFIX));
        String binId = createBin();
        sendNotification(binId, notification);
    }

    private static String createBin()  {
        // Create a bin on the server
        try {
            HttpRequest request = HttpRequest.newBuilder()
                    .uri(new URI("https://www.toptal.com/developers/postbin/api/bin"))
                    .POST(HttpRequest.BodyPublishers.ofString(" "))
                    .build();

            HttpClient httpClient = HttpClient.newHttpClient();
            HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());

            if (response.statusCode() == HTTP_CREATED) {
                String binId = response.body().replaceAll(".*\"binId\":\"([^\"]+)\".*", "$1");
                LOGGER.info("Bin created: " + response.body());
                return binId;
            }
        } catch (URISyntaxException | InterruptedException | IOException e) {
            throw new RuntimeException(e);
        }
        return null;
    }

    private static void sendNotification (String binId, Notification notification) {
        // Get notification from the bin
        try {
            ObjectMapper mapper = new ObjectMapper();
            String notificationString = mapper.writeValueAsString(notification);
            HttpRequest request = HttpRequest.newBuilder()
                    .uri(new URI("https://www.toptal.com/developers/postbin/" + binId))
                    .header("Content-Type", "application/json")
                    .POST(HttpRequest.BodyPublishers.ofString(notificationString))
                    .build();

            HttpClient httpClient = HttpClient.newHttpClient();
            HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());

            if (response.statusCode() == HTTP_OK) {
                LOGGER.info("Notification received : " + response.body());
            }
        } catch (URISyntaxException | InterruptedException | IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void close() { (4)
    }
}
  1. name() 方法返回通知通道的名称。要使 Debezium 能够使用通道，请在连接器的 notification.enabled.channels 属性中指定 http。
  2. init() 方法可用于初始化通道所需的特定配置、变量或连接。
  3. send() 方法将通知发送到通道。该通知包含由 Debezium 连接器处理的 SignalRecord 对象。
  4. close() 方法关闭所有分配的资源。

分别在 META-INF/services 目录下的 io.debezium.pipeline.signal.SignalChannelReader 和 io.debezium.pipeline.notification.channels.NotificationChannel 文件下声明 HTTP 信号和通知通道。

编译 Java 项目并将其导出为 JAR 文件。这可以使用 Maven 或您喜欢的构建工具来完成。将 JAR 文件复制到包含要使用的 Debezium 连接器的 JAR 文件的目录。例如，如果您想要将自定义信号和通知通道与 Debezium Postgres 连接器一起使用，请将 JAR 文件复制到 /kafka/connect/debezium-connector-postgres 目录。

此示例提供了一个 Docker Compose 文件，其中定义了必要的服务，包括 Mock Server、Zookeeper、Kafka Connect 和 Postgres 数据库。

要启动服务，请运行以下命令：

export DEBEZIUM_VERSION=2.3
docker-compose up -d

确保服务已启动并正在运行，并且 Postgres 数据库已准备好接受连接后，下一步是注册连接器。这涉及创建连接器配置文件。让我们创建一个名为 register-postgres.json 的文件，其中包含以下属性：

{
  "name": "inventory-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": 1,
    "database.hostname": "postgres",
    "database.port": 5432,
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname" : "postgres",
    "topic.prefix": "dbserver1",
    "schema.include.list": "inventory",
    "signal.enabled.channels": "http", 1
    "notification.enabled.channels": "http" 2
  }
}
  1. signal.enabled.channels 属性指定连接器要使用的信号通道。在这种情况下，连接器使用 http 信号通道。
  2. notification.enabled.channels 属性指定连接器要使用的通知通道。在这种情况下，连接器使用 http 通知通道。

现在我们已经准备好了连接器配置文件，我们可以通过执行以下命令来向 Kafka Connect 注册连接器：

curl -i -X POST -H "Accept:application/json" \
    -H  "Content-Type:application/json" http://localhost:8083/connectors/ \
    -d @register-postgres.json

连接器成功注册后，您可以查看连接器日志以观察信号事件。这些日志提供了有关连接器的处理和进度的见解，包括任何与信号相关的信息。您将遇到类似于以下内容的日志消息：

Recorded signal event 'SignalRecord{id='924e3ff8-2245-43ca-ba77-2af9af02fa07', type='log', data='{"message":"Signal message received from http endpoint."}', additionalData={}}'    [io.debezium.examples.signal.HttpSignalChannel]

此外，您可能会注意到与发送到邮筒的通知事件相关的日志消息。例如：

[HTTP NOTIFICATION SERVICE] Sending notification to http channel   [io.debezium.examples.notification.HttpNotificationChannel]
Bin created: {"binId":"1688742588469-1816775151528","now":1688742588470,"expires":1688744388470}   [io.debezium.examples.notification.HttpNotificationChannel]

它提供有关通知事件的信息，例如创建具有唯一标识符 (binId) 的 bin 以及其他相关详细信息。要从 Postbin 检索通知事件，请从日志消息中获取 binId 并使用它从 Postbin 请求相应的通知事件。要查看通知事件，您可以使用以下 URL 访问 Postbin：https://www.toptal.com/developers/postbin/b/:binId。将 URL 中的 :binId 替换为从连接器日志中获取的实际 binId。

发送到 Postbin 的通知事件如下所示：

在这里插入图片描述

二、结论

在本教程中，我们探讨了如何为 Debezium 连接器创建自定义信号和通知通道。我们创建了一个自定义信号通道，用于从 HTTP 端点接收信号事件。我们还创建了一个自定义通知通道，用于将通知事件发送到 HTTP 端点。

Debezium 的综合信号和通知系统可与第三方解决方案无缝集成，使用户能够随时了解 Debezium 连接器的状态和进度。该系统的可扩展性使用户能够自定义信号和通知渠道，以满足他们的定制需求。

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

微服务——服务异步通讯RabbitMQ

微服务——服务异步通讯RabbitMQ

前置文章 消息队列——RabbitMQ基本概念容器化部署和简单工作模式程序_北岭山脚鼠鼠的博客-CSDN博客 消息队列——rabbitmq的不同工作模式_北岭山脚鼠鼠的博客-CSDN博客 消息队列——spring和springboot整合rabbitmq_北岭山脚鼠鼠的博客-CSDN博客 目录 Work queues 工作队列…
阅读更多...
JS——输入输出语法数组的操作

JS——输入输出语法数组的操作

JavaScript输入输出语法 目标&#xff1a;能写出常见的JavaScript输入输出语法 输出语法 语法1&#xff1a; document.write(要输出的内容)作用&#xff1a; 向body内输出内容 注意&#xff1a; 如果输出的内容写的是标签&#xff0c;也会被解析成网页元素 语法2&#xff1a…
阅读更多...
Verilog语法学习——LV9_使用子模块实现三输入数的大小比较

Verilog语法学习——LV9_使用子模块实现三输入数的大小比较

LV9_使用子模块实现三输入数的大小比较 题目来源于牛客网 [牛客网在线编程_Verilog篇_Verilog快速入门 (nowcoder.com)](https://www.nowcoder.com/exam/oj?page1&tabVerilog篇&topicId301) 题目 描述 在数字芯片设计中&#xff0c;通常把完成特定功能且相对独立的…
阅读更多...
特殊矩阵的压缩存储

特殊矩阵的压缩存储

1 数组的存储结构 1.1 一维数组 各数组元素大小相同&#xff0c;且物理上连续存放。第i个元素的地址位置是&#xff1a;a[i] LOC i*sizeof(ElemType) (LOC为起始地址) 1.2 二维数组 对于多维数组有行优先、列优先的存储方法 行优先&#xff1a;先行后列&#xff0c;先存储…
阅读更多...
无涯教程-jQuery - Select menu组件函数

无涯教程-jQuery - Select menu组件函数

小部件选择菜单功能可与JqueryUI中的小部件一起使用&#xff0c;它提供了可替换样式的选择元素。一个简单的选择菜单如下所示。 Select menu - 语法 $( "#menu" ).selectmenu(); Select menu - 示例 以下是显示选择菜单用法的简单示例- <!doctype html> &…
阅读更多...
自动驾驶感知系统-全球卫星定位系统

自动驾驶感知系统-全球卫星定位系统

卫星定位系统 车辆定位是让无人驾驶汽车获取自身确切位置的技术&#xff0c;在自动驾驶技术中定位担负着相当重要的职责。车辆自身定位信息获取的方式多样&#xff0c;涉及多种传感器类型与相关技术。自动驾驶汽车能够持续安全可靠运行的一个关键前提是车辆的定位系统必须实时…
阅读更多...
Go语言进阶 + 依赖管理

Go语言进阶 + 依赖管理

依赖配置 - version开始&#xff0c;就开始很难听懂了&#xff0c;需要结合很多课后配套资料查阅很多文档和网站....然而好像没有那么多时间&#xff0c;一天给3小时学Go真的顶天了.....还有算法和Linux的Mysql... 这几天学Go已经把算法给挤掉了.....下步要权衡一下&#xff0c…
阅读更多...
Centos7中实现脚本使用mysqldump实现分库分表备份

Centos7中实现脚本使用mysqldump实现分库分表备份

脚本 #!/bash/bin userroot #用户名 password123456 #密码 back_path/backup/db databases_file/backup/databases.list [ -f $databases_file ] || touch /backup/databases.list if [[ ! -s ${databases_file} ]] then while read line do[ -d ${back_path}/$line ] …
阅读更多...
ERROR in unable to locate ‘***/public/**/*‘ glob

ERROR in unable to locate ‘***/public/**/*‘ glob

前提 自己搭了一个react项目的脚手架&#xff0c;npm包下载一切都很正常&#xff0c;启动的时候突然就报ERROR in unable to locate ***/public/**/* glob这个错误&#xff0c;根据百度分析了一下产生的原因&#xff1a;webpack配置文件中的CopyWebpackPlugin导致的 网上给出的…
阅读更多...
C语言指针应该这么学？

C语言指针应该这么学？

数组名的意义&#xff1a; 1. sizeof(数组名)&#xff0c;这里的数组名表示整个数组&#xff0c;计算的是整个数组的大小。 2. &数组名&#xff0c;这里的数组名表示整个数组&#xff0c;取出的是整个数组的地址。 3. 除此之外所有的数组名都表示首元素的地址。 根据以上数…
阅读更多...
用asp.net开发h5网页版视频播放网站,类似优酷,jellyfin,emby

用asp.net开发h5网页版视频播放网站,类似优酷,jellyfin,emby

之前用jellyfin开源软件搞了一个视频播放服务器,用来共享给家里人看电影和电视剧,jellyfin虽然各方面功能都很强大,但是界面和使用习惯都很不适合,于是就想着利用下班休息时间做一套自己喜欢的视频网站出来. 本来是打算直接用jellyfin的源码进行修改,源码是用C# netcore 写的服…
阅读更多...
【C++进阶：哈希--unordered系列的容器及封装】

【C++进阶：哈希--unordered系列的容器及封装】

本课涉及到的所有代码都见以下链接&#xff0c;欢迎参考指正&#xff01; practice: 课程代码练习 - Gitee.comhttps://gitee.com/ace-zhe/practice/tree/master/Hash unordered系列关联式容器 在C98中&#xff0c;STL提供了底层为红黑树结构的一系列关联式容器&#xff0c;在…
阅读更多...
使用Pytest生成HTML测试报告

使用Pytest生成HTML测试报告

背景 最近开发有关业务场景的功能时&#xff0c;涉及的API接口比较多&#xff0c;需要自己模拟多个业务场景的自动化测试&#xff08;暂时不涉及性能测试&#xff09;&#xff0c;并且在每次测试完后能够生成一份测试报告。 考虑到日常使用Python自带的UnitTest&#xff0c;所…
阅读更多...
观察者模式与观察者模式实例EventBus

观察者模式与观察者模式实例EventBus

什么是观察者模式 顾名思义&#xff0c;观察者模式就是在多个对象之间&#xff0c;定义一个一对多的依赖&#xff0c;当一个对象状态改变时&#xff0c;所有依赖这个对象的对象都会自动收到通知。 观察者模式也称为发布订阅模式(Publish-Subscribe Design Pattern)&#xff0…
阅读更多...
ViT-vision transformer

ViT-vision transformer

ViT-vision transformer 介绍 Transformer最早是在NLP领域提出的&#xff0c;受此启发&#xff0c;Google将其用于图像&#xff0c;并对分类流程作尽量少的修改。 起源&#xff1a;从机器翻译的角度来看&#xff0c;一个句子想要翻译好&#xff0c;必须考虑上下文的信息&…
阅读更多...
【Linux后端服务器开发】MAC地址与其他重要协议

【Linux后端服务器开发】MAC地址与其他重要协议

目录 一、以太网 二、MAC地址 三、MTU 四、ARP协议 五、DNS系统 六、ICMP协议 七、NAT技术 八、代理服务器 一、以太网 “以太网”不是一种具体的网路&#xff0c;而是一种技术标准&#xff1a;既包含了数据链路层的内容&#xff0c;也包含了一些物理层的内容&#xf…
阅读更多...
k8s容器日志收集方案

k8s容器日志收集方案

背景 由于以下容器本身特性和现有日志采集工具的缺陷&#xff0c;开发者在收集Kubernetes分布式集群日志时常常遇到困扰&#xff1a; 容器本身特性&#xff1a; 采集目标多&#xff1a;容器本身的特性导致采集目标多&#xff0c;需要采集容器内日志、容器stdout。对于容器…
阅读更多...
实验六 调度器-实验部分

实验六 调度器-实验部分

目录 一、知识点 1.进程调度器设计的目标 1.1.进程的生命周期 1.2.用户进程创建与内核进程创建 1.3.进程调度器的设计目标 2.ucore 调度器框架 2.1.调度初始化 2.2.调度过程 2.2.1.调度整体流程 2.2.2.设计考虑要点 2.2.3.数据结构 2.2.4.调度框架应与调度算法无关…
阅读更多...
Zabbix分布式监控快速入门

Zabbix分布式监控快速入门

目录 1 Zabbix简介1.1 软件架构1.2 版本选择1.3 功能特性 2 安装与部署2.1 时间同步需求2.2 下载仓库官方源2.3 Zabbix-Server服务端的安装2.3.1 安装MySQL2.3.1.1 创建Zabbix数据库2.3.1.2 导入Zabbix库的数据文件 2.3.2 配置zabbix_server.conf2.3.3 开启Zabbix-Server服务2.…
阅读更多...
ElementUI Select选择器如何根据value值显示对应的label

ElementUI Select选择器如何根据value值显示对应的label

修改前效果如图所示&#xff0c;数据值状态应显示为可用&#xff0c;但实际上仅显示了状态码1&#xff0c;并没有显示其对应的状态信息。在排查了数据类型对应关系问题后&#xff0c;并没有产生实质性影响&#xff0c;只好对代码进行了如下修改。 修改前代码&#xff1a; <…
阅读更多...
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站