【Kafka】Kafka Stream简单使用

爱被打了一巴掌 2024-03-24 22:13 117阅读 0赞

### 一、实时流式计算 ###

#### 1. 概念 ####

> 一般**流式计算**会与**批量计算**相比较。在流式计算模型中，输入是持续的，可以认为在时间上是无界的，也就意味着，永远拿不到全量数据去做计算。同时，计算结果是持续输出的，也即计算结果在时间上也是无界的。流式计算一般对实时性要求较高，同时一般是先定义目标计算，然后数据到来之后将计算逻辑应用于数据。同时为了提高计算效率，往往尽可能采用增量计算代替全量计算  
> ![在这里插入图片描述][4a041929a1ab4730a0095f4079fc2f5d.png]  
> **流式计算**就相当于上图的右侧扶梯，是可以源源不断的产生数据，源源不断的接收数据，没有边界。

#### 2. 应用场景 ####

>  *  日志分析: 网站的用户访问日志进行实时的分析，计算访问量，用户画像，留存率等等，实时的进行数据分析，帮助企业进行决策
>  *  大屏看板统计: 可以实时的查看网站注册数量，订单数量，购买数量，金额等。
>  *  公交实时数据: 可以随时更新公交车方位，计算多久到达站牌等
>  *  实时文章分值计算

比如应用较广的 **头条类文章的分值计算，通过用户的行为实时文章的分值，分值越高就越被推荐**。

#### 3. Kafka Stream ####

> 近些年来，开源流处理领域涌现出了很多优秀框架。光是在 Apache 基金会孵化的项目，关于流处理的大数据框架就有十几个之多，比如早期的`Apache Samza`、`Apache Storm`，以及这些年火爆的 `Spark` 以及 `Flink` 等。

##### 3.1 Kafka Streams的特点 #####

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

![在这里插入图片描述][6cce41129c2d4807abab124913d7017b.png]

##### 3.2 关键概念 #####

一个最简单的`Streaming`的结构如下图所示：  
![在这里插入图片描述][0234903f20284a0faf3f9283d3dc7b58.png]

从一个`Topic`中读取到数据，经过一些处理操作之后，写入到另一个`Topic`中，这就是一个最简单的`Streaming流式计算`。其中，`Source Topic`中的数据会源源不断的产生新数据。  
那么，我们再在上面的结构之上扩展一下，假设定义了多个`Source Topic`及`Destination Topic`，那就构成如下图所示的较为复杂的拓扑结构：  
![在这里插入图片描述][cf379e9599904bd2845711d01ad8ccff.png]

*  **源处理器（Source Processor）**：源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个`kafka主题`生成`输入流`。通过消费这些主题的消息并将它们转发到`下游处理器`。
 *  **Sink处理器**：sink处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的**Kafka主题**。  
    ![在这里插入图片描述][fea09e50e2324b5fa410e0fb229e5da2.png]  
    **Kafka Streams被认为是开发实时应用程序的最简单方法。它是一个Kafka的客户端API库，编写简单的java就可以实现流式处理。**

##### 3.3 KStream #####

`KStream`:数据结构类似于`map`，如下图，`key-value`键值对

![在这里插入图片描述][86446e71e551453b986e66146d642dd8.png]

`KStream`数据流（data stream），是一段顺序的，可以无限长，不断更新的数据集。  
数据流中比较常记录的是事件，这些事件可以是一次鼠标点击（click），一次交易，或是传感器记录的位置数据。

`KStream`负责抽象的，就是数据流。与Kafka自身`topic`中的数据一样，类似日志，每一次操作都是**向其中插入（insert）新数据。**

### 二、测试kafkaStream ###

先看下简单的`kafkaStream`的`KStream`测试

**需求分析：求单词个数（word count）**  
![在这里插入图片描述][647f17195a2b4fbab5b00856d8c1a709.png]

#### 1. `pom.xml`引入依赖： ####

<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>
    
            <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>

#### 2. 配置文件 ####

server:
      port: 9991
    spring:
      application:
        name: kafka-demo
      kafka:
        bootstrap-servers: 192.168.200.130:9092
        producer:
          retries: 10
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
          compression-type: lz4
        consumer:
          group-id: ${
        spring.application.name}-test
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

#### 3. 编写生产者 ####

`ProducerQuickStart.java`

package com.kafka.sample;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.clients.producer.*;
    
    import java.util.Properties;
    
    @Slf4j
    public class ProducerQuickStart {
        
    
        public static void main(String[] args) {
        
    
            //1. kafka的配置信息
            Properties prop = new Properties();
            //kafka的链接信息
            prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
            //配置重试次数
            prop.put(ProducerConfig.RETRIES_CONFIG, 5);
            //数据压缩
            prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
            //ack配置  消息确认机制   默认ack=1,即只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应
    //        prop.put(ProducerConfig.ACKS_CONFIG,"all");
    
            消息key的序列化器
            prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            //消息value的序列化器
            prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
    
            //2. 生产者对象
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(prop);
            //封装发送的消息
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("itcast-topic-input", "key_001", "hello kafka");
    
            //3. 发送消息
            for (int i = 0; i < 5; i++) {
        
                producer.send(producerRecord);
            }
    
            //4. 关闭消息通道  必须关闭，否则消息发不出去
            producer.close();
    
        }
    }

#### 4 编写kafkaStream流式处理 ####

`KafkaStreamQuickStart.java`

package com.kafka.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");
            prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            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();
        }
    
        /**
         * 流式计算
         * 消息的内容：hello kafka  hello itcast
         * @param streamsBuilder
         */
        private static void streamProcessor(StreamsBuilder streamsBuilder) {
        
            //创建kstream对象，同时指定从那个topic中接收消息
            KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
            /**
             * 处理消息的value
             */
            stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
        
                @Override
                public Iterable<String> apply(String value) {
        
                    return Arrays.asList(value.split(" "));
                }
            })
                    //按照value进行聚合处理
                    .groupBy((key,value)->value)
                    //时间窗口
                    .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");
    
        }
    }

#### 5. 编写消费者 ####

`ConsumerQuickStart.java`

package com.kafka.sample;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Properties;
    
    public class ConsumerQuickStart {
        
    
        public static void main(String[] args) {
        
    
            //1. 添加kafka的配置信息
            Properties properties = new Properties();
            // 配置链接信息
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
            //配置消费者组
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group-2");
            //配置消息的反序列化器
            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");
    
            //2. 消费者对象
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
    
            //3. 订阅主题
            consumer.subscribe(Collections.singletonList("itcast-topic-out"));
    
            //当前线程一直监听消息
            while(true){
        
                //4. 消费者拉取消息： 每秒拉取一次
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
        
                    System.out.println(record.key());
                    System.out.println(record.value());
                }
            }
    
        }
    }

> 启动项目：
> 
> 1.  在远端（`192.168.200.130:9092`）启动docker中的`kafka容器`
> 2.  启动消费者`ConsumerQuickStart`的`main`函数
> 3.  启动`kafkastream`的`mian`函数
> 4.  启动生产者`ProducerQuickStart`的`main`函数

#### 5. 控制台打印结果： ####

![在这里插入图片描述][630f1f46a5fa42248c4ac2c486319b78.png]  
![在这里插入图片描述][c3e175d036c248f1b1d4e3f7066308d7.png]

> **整个过程：**  
> 生产者向`kafka`中发送了5条`“hello kafka”`消息，topic均为`itcast-topic-input`。kafkastream监听这个topic，每10秒进行一次流式处理，将`“hello kakfa”`字符串分割，并统计每个单词出现的次数。然后转为`kstream`，发送消息到kafka中的`topic=itcast-topic-out”`。消费者监听`“itcast-topic-out”`的topic，消费消息。

### 三、Springboot整合kafkaStream ###

#### 1. 配置文件新增 ####

`application.yml`

#### 2. 在微服务中新增配置类 ####

`KafkaStreamConfig.java`

package com.kafka.config;
    
    import lombok.Getter;
    import lombok.Setter;
    import org.apache.kafka.common.serialization.Serdes;
    import org.apache.kafka.streams.StreamsConfig;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.kafka.annotation.EnableKafkaStreams;
    import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
    import org.springframework.kafka.config.KafkaStreamsConfiguration;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * 通过重新注册KafkaStreamsConfiguration对象，设置自定配置参数
     */
    
    @Setter
    @Getter
    @Configuration
    @EnableKafkaStreams
    @ConfigurationProperties(prefix="kafka")
    public class KafkaStreamConfig {
        
        private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;
        private String hosts;
        private String group;
        @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
        public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
        
            Map<String, Object> props = new HashMap<>();
            props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
            props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");
            props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");
            props.put(StreamsConfig.RETRIES_CONFIG, 10);
            props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            return new KafkaStreamsConfiguration(props);
        }
    }

#### 3. 使用kafkaStream监听消息 ####

`KafkaStreamHelloListener.java`

package com.kafka.stream;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.kafka.streams.KeyValue;
    import org.apache.kafka.streams.StreamsBuilder;
    import org.apache.kafka.streams.kstream.KStream;
    import org.apache.kafka.streams.kstream.TimeWindows;
    import org.apache.kafka.streams.kstream.ValueMapper;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import java.time.Duration;
    import java.util.Arrays;
    
    @Configuration
    @Slf4j
    public class KafkaStreamHelloListener {
        
    
        @Bean
        public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        
            //创建kstream对象，同时指定从那个topic中接收消息
            KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
            stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
        
                @Override
                public Iterable<String> apply(String value) {
        
                    return Arrays.asList(value.split(" "));
                }
            })
                    //根据value进行聚合分组
                    .groupBy((key,value)->value)
                    //聚合计算时间间隔
                    .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                    //求单词的个数
                    .count()
                    .toStream()
                    //处理后的结果转换为string字符串
                    .map((key,value)->{
        
                        System.out.println("key:"+key+",value:"+value);
                        return new KeyValue<>(key.key().toString(),value.toString());
                    })
                    //发送消息
                    .to("itcast-topic-out");
            return stream;
        }
    }

#### 测试： ####

启动`springboot`应用程序，运行之前的`ProducerQuickStart`来生产消息，约10秒后，看到`kafkaStream`消息的处理结果  
![在这里插入图片描述][785204e6ef5149d9a6596e68380829e3.png]

**说明`kafkaStream`接收到消息并将多条消息进行了统一处理。**

> 参考（推荐阅读）：
> 
> 1.  [https://cloud.tencent.com/developer/article/2100664][https_cloud.tencent.com_developer_article_2100664]
> 2.  [https://www.cnblogs.com/tree1123/p/11457851.html][https_www.cnblogs.com_tree1123_p_11457851.html]

[4a041929a1ab4730a0095f4079fc2f5d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/22b42082cc1144a586a1e1e97b2d13ed.png
[6cce41129c2d4807abab124913d7017b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/53342f1f5fc34ecb9ef4f0ad0319d7ef.png
[0234903f20284a0faf3f9283d3dc7b58.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/ee1c25cd4d7c440b8ff85107e501bf08.png
[cf379e9599904bd2845711d01ad8ccff.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/56b59edf779849d99d8bfcb789c2bbd4.png
[fea09e50e2324b5fa410e0fb229e5da2.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/f1a732400f5846f2b0f38d10386594f9.png
[86446e71e551453b986e66146d642dd8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/826ad112f8fa43c2a8ec53e29b096657.png
[647f17195a2b4fbab5b00856d8c1a709.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/97ce50357a51418ca364bdcbdd587993.png
[630f1f46a5fa42248c4ac2c486319b78.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/0ca5bb36ef0749c9bfbbee3b60090530.png
[c3e175d036c248f1b1d4e3f7066308d7.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/62d4f1a3f8e2404db362318bed896152.png
[785204e6ef5149d9a6596e68380829e3.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/9115d12cc9a14de18c0667b4c81173b4.png
[https_cloud.tencent.com_developer_article_2100664]: https://cloud.tencent.com/developer/article/2100664
[https_www.cnblogs.com_tree1123_p_11457851.html]: https://www.cnblogs.com/tree1123/p/11457851.html