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

SpringBoot 集成 Kafka 配置_spring kafka配置

作者：码创造者 | 2024-08-09 02:43:58

赞

踩

spring kafka配置

原生模式

  1.         <dependency>
  2.             <groupId>org.apache.kafka</groupId>
  3.             <artifactId>kafka-clients</artifactId>
  4.             <version>3.0.0</version>
  5.         </dependency>

自定义分区器

  1. /**
  2.  * 自定义分区器
  3.  *
  4.  * @Author: chen yang
  5.  * @Date: 2023/5/7 11:34
  6.  */
  7. public class CustomerPartitioner implements Partitioner {
  8.     @Override
  9.     public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
  10.  
  11.         List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
  12.         int partition = 2;
  13.         return partitionInfos.size() % partition;
  14.     }
  15.  
  16.     @Override
  17.     public void close() {
  18.  
  19.     }
  20.  
  21.     @Override
  22.     public void configure(Map<String, ?> configs) {
  23.  
  24.     }
  25. }

生产者

  1. /**
  2.  * @Author: chen yang
  3.  * @Date: 2023/5/6 21:59
  4.  */
  5. public class Producer {
  6.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  7.  
  8.         Properties properties = new Properties();
  9.         // 连接集群
  10.         properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  11.  
  12.         // 序列化类型
  13.         properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  14.         properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  15.  
  16.         // 添加自定义分区器
  17.         properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomerPartitioner.class.getName());
  18.  
  19.  
  20.         // batch.size：批次大小，默认 16K
  21.         properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
  22.         // linger.ms：等待时间，默认 0
  23.         properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
  24.         // RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
  25.         properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
  26.         // compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
  27.         properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
  28.  
  29.  
  30.         // 设置 ack
  31.         // 0:不需要等待数据落盘应答，1:需要等 leader 落盘应答，-1(all):需要等 leader 和所有的 follower(isr队列) 落盘应答
  32.         // type: String, valid values [all, -1, 0, 1], default: all
  33.         properties.put(ProducerConfig.ACKS_CONFIG, "all");
  34.  
  35.         // 设置重试次数，默认为 int 最大值
  36.         properties.put(ProducerConfig.RETRIES_CONFIG, 3);
  37.  
  38.         // 设置事务id
  39.         properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "default_transactional_id_23");
  40.  
  41.         KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
  42.  
  43.         // 开启事务
  44.         producer.initTransactions();
  45.         producer.beginTransaction();
  46.  
  47.         try {
  48.             // 异步发送
  49.             producer.send(new ProducerRecord<>("test", "this is async message"));
  50.  
  51.             producer.send(new ProducerRecord<>("test", "this is a async rollback message!"), new Callback() {
  52.                 @Override
  53.                 public void onCompletion(RecordMetadata recordMetadata, Exception e) {
  54.                     // 发送消息失败会自动重试，不需要在回调函数中手动重试
  55.                     if (Objects.isNull(e)){
  56.                         System.out.println("result: " + recordMetadata.topic() + ", partitions: " + recordMetadata.partition());
  57.                     }
  58.                 }
  59.             });
  60.  
  61.             // 同步发送，只需要在异步发送的基础上再调用 get() 犯法即可
  62.             producer.send(new ProducerRecord<>("test", 1,"","this is sync message")).get();
  63.             producer.commitTransaction();
  64.  
  65.         }catch (Exception e){
  66.             producer.abortTransaction();
  67.         }finally {
  68.             producer.close();
  69.         }
  70.     }
  71. }

消费者

  1. /**
  2.  * @Author: chen yang
  3.  * @Date: 2023/7/9 10:37
  4.  */
  5. public class Consumer {
  6.  
  7.     public static void main(String[] args) {
  8.  
  9.         Properties properties = new Properties();
  10.         properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  11.         properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  12.         properties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer_01");
  13.         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  14.         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  15.  
  16.         try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
  17.               // 订阅 topic
  18. //            ArrayList<String> topics = new ArrayList<>();
  19. //            consumer.subscribe(topics);
  20.  
  21.             // 订阅 topic 下的 partition
  22.             ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
  23.             topicPartitions.add(new TopicPartition("night_topic", 1));
  24.             consumer.assign(topicPartitions);
  25.  
  26.             // 从指定的 offset 开始消费
  27. //            consumer.seek(new TopicPartition("night_topic", 1), 3);
  28.  
  29.             ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(2));
  30.             consumerRecords.forEach(System.out::println);
  31.  
  32.             // 手动提交 offset
  33.             consumer.commitAsync();
  34.         } catch (Exception e) {
  35.             throw new RuntimeException(e);
  36.         }
  37.     }
  38. }

KafkaTemplate

  1.         <dependency>
  2.             <groupId>org.springframework.kafka</groupId>
  3.             <artifactId>spring-kafka</artifactId>
  4.         </dependency>

配置文件

  1. spring:
  2.   kafka:
  3.     bootstrap-servers: localhost:9092
  4.     producer:
  5.       batch-size: 16384
  6.       acks: -1
  7.       retries: 10
  8.       transaction-id-prefix: transaction_05
  9.       buffer-memory: 33554432
  10.       key-serializer: org.apache.kafka.common.serialization.StringSerializer
  11.       value-serializer: org.apache.kafka.common.serialization.StringSerializer
  12.       properties:
  13.         linger:
  14.           ms: 2000
  15.         partitioner:
  16.           class: com.night.config.CustomerPartitionHandler
  17.     consumer:
  18.       group-id: g_01
  19.       enable-auto-commit: false
  20.       auto-offset-reset: latest
  21.       max-poll-records: 500
  22. #      auto-commit-interval: 2000  autoCommit = false
  23.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  24.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
  25.       properties:
  26.         session:
  27.           timeout:
  28.             ms: 120000 # 消费会话超时时间（超过这个时间 consumer 没有发送心跳，就会触发 rebalance 操作）
  29.         request:
  30.           timeout:
  31.             ms: 18000
  32.  
  33.     listener:
  34.       missing-topics-fatal: false # consumer listener topics 不存在时，启动项目就会报错
  35.       type: batch

自定义分区器

  1. /**
  2.  * @Author: chen yang
  3.  * @Date: 2023/7/8 11:02
  4.  */
  5. @Component
  6. public class CustomerPartitionHandler implements Partitioner {
  7.  
  8.     @Override
  9.     public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
  10.         if (value.toString().contains("二")){
  11.             return 2;
  12.         }else if (value.toString().contains("一")){
  13.             return 1;
  14.         }else {
  15.             return 0;
  16.         }
  17.     }
  18.  
  19.     @Override
  20.     public void close() {
  21.  
  22.     }
  23.  
  24.     @Override
  25.     public void configure(Map<String, ?> configs) {
  26.  
  27.     }
  28. }

生产者

  1. @RestController
  2. @RequiredArgsConstructor
  3. public class HelloController {
  4.  
  5.     private final KafkaTemplate<String, String> kafkaTemplate;
  6.  
  7.     @GetMapping("/send")
  8.     @Transactional // 配置文件中设置了事务id，那么启用时要加上 该注解或者使用 kafka 事务处理
  9.     public Boolean send(String msg){
  10.         for (int i = 0; i < 10; i++) {
  11.             kafkaTemplate.send("night.topic", null, "night key - " + i, msg + " - " + i).addCallback(success -> {
  12.                 // 成功回调
  13.                 if (success == null || success.getRecordMetadata() == null){
  14.                     return;
  15.                 }
  16.                 String topic = success.getRecordMetadata().topic();
  17.                 int partition = success.getRecordMetadata().partition();
  18.                 long offset = success.getRecordMetadata().offset();
  19.                 String key = success.getProducerRecord().key();
  20.                 System.out.println("send topic:" + topic +", partition: " + partition + ", key:" + key + ", offset: " + offset);
  21.             }, failure -> {
  22.                 // 失败回调
  23.                 System.out.println("发送消息失败：" + failure.getMessage());
  24.             });
  25.         }
  26.         return true;
  27.     }
  28. }

消费者配置

消费数据过滤
  1. /**
  2.  * 消费数据过滤
  3.  *
  4.  * @Author: chen yang
  5.  * @Date: 2023/7/8 12:20
  6.  */
  7. @Component
  8. public class ConsumerFilterStrategy implements RecordFilterStrategy<String, String> {
  9.     @Override
  10.     public boolean filter(ConsumerRecord<String, String> consumerRecord) {
  11.         // return true: 丢弃消息
  12.         return consumerRecord.value().contains("无效数据");
  13.     }
  14. }

消费异常处理类
  1. /**
  2.  * 消费异常处理类
  3.  *
  4.  * @Author: chen yang
  5.  * @Date: 2023/7/8 11:55
  6.  */
  7. @Component
  8. public class ConsumerExceptionHandler implements ConsumerAwareListenerErrorHandler {
  9.  
  10.     @Override
  11.     public Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {
  12.         System.out.println("消费异常：" + message.getPayload() + ", ex: " + e.getMessage());
  13.         return null;
  14.     }
  15. }

消费者配置 
  1. /**
  2.  * @Author: chen yang
  3.  * @Date: 2023/7/8 12:22
  4.  */
  5. @Configuration
  6. @RequiredArgsConstructor
  7. public class KafkaConsumerConfig {
  8.  
  9.     private final KafkaTemplate<String, String> kafkaTemplate;
  10.  
  11.     private final ConsumerFilterStrategy consumerFilterStrategy;
  12.  
  13.  
  14.     @Bean
  15.     public ConcurrentKafkaListenerContainerFactory<String, String> filterContainerFactory(){
  16.         ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
  17.         // @SendTo 使用，console exception:  a KafkaTemplate is required to support replies
  18.         factory.setReplyTemplate(kafkaTemplate);
  19.  
  20.         factory.setConsumerFactory(initConsumerFactory());
  21.         // 设置并发量，小于或等于Topic的分区数,并且要在consumerFactory设置一次拉取的数量
  22.         factory.setConcurrency(1);
  23.  
  24.         // 设置为批量监听
  25.         factory.setBatchListener(true);
  26.  
  27.         // 配合RecordFilterStrategy使用，被过滤的信息将被丢弃
  28.         factory.setAckDiscarded(true);
  29.         factory.setRecordFilterStrategy(consumerFilterStrategy);
  30.         return factory;
  31.     }
  32.  
  33.  
  34.  
  35.     @Bean
  36.     public ConsumerFactory<String, String> initConsumerFactory(){
  37.         HashMap<String, Object> configs = new HashMap<>();
  38.         configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  39.         configs.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer_group_id_02");
  40.         configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
  41.         configs.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 10 * 1000);
  42.         configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  43.         configs.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
  44.         configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  45.         configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  46.         return new DefaultKafkaConsumerFactory<>(configs);
  47.     }
  48. }

@KafkaListener

  1. @Component
  2. public class HelloListener {
  3.  
  4.     @KafkaListener(topicPartitions = {
  5.             @TopicPartition(topic = "night.topic", partitions = {"0"})
  6.     }, errorHandler = "consumerExceptionHandler", containerFactory = "filterContainerFactory")
  7.     public void consumer0(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer){
  8.         System.out.println("first consumer receive list size: " + records.size());
  9.         records.forEach(System.out::println);
  10.         consumer.commitSync();
  11.     }
  12.  
  13.  
  14.     @KafkaListener(topicPartitions = {
  15.             @TopicPartition(topic = "night.topic", partitions = {"1"})
  16.     }, errorHandler = "consumerExceptionHandler")
  17.     @SendTo("singleTopic")
  18.     public String consumer1(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer){
  19.         System.out.println("second consumer receive list size: " + records.size());
  20.         records.forEach(System.out::println);
  21.         consumer.commitSync();
  22.         return "@SendTo annotation msg";
  23.     }
  24.  
  25.  
  26.     @KafkaListener(topicPartitions = {
  27.             @TopicPartition(topic = "night.topic", partitions = {"2"})
  28.     }, errorHandler = "consumerExceptionHandler")
  29.     public void consumer2(List<ConsumerRecord<String, String>> records, Consumer<String, String> consumer){
  30.         System.out.println("third consumer receive list size: " + records.size());
  31.         records.forEach(System.out::println);
  32.         consumer.commitSync();
  33.     }
  34. }

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

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

  

闽ICP备14008679号