北京网站运营优化公司,dede和wordpress,广西壮族自治区人口,网络系统管理技能大赛简介#xff1a; kafkaStream#xff1a;提供了对存储在kafka中的数据进行流式处理和分析的功能
特点#xff1a; KafkasSream提供了一个非常简单轻量的Library#xff0c;它可以非常方便的嵌入到java程序中#xff0c;也可以任何方式打包部署
入门案例#xff1a; 1、…简介 kafkaStream提供了对存储在kafka中的数据进行流式处理和分析的功能
特点 KafkasSream提供了一个非常简单轻量的Library它可以非常方便的嵌入到java程序中也可以任何方式打包部署
入门案例 1、新建工程kafka-demo 引入kafkaStream依赖 dependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependency!-- kafkfa --dependencygroupIdorg.springframework.kafka/groupIdartifactIdspring-kafka/artifactIdexclusionsexclusiongroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactId/exclusion/exclusions/dependencydependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactId/dependencydependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactId/dependency!--kafkaStream--dependencygroupIdorg.apache.kafka/groupIdartifactIdkafka-streams/artifactIdexclusionsexclusionartifactIdconnect-json/artifactIdgroupIdorg.apache.kafka/groupId/exclusionexclusiongroupIdorg.apache.kafka/groupIdartifactIdkafka-clients/artifactId/exclusion/exclusions/dependency/dependencies 2、新建流式处理类 代码如下
package com.heima.kafkademo.sample;import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;/*
* 流式处理
* */
public class KafkaStreamQuickStart {public static void main(String[] args) {/*创建kafka配置中心并配置参数*/Properties prop new Properties();//连接地址prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,192.168.200.130:9092);//key序列化prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());//value序列化prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());//创建id名称prop.put(StreamsConfig.APPLICATION_ID_CONFIG,streams-quickstart);//stream构造器StreamsBuilder streamsBuilder new StreamsBuilder();//流式计算streamProcessor(streamsBuilder);//创建KafkaStream对象KafkaStreams kafkaStreams new KafkaStreams(streamsBuilder.build(),prop);//开启流式计算kafkaStreams.start();}//流式计算方法private static void streamProcessor(StreamsBuilder streamsBuilder) {//创建kafka对象同时指定从哪个topic获取消息KStreamString, String stream streamsBuilder.stream(itcast-topic-input);//处理消息的valuestream.flatMapValues(new ValueMapperString, Iterable?() {Overridepublic IterableString apply(String value) {return Arrays.asList(value.split( ));}}) //按照value进行聚合.groupBy((key,value)-value)//时间窗口,每隔10秒更新一次.windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//统计单词个数.count()//转换为kStream.toStream().map((key,value)-{System.out.println(key:key,vlaue:value);return new KeyValue(key.key().toString(),value.toString());})//发送消息.to(itcast-topic-out);}
}3、启动消费者类和流式处理类监听消息 使用生产者类发送消息 消费者和生产者类代码参考Kafka安装和配置_Success___的博客-CSDN博客 4、测试 成功接收到消息
