springboot~kafka-stream实现实时统计

实时统计，也可以理解为流式计算，一个输入流，一个输出流，源源不断。
Kafka Stream

Kafka Stream是Apache Kafka从0.10版本引入的一个新Feature。它是提供了对存储于Kafka内的数据进行流式处理和分析的功能。
Kafka Stream的特点

• Kafka Stream提供了一个非常简单而轻量的Library，它可以非常方便地嵌入任意Java应用中，也可以任意方式打包和部署
  
• 除了Kafka外，无任何外部依赖
  
• 充分利用Kafka分区机制实现水平扩展和顺序性保证
  
• 通过可容错的state store实现高效的状态操作（如windowed join和aggregation）
  
• 支持正好一次处理语义
  
• 提供记录级的处理能力，从而实现毫秒级的低延迟
  
• 支持基于事件时间的窗口操作，并且可处理晚到的数据（late arrival of records）
  
• 同时提供底层的处理原语Processor（类似于Storm的spout和bolt），以及高层抽象的DSL（类似于Spark的map/group/reduce）
  

相关术语

源处理器和Sink处理器是Kafka Streams中的两个重要组件，它们分别用于从输入流获取数据并将处理后的数据发送到输出流。以下是它们的工作流程的文字图示表达：

1. [Source Processor] -> [Processor Topology] -> [Sink Processor]

复制代码

• 源处理器（Source Processor）：
  
  
  
  • 源处理器负责从一个或多个输入主题（topics）中提取数据，并将数据转换为KStream或KTable对象。
    
  • 它通常是处理拓扑结构的起点，从一个或多个输入主题中读取数据，并将其发送到处理拓扑中的下一个处理器。
    
  
  
• Sink 处理器（Sink Processor）：
  
  
  
  • Sink处理器负责将经过处理的数据发送到一个或多个输出主题，或者执行其他终端操作。
    
  • 它通常是处理拓扑结构的终点，在处理拓扑的最后阶段接收处理后的数据，并将其发送到输出主题，或者执行其他终端操作，如存储到数据库、发送到外部系统等。
    
  
  
• Processor Topology：
  
  
  
  • 处理拓扑包含了源处理器、中间处理器和Sink处理器，它定义了数据流的处理逻辑。
    
  • 在处理拓扑中，数据流会通过一系列的处理器进行转换、聚合和处理，最终到达Sink处理器，完成整个处理流程。
    
  
  

通过这种处理流程，Kafka Streams可以实现对数据流的灵活处理和转换，使得你能够方便地构建实时流处理应用程序。
kafka stream demo

依赖

1. <dependency>
2.     <groupId>org.springframework.kafka</groupId>
3.     <artifactId>spring-kafka</artifactId>
4.     <version>2.5.5.RELEASE</version>
5. </dependency>
6. <dependency>
7.     <groupId>org.apache.kafka</groupId>
8.     <artifactId>kafka-clients</artifactId>
9.     <version>2.5.1</version>
10. </dependency>
11. <dependency>
12.     <groupId>org.apache.kafka</groupId>
13.     <artifactId>kafka-streams</artifactId>
14.   <version>2.5.1</version>
15. </dependency>

复制代码

环境准备

• 安装kafka
  
• 建立topic，我以keycloak为例，它有login_in这个主题，用来记录登录信息
  
• 建立topic，如total_record，用来存储login_in的实时统计的结果
  
• 可使用springboot继承的消费者，去消费total_record，如写入数据库进行持久化
  

业务代码

• 配置类
  

1. @Configuration
2. @EnableKafkaStreams
3. public class KafkaStreamConfig {
5.         private static final int MAX_MESSAGE_SIZE = 16 * 1024 * 1024;
7.         @Value("${spring.kafka.bootstrap-servers}")
8.         private String hosts;
10.         @Value("${spring.kafka.consumer.group-id}")
11.         private String group;
13.         @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
14.         public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
15.                 Map<String, Object> props = new HashMap<>();
16.                 props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
17.                 props.put(StreamsConfig.APPLICATION_ID_CONFIG, group + "_stream_aid");
18.                 props.put(StreamsConfig.CLIENT_ID_CONFIG, group + "_stream_cid");
19.                 props.put(StreamsConfig.RETRIES_CONFIG, 3);
20.                 props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");//从最近的消息开始消费
21.                 props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
22.                 props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
23.                 return new KafkaStreamsConfiguration(props);
24.         }
26. }

复制代码

• 消费类
  

1. @Configuration
2. @Slf4j
3. public class KafkaStreamListener {
5.         @Autowired
6.         ReportLoginTypeMapper reportLoginTypeMapper;
7.         @KafkaListener(topics = "total_record")
8.         public void listen(ConsumerRecord<String, String> record) {
9.                 // 将时间戳转换为本地日期时间
10.                 LocalDateTime dateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(record.timestamp()), ZoneId.systemDefault());
11.                 ReportLoginType reportLoginType=new ReportLoginType();
12.                 reportLoginType.setLoginType(record.key());
13.                 reportLoginType.setCreateAt(dateTime);
14.                 reportLoginType.setCount(Integer.parseInt(record.value()));
15.                 reportLoginTypeMapper.insert(reportLoginType);
16.         }
18.         @Bean
19.         public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
20.                 KStream<String, String> stream = streamsBuilder.stream("KC_LOGIN");
21.                 KStream<String, String> serializedStream = stream.mapValues(jsonString -> {
22.                         // 分组依据
23.                         if (JSONUtil.parseObj(jsonString).containsKey("details")) {
24.                                 JSONObject details = JSONUtil.parseObj(jsonString).getJSONObject("details");
25.                                 if (details.containsKey("loginType")) {
26.                                         String loginType = details.getStr("loginType");
27.                                         return loginType;
28.                                 }
29.                                 return "";
30.                         }
31.                         else {
32.                                 return "";
33.                         }
34.                 });
35.                 /**
36.                  * 处理消息的value
37.                  */
38.                 serializedStream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
39.                         @Override
40.                         public Iterable<String> apply(String value) {
41.                                 return Arrays.asList(value.split(" "));
42.                         }
43.                 }).filter((key, value) -> !value.equals(""))
44.                                 // 按照value进行聚合处理
45.                                 .groupBy((key, value) -> value)// 这进而的value是kafka的消息内容
46.                                 // 时间窗口
47.                                 .windowedBy(TimeWindows.of(Duration.ofSeconds(60)))
48.                                 // 统计单词的个数
49.                                 .count()
50.                                 // 转换为kStream
51.                                 .toStream().map((key, value) -> {
52.                                         // key是分组的key,value是分组count的结果
53.                                         return new KeyValue<>(key.key().toString(), value.toString());
54.                                 })
55.                                 // 发送消息
56.                                 .to("topic-out");
57.                 return stream;
58.         }
60. }

复制代码

最后看一下total_record的内容



免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！