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

以下结合案例:统计消息中单词出现次数,来测试并说明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.mzph.cn/web/44089.shtml

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

相关文章

Okhttp实现原理

OkHttp 是一个高效的 HTTP 客户端库&#xff0c;广泛应用于 Android 和 Java 应用中。它提供了简洁的 API&#xff0c;支持多种协议&#xff0c;如 HTTP/1.x 和 HTTP/2&#xff0c;并且内置了缓存和重试机制。下面是结合源码分析的 OkHttp 的实现原理&#xff1a; 核心组件 O…

Swift 数据类型

Swift 数据类型 Swift 是一种强类型语言,这意味着在 Swift 中声明的每个变量和常量都必须具有明确的类型。Swift 的类型系统旨在帮助开发者编写清晰、安全的代码。本文将详细介绍 Swift 中的基本数据类型,包括整数、浮点数、布尔值、字符和字符串。 整数类型 Swift 提供了…

音频语言学习领域数据集现状、分类及评估

Audio Language Learning (Audio-Text Learning) 是一个新兴的研究领域&#xff0c;专注于处理、理解和描述声音。它的发展动力是机器学习技术的进步以及越来越多地将声音与其相应的文本描述相结合的数据集的可用性。 Audio Language Models (ALMs) 是这个领域的关键技术&#…

MATLAB中的SDPT3、LMILab、SeDuMi工具箱

MATLAB中的SDPT3、LMILab、SeDuMi工具箱都是用于解决特定数学优化问题的工具箱&#xff0c;它们在控制系统设计、机器学习、信号处理等领域有广泛的应用。以下是对这三个工具箱的详细介绍&#xff1a; 1. SDPT3工具箱 简介&#xff1a; SDPT3&#xff08;Semidefinite Progra…

基于QT开发的反射内存小工具

前言 最近项目需要需要开发一个反射内存小工具&#xff0c;经过2天的修修改终于完成了。界面如下&#xff1a; 功能简介 反射内存指定地址数据读取反射内存指定地址数据写入反射内存指定地址数据清理十进制、十六进制、二进制数据相互转换 部分代码 void RfmMain::setWOthe…

SqlSugar-使用SqlSugar进行多数据库操作

使用SqlSugar进行多数据库操作主要涉及以下几个步骤&#xff1a; 1. 配置数据库连接 首先&#xff0c;你需要在项目的配置文件中&#xff08;如appsettings.json、web.config或app.config&#xff09;配置多个数据库的连接字符串。每个连接字符串都对应一个不同的数据库。 例…

攻防世界(PHP过滤器过滤)file_include

转换过滤器官方文档&#xff1a;https://www.php.net/manual/zh/filters.convert.php#filters.convert.iconv 这道题因为convert.base64-encode被过滤掉了&#xff0c;所以使用convert.iconv.*过滤器 在激活 iconv 的前提下可以使用 convert.iconv.* 压缩过滤器&#xff0c; 等…

Win10安装MongoDB(详细版)

文章目录 1、安装MongoDB Server1.1. 下载1.2. 安装 2、手动安装MongoDB Compass(GUI可视工具)2.1. 下载2.2.安装 3、测试连接3.1.MongoDB Compass 连接3.2.使用Navicat连接 1、安装MongoDB Server 1.1. 下载 官网下载地址 https://www.mongodb.com/try/download/community …

【第28章】MyBatis-Plus之插件主体

文章目录 前言一、MybatisPlusInterceptor 概览1. 属性2. InnerInterceptor 接口 二、使用示例1.Spring 配置2.Spring Boot 配置3 .mybatis-config.xml 配置 三、拦截忽略注解 InterceptorIgnore四、手动设置拦截器忽略执行策略五、本地缓存 SQL 解析总结 前言 MyBatis-Plus 提…

android 固定图片大小

在Android中&#xff0c;固定图片大小可以通过多种方法实现&#xff0c;这些方法主要涉及到ImageView控件的使用、Bitmap类的操作&#xff0c;以及第三方库&#xff08;如Glide&#xff09;的辅助。以下是几种常见的方法&#xff1a; 1. 使用ImageView控件 在Android的布局文…

利用docker容器安装node,使用vue的开发环境

目录 vue-app ├── docker-data │ ├── site │ ├── app ├── docker-compose.yaml └── deploy.sh docker-compose.yaml yaml文件执行 version: 3.8services:node:image: node:latestcontainer_name: vue-appports:- "8080:8080" # 宿主8080映射容器8…

系统服务综合项目

要求&#xff1a; 现有主机 node01 和 node02&#xff0c;完成如下需求&#xff1a; 1、在 node01 主机上提供 DNS 和 WEB 服务 2、dns 服务提供本实验所有主机名解析 3、web服务提供 www.rhce.com 虚拟主机 4、该虚拟主机的documentroot目录在 /nfs/rhce 目录 5、该目录由 no…

如何保证语音芯片的稳定性能和延长使用寿命

要让语音芯片保持稳定性能&#xff0c;首先需要深入理解其工作原理和内部构造。语音芯片&#xff0c;作为现代电子设备中的核心组件之一&#xff0c;承载着声音信号的处理与输出功能。为了确保其稳定运行&#xff0c;我们需要从多个方面进行细致的考虑和操作。‌ 1、避免长期高…

Windows系统MySQL的安装,客户端工具Navicat的安装

下载mysql安装包&#xff0c;可以去官网下载&#xff1a;www.mysql.com。点击downloads 什么&#xff1f;后面还有福利&#xff1f; 下载MySQL 下载企业版&#xff1a; 下载Windows版 5点多的版本有点低&#xff0c;下载8.0.38版本的。Window系统。下载下面的企业版。不下载…

乡镇集装箱生活污水处理设备处理效率高

乡镇集装箱生活污水处理设备处理效率高 乡镇集装箱生活污水处理设备优势 结构紧凑&#xff1a;集装箱式设计减少了占地面积&#xff0c;便于在土地资源紧张的乡镇地区部署。 安装方便&#xff1a;设备出厂前已完成组装和调试&#xff0c;现场只需进行简单的连接和调试即可投入使…

[数字图像处理]基础知识整理(部分,持续更新)

程序中描述一副图像&#xff0c;已知其横向纵向的像素个数即可&#xff08;&#xff09; 灰度直方图能反映一副图像各个灰度级像素占图像的面积比&#xff08;√&#xff09; 从程序编写的角度看&#xff0c;描述一副图像的基本属性通常包括其分辨率&#xff0c;即图像的宽度…

Docker镜像和容器的管理

1 Docker镜像管理操作 开启镜像加速 根据关键字查询镜像 下载查看镜像 详细镜像信息 查看latest版本 上传镜像到阿里云仓库 2 Docker容器操作 关于容器根据第一个pid进程是否能正常在前台运行

19. 地址转换

地址转换 题目描述 Excel 是最常用的办公软件。每个单元格都有唯一的地址表示。比如&#xff1a;第 12 行第 4 列表示为&#xff1a;"D12"&#xff0c;第 5 行第 255 列表示为"IU5"。 事实上&#xff0c;Excel 提供了两种地址表示方法&#xff0c;还有一…

算法训练营第30天|122.买卖股票的最佳时机II|55. 跳跃游戏|45.跳跃游戏II|1005.K次取反后最大化的数组和

122.买卖股票的最佳时机II 思路&#xff1a;只有前一天与后一天的利润为正时&#xff0c;才将其加入总利润。 55. 跳跃游戏 思路&#xff1a;找最大覆盖范围 出错点&#xff1a;数组的遍历&#xff0c;遍历范围应该是覆盖范围内 45.跳跃游戏II 思路&#xff1a; 局部最优&am…

批量爬取B站网络视频信息

使用XPath爬取B站视频链接等相关信息 分析B站html框架获取内容完整代码 对于B站&#xff0c;目前网上的爬虫大多都是使用通过解析服务器的响应来爬取想要的内容&#xff0c;下面我们通过使用XPath来爬取B站上一些想要的信息 此次任务我们需要对B站搜索到的关键字&#xff0c;并…