devbox
IT小白
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Kafka Stream流式计算

作者：IT小白 | 2024-05-29 00:47:41

赞

踩

kafka stream流

用例

- 日志分析

- 大屏看板统计

- 公交实时数据

- 实时文章分值计算

执行流程

对单词个数进行统计

演示

1. 生产者

  1. /**
  2.  * 生产者
  3.  */
  4. public class ProducerQuickStart {
  5.  
  6.     public static void main(String[] args) {
  7.         //1.kafka的配置信息
  8.         Properties properties = new Properties();
  9.         //kafka的连接地址
  10.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.128:9092");
  11.         //发送失败，失败的重试次数
  12.         properties.put(ProducerConfig.RETRIES_CONFIG,5);
  13.         //消息key的序列化器
  14.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  15.         //消息value的序列化器
  16.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
  17. //数据压缩
  18.         properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");
  19.         //2.生产者对象
  20.         KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
  21.  
  22.         //封装发送的消息
  23.         ProducerRecord<String,String> record = new ProducerRecord<String, String>("itcast-topic-input","key","hello 1");
  24.         RecordMetadata recordMetadata = null;
  25.         //3.发送消息
  26.         //同步
  27. //        try {
  28. //            recordMetadata = producer.send(record).get();
  29. //        } catch (InterruptedException e) {
  30. //            throw new RuntimeException(e);
  31. //        } catch (ExecutionException e) {
  32. //            throw new RuntimeException(e);
  33. //        }
  34.         //异步
  35.         //异步消息发送
  36.         for (int i = 0; i < 10; i++) {
  37.             producer.send(record, (recordMetadata1, e) -> {
  38.                 if(e != null){
  39.                     System.out.println("记录异常信息到日志表中");
  40.                 }
  41.                 System.out.println(recordMetadata1.offset());
  42.             });
  43.         }
  44.  
  45. //        System.out.println(recordMetadata.offset());
  46.  
  47.  
  48.         //4.关闭消息通道，必须关闭，否则消息发送不成功
  49.         producer.close();
  50.     }
  51.  
  52. }

2. Stream

  1. public class KafkaStreamQuickStart {
  2.  
  3.     public static void main(String[] args) {
  4.  
  5.         //kafka的配置信息
  6.         Properties prop = new Properties();
  7.         prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.128:9092");
  8. //        下方是Stream相关的序列化器
  9.         prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  10.         prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  11.         prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-quickstart");
  12.  
  13.         //stream 构建器
  14.         StreamsBuilder streamsBuilder = new StreamsBuilder();
  15.  
  16.         //流式计算
  17.         streamProcessor(streamsBuilder);
  18.  
  19.  
  20.         //创建kafkaStream对象
  21.         KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(),prop);
  22.         //开启流式计算
  23.         kafkaStreams.start();
  24.     }
  25.  
  26.     /**
  27.      * 流式计算
  28.      * 消息的内容：hello kafka  hello itcast
  29.      * @param streamsBuilder
  30.      */
  31.     private static void streamProcessor(StreamsBuilder streamsBuilder) {
  32.         //创建kstream对象，同时指定从那个topic中接收消息
  33.         KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
  34.  
  35.         /**
  36.          * 处理消息的value
  37.          */
  38.         stream.flatMapValues((ValueMapper<String, Iterable<String>>) value -> Arrays.asList(value.split(" ")))
  39.                 //按照value进行聚合处理
  40.                 .groupBy((key,value)->value)
  41.                 //时间窗口 每十秒执行一次
  42.                 .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
  43.                 //统计单词的个数
  44.                 .count()
  45.                 //转换为kStream
  46.                 .toStream()
  47.                 .map((key,value)->{
  48.                     System.out.println("key:"+key+",value:"+value);
  49.                     return new KeyValue<>(key.key().toString(),value.toString());
  50.                 })
  51.                 //发送消息
  52.                 .to("itcast-topic-out");
  53.     }
  54. }

3. 消费者

  1. public class ConsumerQuickStart {
  2.  
  3.     public static void main(String[] args) {
  4.         //1.添加kafka的配置信息
  5.         Properties properties = new Properties();
  6.         //kafka的连接地址
  7.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.128:9092");
  8.         //消费者组
  9.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
  10.         //消息的反序列化器
  11.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  12.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  13.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
  14.         //2.消费者对象
  15.         KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
  16.  
  17.         //3.订阅主题
  18.         consumer.subscribe(Collections.singletonList("itcast-topic-input"));
  19.  
  20.         //当前线程一直处于监听状态
  21.  
  22.         try{
  23.             while (true){
  24.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
  25.                 for (ConsumerRecord<String, String> record : records) {
  26.                     System.out.println(record.value());
  27.                     System.out.println(record.key());
  28.                     System.out.println(record.offset());
  29.                     //同步提交偏移量
  30. //                try {
  31. //                    consumer.commitSync();//同步提交当前最新的偏移量
  32. //                }catch (CommitFailedException e){
  33. //                    System.out.println("记录提交失败的异常："+e);
  34. //                }
  35.  
  36.                     //异步提交偏移量
  37. //                consumer.commitAsync((map, e) -> {
  38. //                    if(e!=null){
  39. //                        System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
  40. //                    }
  41. //                });
  42.  
  43.                 }
  44.  
  45.                 consumer.commitAsync();
  46.             }
  47.         }catch (RuntimeException e){
  48.             e.printStackTrace();
  49.         }finally {
  50.             try{
  51.                 //默认提交拉取的这些数据中最大的偏移量
  52.                 consumer.commitSync();
  53.             }finally {
  54.                 consumer.close();
  55.             }
  56.  
  57.         }
  58.  
  59.  
  60.  
  61.  
  62.     }
  63.  
  64. }

kafka流的配置主要包含四项

1. 编码器

2. 解码器

3. kafka地址

4. 分配ID

  1. Properties prop = new Properties();
  2. prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.128:9092");
  3. prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  4. prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  5. prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-quickstart");

Kafka Stream 集成SpringBoot

需要使用配置类对kafka stream进行配置

  1. @Setter
  2. @Getter
  3. @Configuration
  4. @EnableKafkaStreams
  5. @ConfigurationProperties(prefix="kafka")
  6. public class KafkaStreamConfig {
  7.     private static final int MAX_MESSAGE_SIZE = 16* 1024 * 1024;
  8.     private String hosts;
  9.     private String group;
  10.     @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
  11.     public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
  12.         Map<String, Object> props = new HashMap<>();
  13.         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
  14.         props.put(StreamsConfig.APPLICATION_ID_CONFIG, this.getGroup()+"_stream_aid");
  15.         props.put(StreamsConfig.CLIENT_ID_CONFIG, this.getGroup()+"_stream_cid");
  16.         props.put(StreamsConfig.RETRIES_CONFIG, 10);
  17.         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  18.         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
  19.         return new KafkaStreamsConfiguration(props);
  20.     }
  21. }

同时需要 指定需要进行stream的topic，以及相关逻辑操作

  1. @Configuration
  2. @Slf4j
  3. public class KafkaStreamHelloListener {
  4.  
  5.     @Bean
  6.     public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
  7.         //创建kstream对象，同时指定从那个topic中接收消息
  8.         KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
  9.         stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
  10.                     @Override
  11.                     public Iterable<String> apply(String value) {
  12.                         return Arrays.asList(value.split(" "));
  13.                     }
  14.                 })
  15.                 //根据value进行聚合分组
  16.                 .groupBy((key,value)->value)
  17.                 //聚合计算时间间隔
  18.                 .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
  19.                 //求单词的个数
  20.                 .count()
  21.                 .toStream()
  22.                 //处理后的结果转换为string字符串
  23.                 .map((key,value)->{
  24.                     System.out.println("key:"+key+",value:"+value);
  25.                     return new KeyValue<>(key.key().toString(),value.toString());
  26.                 })
  27.                 //发送消息
  28.                 .to("itcast-topic-out");
  29.         return stream;
  30.     }
  31. }

代码配置

  1. server:
  2.   port: 9991
  3. spring:
  4.   application:
  5.     name: kafka-demo
  6.   kafka:
  7.     bootstrap-servers: 192.168.200.128:9092
  8.     producer:
  9.       retries: 10
  10.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  11.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  12.       compressionType: lz4
  13.     consumer:
  14.       group-id: ${spring.application.name}-test
  15.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  16.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  17.  
  18. kafka:
  19.   hosts: 192.168.200.128:9092
  20.   group: ${spring.application.name}

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/IT小白/article/detail/639883
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号