浅析Kafka Streams消息流式处理流程及原理

以下结合案例：统计消息中单词出现次数，来测试并说明kafka消息流式处理的执行流程

Maven依赖

    <dependencies><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-streams</artifactId><exclusions><exclusion><artifactId>connect-json</artifactId><groupId>org.apache.kafka</groupId></exclusion><exclusion><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency></dependencies>

准备工作

首先编写创建三个类，分别作为消息生产者、消息消费者、流式处理者
KafkaStreamProducer：消息生产者

public class KafkaStreamProducer {public static void main(String[] args) throws ExecutionException, InterruptedException {Properties properties = new Properties();//kafka的连接地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.246.128:9092");//发送失败，失败的重试次数properties.put(ProducerConfig.RETRIES_CONFIG, 5);//消息key的序列化器properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");//消息value的序列化器properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");KafkaProducer<String, String> producer = new KafkaProducer<>(properties);for (int i = 0; i < 5; i++) {ProducerRecord<String, String> producerRecord = new ProducerRecord<>("kafka-stream-topic-input", "hello kafka");producer.send(producerRecord);}producer.close();}
}

该消息生产者向主题kafka-stream-topic-input发送五次hello kafka
KafkaStreamConsumer：消息消费者

public class KafkaStreamConsumer {public static void main(String[] args) {Properties properties = new Properties();//kafka的连接地址properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.246.128:9092");//消费者组properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");//消息的反序列化器properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");//手动提交偏移量properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);//订阅主题consumer.subscribe(Collections.singletonList("kafka-stream-topic-output"));try {while (true) {ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {System.out.println("consumerRecord.key() = " + consumerRecord.key());System.out.println("consumerRecord.value() = " + consumerRecord.value());}// 异步提交偏移量consumer.commitAsync();}} catch (Exception e) {e.printStackTrace();} finally {// 同步提交偏移量consumer.commitSync();}}
}

KafkaStreamQuickStart：流式处理类

public class KafkaStreamQuickStart {public static void main(String[] args) {Properties properties = new Properties();properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.246.128:9092");properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-quickstart");StreamsBuilder streamsBuilder = new StreamsBuilder();//流式计算streamProcessor(streamsBuilder);KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), properties);kafkaStreams.start();}/*** 消息格式：hello world hello world* 配置并处理流数据。* 使用StreamsBuilder创建并配置KStream，对输入的主题中的数据进行处理，然后将处理结果发送到输出主题。* 具体处理包括：分割每个消息的值，按值分组，对每个分组在10秒的时间窗口内进行计数，然后将结果转换为KeyValue对并发送到输出主题。** @param streamsBuilder 用于构建KStream对象的StreamsBuilder。*/private static void streamProcessor(StreamsBuilder streamsBuilder) {// 从"kafka-stream-topic-input"主题中读取数据流KStream<String, String> stream = streamsBuilder.stream("kafka-stream-topic-input");System.out.println("stream = " + stream);// 将每个值按空格分割成数组，并将数组转换为列表，以扩展单个消息的值stream.flatMapValues((ValueMapper<String, Iterable<String>>) value -> {String[] valAry = value.split(" ");return Arrays.asList(valAry);})// 按消息的值进行分组，为后续的窗口化计数操作做准备.groupBy((key, value) -> value)// 定义10秒的时间窗口，在每个窗口内对每个分组进行计数.windowedBy(TimeWindows.of(Duration.ofSeconds(10))).count()// 将计数结果转换为流，以便进行进一步的处理和转换.toStream()// 显示键值对的内容，并将键和值转换为字符串格式.map((key, value) -> {System.out.println("key = " + key);System.out.println("value = " + value);return new KeyValue<>(key.key().toString(), value.toString());})// 将处理后的流数据发送到"kafka-stream-topic-output"主题.to("kafka-stream-topic-output");}}

该处理类首先从主题kafka-stream-topic-input中获取消息数据，经处理后发送到主题kafka-stream-topic-output中，再由消息消费者KafkaStreamConsumer进行消费

执行结果

在这里插入图片描述
在这里插入图片描述

流式处理流程及原理说明

初始阶段

当从输入主题kafka-stream-topic-input读取数据流时，每个消息都是一个键值对。假设输入消息的键是null或一个特定的字符串，这取决于消息是如何被发送到输入主题的。

KStream<String, String> stream = streamsBuilder.stream("kafka-stream-topic-input");

分割消息值

使用flatMapValues方法分割消息的值，但这个操作不会改变消息的键。如果输入消息的键是null，那么在这个阶段消息的键仍然是null。

stream.flatMapValues((ValueMapper<String, Iterable<String>>) value -> {String[] valAry = value.split(" ");return Arrays.asList(valAry);
})

按消息的值进行分组

在 Kafka Streams 中，当使用groupBy方法对流进行分组时，实际上是在指定一个新的键，这个键将用于后续的窗口化操作和聚合操作。在这个案例中groupBy方法被用来按消息的值进行分组：

.groupBy((key, value) -> value)

这意味着在分组操作之后，流中的每个消息的键被设置为消息的值。因此，当你在后续的map方法中看到key参数时，这个key实际上是消息的原始值，因为在groupBy之后，消息的值已经变成了键。

定义时间窗口并计数

在这个阶段，消息被窗口化并计数，但是键保持不变。

.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
.count()

将计数结果转换为流

当将计数结果转换为流时，键仍然是之前分组时的键

.toStream()

处理和转换结果

在map方法中，你看到的key参数实际上是分组后的键，也就是消息的原始值：

.map((key, value) -> {System.out.println("key = " + key);System.out.println("value = " + value);return new KeyValue<>(key.key().toString(), value.toString());
})

map方法中的key.key().toString()是为了获取键的字符串表示，而value.toString()是为了将计数值转换为字符串。

将处理后的数据发送到输出主题

.to("kafka-stream-topic-output");

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

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

相关文章

kafka发送消息流程

kafka发送消息流程

配置props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, RoundRobinPartitioner.class); public Map<String,Object> producerConfigs(){Map<String,Object> props new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServers…
阅读更多...
文件的顺序读写

文件的顺序读写

文件读写函数介绍 文件顺序读写函数 函数名功能适用于fputc 字符输出函数 所有输出流 fgetc 字符输⼊函数 所有输⼊流 fputs ⽂本⾏输出函数 所有输出流fgets ⽂本⾏输⼊函数 所有输⼊流fprintf 格式化输出函数 所有输出流fscanf 格式化输⼊函数 所有输⼊流fwrite ⼆进制输出…
阅读更多...
Spring源码中的模板方法模式

Spring源码中的模板方法模式

1. 什么是模板方法模式 模板方法模式&#xff08;Template Method Pattern&#xff09;是一种行为设计模式&#xff0c;它在操作中定义算法的框架&#xff0c;将一些步骤推迟到子类中。模板方法让子类在不改变算法结构的情况下重新定义算法的某些步骤。 模板方法模式的定义&…
阅读更多...
被动的机器人非线性MPC控制

被动的机器人非线性MPC控制

MPC是一种基于数学模型的控制策略&#xff0c;它通过预测系统在未来一段时间内的行为&#xff0c;并求解优化问题来确定当前的控制输入&#xff0c;以实现期望的控制目标。对于非线性系统&#xff0c;MPC可以采用非线性模型进行预测和优化&#xff0c;这种方法被称为非线性模型…
阅读更多...
DBeaver导入脚本和导出数据

DBeaver导入脚本和导出数据

DBeaver导入脚本和导出数据 前言&#xff1a; 通常产品会要求&#xff0c;把xx表导出Excel&#xff0c;navicat一般公司不让用。讲解使用DBeaver 导入脚本 我们将sql脚本导入DBeaver 1&#xff0c;选择数据库&#xff0c;找到执行脚本 2&#xff0c;选用sql脚本&#xff0…
阅读更多...
【Linux】权限管理与相关指令

【Linux】权限管理与相关指令

文章目录 1.权限、文件权限、用户文件权限的理解以及对应八进制数值表示、设置目录为粘滞位文件类型 2.权限相关的常用指令susudochmodchownchgrpumaskwhoamifile 1.权限、文件权限、用户 通过一定条件&#xff0c;拦住一部分人&#xff0c;给另一部分权利来访问资源&#xff0…
阅读更多...
IDEA实现SpringBoot项目的自打包自发布自部署

IDEA实现SpringBoot项目的自打包自发布自部署

目录 前言 正文 操作背景 自发布 自部署 尾声 &#x1f52d; Hi,I’m Pleasure1234&#x1f331; I’m currently learning Vue.js,SpringBoot,Computer Security and so on.&#x1f46f; I’m studying in University of Nottingham Ningbo China&#x1f4eb; You can reach…
阅读更多...
U盘坏了怎么把数据弄出来

U盘坏了怎么把数据弄出来

在数字化时代&#xff0c;U盘作为一种便携式存储设备&#xff0c;为我们的数据保存和传输提供了很多便利。不过&#xff0c;我们在使用U盘过程中难免会遇到各种问题&#xff0c;比如常见的U盘损坏问题。U盘损坏会导致数据无法读取&#xff0c;给我们造成困扰。幸运的是&#xf…
阅读更多...
Spring相关的面试题

Spring相关的面试题

1、spring中bean的生命周期 spring bean的生命周期主要分为三大类 &#xff0c;分别是创建-》使用-〉销毁。 在三大类下面又可以分为5个小类。分别是 实列化-〉初始化-》组册destruction回调-〉使用-〉销毁 这这其中 初始化也可以细分为 设置属性值&#xff0c;前置处理&#…
阅读更多...
【学习笔记】无人机（UAV）在3GPP系统中的增强支持(七)-通过无人机实现无线接入的独立部署

【学习笔记】无人机（UAV）在3GPP系统中的增强支持(七)-通过无人机实现无线接入的独立部署

引言 本文是3GPP TR 22.829 V17.1.0技术报告&#xff0c;专注于无人机&#xff08;UAV&#xff09;在3GPP系统中的增强支持。文章提出了多个无人机应用场景&#xff0c;分析了相应的能力要求&#xff0c;并建议了新的服务级别要求和关键性能指标&#xff08;KPIs&#xff09;。…
阅读更多...
dhtmlx-gantt甘特图数据展示

dhtmlx-gantt甘特图数据展示

官网文档&#xff1a;甘特图文档 实现效果&#xff1a; 首先需要下载 dhtmlx-gantt组件 npm i dhtmlx-gantt //我项目中使用的是"dhtmlx-gantt": "^8.0.6" 这个版本&#xff0c;不同的版本api或是文档中存在的方法稍有差异 界面引用 <template>&l…
阅读更多...
什么是MOW，以bitget钱包为例

什么是MOW，以bitget钱包为例

元描述&#xff1a;MOW凭借其富有创意的故事情节和广阔的潜力在Solana上脱颖而出。本文深入探讨了其独特的概念和光明的未来。 Mouse in a Cats World (MOW)是一个基于Solana区块链的创新meme项目&#xff0c;它重新构想了一个异想天开且赋予权力的故事。在这个奇幻的宇宙中&am…
阅读更多...
Leetcode算法题（链表的中间节点+返回倒数第k个节点+合并两个有序链表）

Leetcode算法题（链表的中间节点+返回倒数第k个节点+合并两个有序链表）

题目1&#xff1a; 本题力扣链接&#xff1a;https://leetcode.cn/problems/middle-of-the-linked-list/solutions/164351/lian-biao-de-zhong-jian-jie-dian-by-leetcode-solut/ 思路1&#xff1a;单指针法 首先我们对链表进行遍历&#xff0c;记录链表的总长度N&#xff0c;…
阅读更多...
微信小程序毕业设计-学习资料库系统项目开发实战(附源码+论文)

微信小程序毕业设计-学习资料库系统项目开发实战(附源码+论文)

大家好&#xff01;我是程序猿老A&#xff0c;感谢您阅读本文&#xff0c;欢迎一键三连哦。 &#x1f49e;当前专栏&#xff1a;微信小程序毕业设计 精彩专栏推荐&#x1f447;&#x1f3fb;&#x1f447;&#x1f3fb;&#x1f447;&#x1f3fb; &#x1f380; Python毕业设计…
阅读更多...
使用APEXSQL LOG解析sql server事务日志，进行审计与数据恢复

使用APEXSQL LOG解析sql server事务日志，进行审计与数据恢复

一 下载 https://download.csdn.net/download/sunke861/11449739 二 使用 解压安装包后&#xff0c;点击&#xff1a;ApexSQLLog.exe 2.1 连接数据库 连接要审计的数据库&#xff1a; 假如报错&#xff1a; 则点击ok关闭该窗口&#xff0c;然后点击左上方的New按钮&#xf…
阅读更多...
平替ChatGPT的多模态智能体来了

平替ChatGPT的多模态智能体来了

在人工智能领域&#xff0c;多模态技术的融合与应用已成为推动技术革新的关键。今天&#xff0c;我们用智匠AI实现了完全由国产模型驱动的多模态智能体——智酱v0.1.0&#xff0c;它不仅能够媲美ChatGPT的多模态能力&#xff0c;更在联网搜索、图片识别、画图及图表生成等方面展…
阅读更多...
Python进阶 2024/7/15

Python进阶 2024/7/15

目录 文件的写入操作 打开文件 文件写入 内容刷新 文件的追加写入操作 打开文件 文件写入 内容刷新 a模式 练习 文件的写入操作 文件的访问模式要用 w 打开文件 f open( " D:\ACM\python.txt "," w ",encoding " UTF - 8") 文件写…
阅读更多...
BL201分布式I/O耦合器连接Profinet网络

BL201分布式I/O耦合器连接Profinet网络

钡铼技术的BL201分布式I/O耦合器是一个用于Profinet网络的设备&#xff0c;用于连接远程输入/输出&#xff08;I/O&#xff09;设备到控制系统&#xff0c;如可编程逻辑控制器&#xff08;PLC&#xff09;&#xff0c;能够实现分布式的I/O连接和通信。 它支持标准Profinet IO …
阅读更多...
Xilinx FPGA UltraScale SelectIO 接口逻辑资源

Xilinx FPGA UltraScale SelectIO 接口逻辑资源

目录 1. 简介 2. Bank Overview 2.1 Diagram 2.2 IOB 2.3 Slice 2.4 Byte Group 2.5 I/O bank 示例 2.6 Pin Definition 2.7 数字控制阻抗(DCI) 2.8 SelectIO 管脚供电电压 2.8.1 VCCO 2.8.2 VREF 2.8.3 VCCAUX 2.8.4 VCCAUX_IO 2.8.5 VCCINT_IO 3. 总结 1. 简介…
阅读更多...
mac电脑pdf合并，macpdf合并成一个pdf

mac电脑pdf合并，macpdf合并成一个pdf

在数字化办公和学习的今天&#xff0c;pdf文件因其跨平台兼容性强、格式稳定而成为了最受欢迎的文档格式之一。但随之而来的问题也接踵而至&#xff0c;如何将多个pdf文件合并为一个&#xff1f;这不仅关系到文档的整洁性&#xff0c;更是时间管理的重要环节。今天&#xff0c;…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023