簡介：

? ? ? ? kafkaStream：提供了對存儲在kafka中的數據進行流式處理和分析的功能

特點：

????????KafkasSream提供了一個非常簡單輕量的Library，它可以非常方便的嵌入到java程序中，也可以任何方式打包部署

入門案例：

? 1、新建工程kafka-demo

? ? ? ? ? ?引入kafkaStream依賴

    <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- kafkfa --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><exclusions><exclusion><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency><!--kafkaStream--><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></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獲取消息KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");//處理消息的valuestream.flatMapValues(new ValueMapper<String, Iterable<?>>() {@Overridepublic Iterable<String> 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、測試

? ? ? ? 成功接收到消息

?

?