实例演示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.mfbz.cn/a/778383.html

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

相关文章

笔记13：switch多分支选择语句

笔记13：switch多分支选择语句

引例&#xff1a; 输入1-5中的任意一共数字&#xff0c;对应的打印字符A,B,C,D,E int num 0; printf("Input a number[1,5]:"); scanf("%d"&#xff0c;&num); if( num 1)printf("A\n"); else if(num2)printf("B\n"); else i…
阅读更多...
【大数据】—FIFA世界杯探索性分析（EDA）

【大数据】—FIFA世界杯探索性分析（EDA）

引言 足球&#xff0c;作为全球最受欢迎的运动之一&#xff0c;拥有庞大的粉丝群体和深远的文化影响。自1930年首届FIFA世界杯举办以来&#xff0c;这项赛事已经成为全球体育盛事&#xff0c;吸引了数十亿观众的目光。世界杯不仅是各国足球技艺的较量&#xff0c;更是国家荣誉…
阅读更多...
02STM32环境搭建新建工程

02STM32环境搭建新建工程

STM32环境搭建&新建工程 软件安装&#xff1a;开发方式&新建工程步骤&架构 个人心得 软件安装&#xff1a; 安装Keil5 MDK 安装器件支持包 软件注册 安装STLINK驱动 安装USB转串口驱动 开发方式&新建工程步骤&架构 STM32开发方式&#xff1a; 1.寄存器 …
阅读更多...
笔记14：程序中的循环结构

笔记14：程序中的循环结构

生活中的循环现象&#xff1a; -日复一日&#xff0c;年复一年 -春夏秋冬&#xff0c;四季交替 -周日&#xff0c;周一&#xff0c;周二&#xff0c;周三&#xff0c;周四&#xff0c;周五&#xff0c;周六 -人生是一个轮回&#xff0c;多年后&#xff0c;又会回到最初的原点 …
阅读更多...
APP渗透-android12夜神模拟器+Burpsuite实现

APP渗透-android12夜神模拟器+Burpsuite实现

一、夜神模拟器下载地址&#xff1a;https://www.yeshen.com/ 二、使用openssl转换证书格式 1、首先导出bp证书 2、将cacert.der证书在kali中转换 使用openssl生成pem格式证书,并授予最高权限 openssl x509 -inform der -in cacert.der -out cacert.pem chmod 777 cacert…
阅读更多...
容器：queue(队列)

容器：queue(队列)

以下是关于queue容器的总结 1、构造函数&#xff1a;queue [queueName] 2、添加、删除元素: push() 、pop() 3、获取队头/队尾元素&#xff1a;front()、back() 4、获取栈的大小&#xff1a;size() 5、判断栈是否为空&#xff1a;empty() #include <iostream> #include …
阅读更多...
解决 MEX 文件 ‘xxx.mexw64‘ 无效: 找不到指定的模块。的问题

解决 MEX 文件 ‘xxx.mexw64‘ 无效: 找不到指定的模块。的问题

1.问题描述 在matlab R2021b中运行编译好后的gptoolbox工具箱中的函数[SVtemp,SFtemp,IF] selfintersect(V,F);报错如下 MEX 文件 E:\MATLAB_File\gptoolbox\mex\selfintersect.mexw64 无效: 找不到指定的模块。出错 offset_bunny (第 15 行) [SVtemp,SFtemp,IF] selfinter…
阅读更多...
MATLAB-分类CPO-RF-Adaboost冠豪猪优化器（CPO）优化RF随机森林结合Adaboost分类预测（二分类及多分类）

MATLAB-分类CPO-RF-Adaboost冠豪猪优化器（CPO）优化RF随机森林结合Adaboost分类预测（二分类及多分类）

MATLAB-分类CPO-RF-Adaboost冠豪猪优化器&#xff08;CPO&#xff09;优化RF随机森林结合Adaboost分类预测&#xff08;二分类及多分类&#xff09; 分类CPO-RF-Adaboost冠豪猪优化器&#xff08;CPO&#xff09;优化RF随机森林结合Adaboost分类预测&#xff08;二分类及多分类…
阅读更多...
MySQL基础篇（三）数据库的修改 删除 备份恢复 查看连接情况

MySQL基础篇（三）数据库的修改 删除 备份恢复 查看连接情况

对数据库的修改主要指的是修改数据库的字符集&#xff0c;校验规则。 将test1数据库字符集改为gbk。 数据库的删除&#xff1a; 执行完该数据库就不存在了&#xff0c;对应数据库文件夹被删除&#xff0c;级联删除&#xff0c;里面的数据表全部被删除。 注意&#xff1a;不要随…
阅读更多...
线程同步66666

线程同步66666

1. 概述 当有多个线程访问同一个共享资源&#xff08;临界资源&#xff09;时&#xff0c;且不允许同时访问&#xff0c;那么就需要线程同步。常见的线程同步方式&#xff1a;互斥锁、读写锁、条件变量、信号量。 2. 互斥锁 互斥锁的方式可以简单概括为&#xff1a;锁定操作…
阅读更多...
ros1仿真导航机器人 navigation

ros1仿真导航机器人 navigation

仅为学习记录和一些自己的思考&#xff0c;不具有参考意义。 1navigation导航框架 2导航设置过程 &#xff08;1&#xff09;启动仿真环境 roslaunch why_simulation why_robocup.launch &#xff08;2&#xff09;启动move_base导航、amcl定位 roslaunch why_simulation nav…
阅读更多...
【OnlyOffice】桌面应用编辑器，插件开发大赛，等你来挑战

【OnlyOffice】桌面应用编辑器，插件开发大赛，等你来挑战

OnlyOffice&#xff0c;桌面应用编辑器&#xff0c;最近版本已从8.0升级到了8.1 从PDF、Word、Excel、PPT等全面进行了升级。随着AI应用持续的火热&#xff0c;OnlyOffice也在不断推出AI相关插件。 因此&#xff0c;在此给大家推荐一下OnlyOffice本次的插件开发大赛。 详细信息…
阅读更多...
磁盘就是一个超大的Byte数组，操作系统是如何管理的？

磁盘就是一个超大的Byte数组，操作系统是如何管理的？

磁盘在操作系统的维度看&#xff0c;就是一个“超大的Byte数组”。 那么操作系统是如何对这块“超大的Byte数组”做管理的呢&#xff1f; 我们知道在逻辑上&#xff0c;上帝说是用“文件”的概念来进行管理的。于是&#xff0c;便有了“文件系统”。那么&#xff0c;文件系统…
阅读更多...
深圳晶彩智能ESP32-2432S028R实时观察LVGL9效果

深圳晶彩智能ESP32-2432S028R实时观察LVGL9效果

深圳晶彩智能ESP32-2432S028R概述&#xff1a; 深圳晶彩智能出品ESP32-32432S028R为2.8寸彩色屏采用分辨率320x240彩色液晶屏&#xff0c;驱动芯片是ILI9431。板载乐鑫公司出品ESP-WROOM-32&#xff0c;Flash 4M。型号尾部“R”标识电阻膜的感压式触摸屏&#xff0c;驱动芯片是…
阅读更多...
以腾讯为例，手把手教你搭建产品帮助中心

以腾讯为例，手把手教你搭建产品帮助中心

一个精心设计的产品帮助中心对于提高用户满意度和体验至关重要。腾讯&#xff0c;作为全球领先的互联网企业&#xff0c;通过其多样化的产品线&#xff08;包括微信、QQ、腾讯游戏、腾讯视频等&#xff09;吸引了亿万用户。下面将以腾讯为例&#xff0c;向您展示如何搭建一个高…
阅读更多...
数据库系统原理 | 查询作业1

数据库系统原理 | 查询作业1

整理自博主本科《数据库系统原理》专业课自己完成的实验课查询作业&#xff0c;以便各位学习数据库系统概论的小伙伴们参考、学习。 *文中若存在书写不合理的地方&#xff0c;欢迎各位斧正。 专业课本&#xff1a; ​ ———— 本次实验使用到的图形化工具&#xff1a;Heidisql…
阅读更多...
进程的控制-ps和kill命令

进程的控制-ps和kill命令

ps 查看进程信息 部分参数&#xff1a; a : 显示现行终端机下的所有程序&#xff0c;包括其他用户的程序 u: 以用户为主的格式来显示程序状况 x: 显示所有程序&#xff0c;不以 终端机来区分 kill 向指定的进程发送信号 kill 可将指定的信息送至程序。预设的信息为 SIG…
阅读更多...
linux下高级IO模型

linux下高级IO模型

高级IO 1.高级IO模型基本概念1.1 阻塞IO1.2 非阻塞IO1.3 信号驱动IO1.4 IO多路转接1.5 异步IO 2. 模型代码实现2.1 非阻塞IO2.2 多路转接-selectselect函数介绍什么才叫就绪呢&#xff1f;demoselect特点 2.3 多路转接-pollpoll函数介绍poll优缺点demo 2.4 多路转接-epoll&…
阅读更多...
24西安电子科技大学马克思主义学院—考研录取情况

24西安电子科技大学马克思主义学院—考研录取情况

01、马克思主义学院各个方向 02、24马克思主义学院近三年复试分数线对比 PS&#xff1a;马院24年院线相对于23年院线增加15分&#xff0c;反映了大家对于马克思主义理论学习与研究的热情高涨&#xff0c;也彰显了学院在人才培养、学科建设及学术研究等方面的不断进步与成就。 6…
阅读更多...
集成学习（一）Bagging

集成学习（一）Bagging

前边学习了&#xff1a;十大集成学习模型&#xff08;简单版&#xff09;-CSDN博客 Bagging又称为“装袋法”&#xff0c;它是所有集成学习方法当中最为著名、最为简单、也最为有效的操作之一。 在Bagging集成当中&#xff0c;我们并行建立多个弱评估器&#xff08;通常是决策…
阅读更多...
最新文章

Copyright @ 2022~2023