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

SpringBoot项目集成kafka及常规配置_springboot集成kafka配置

作者：Cpp五条 | 2024-04-15 15:20:28

赞

踩

springboot集成kafka配置

desc:

        使用 spring-kafka 的api，在springboot项目中集成kafka能力，封装配置。

0.引入依赖

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

1.kafka相关配置

1.1 KafkaConfiguration 公共配置

  1. @Data
  2. @Configuration
  3. public class KafkaConfiguration {
  4.  
  5.     /**
  6.      * 主机地址
  7.      */
  8.     @Value("${kafka.server-host}")
  9.     private String bootstrapServers;
  10.  
  11.     /**
  12.      * sasl 认证账号
  13.      */
  14.     @Value("${iot.kafka.sasl.username:admin}")
  15.     private String userName;
  16.  
  17.     /**
  18.      * sasl 密码
  19.      */
  20.     @Value("${iot.kafka.sasl.password:iot@2021}")
  21.     private String password;
  22.  
  23.  
  24. }

1.2 KafkaConsumerConfiguration 消费者配置

  1. @Data
  2. @Configuration
  3. public class KafkaConsumerConfiguration {
  4.  
  5.     /**
  6.      * 默认组id
  7.      */
  8.     @Value("${iot.kafka.consumer.properties.group-id:default-group}")
  9.     private String groupId;
  10.  
  11.     @Value("${iot.kafka.consumer.properties.fetch-max-wait:5000}")
  12.     private Integer fetchMaxWait;
  13.  
  14.     /**
  15.      * 此设置限制每次调用poll返回的消息数，这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值，可以减少poll间隔，减少重新平衡分组的
  16.      */
  17.     @Value("${iot.kafka.consumer.properties.max-poll-record:100}")
  18.     private Integer maxPollRecordsConfig;
  19.  
  20.     /**
  21.      * 增大poll的间隔，可以为消费者提供更多的时间去处理返回的消息（调用poll（long）返回的消息，通常返回的消息都是一批），缺点是此值越大将会延迟组重新平衡。
  22.      */
  23.     @Value("${iot.kafka.consumer.properties.max-poll-interval-ms:100}")
  24.     private Integer maxPollIntervalConfig;
  25.  
  26.     /**
  27.      * 是否开启自动提交
  28.      */
  29.     @Value("${iot.kafka.consumer.properties.enable-auto-commit:#{false}}")
  30.     private boolean enableAutoCommitConfig;
  31.  
  32.     /**
  33.      * 消费策略
  34.      * earliest  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
  35.      * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
  36.      * none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
  37.      */
  38.     @Value("${iot.kafka.consumer.properties.auto-offset-reset:earliest}")
  39.     private String autoOffsetResetConfig;
  40.  
  41.     @Value("${iot.kafka.consumer.properties.auto-commit-interval:1000}")
  42.     private String autoCommitIntervalMsConfig;
  43.  
  44.     @Value("${iot.kafka.consumer.properties.session-timeout:30000}")
  45.     private String sessionTimeoutMsConfig;
  46.  
  47. }

1.3 KafkaListenerConfiguration 监听配置

  1. @Data
  2. @Configuration
  3. public class KafkaListenerConfiguration {
  4.  
  5.     /**
  6.      * 启用线程数（提高并发）
  7.      */
  8.     @Value("${iot.kafka.listener.concurrency:3}")
  9.     private Integer concurrency;
  10.  
  11.     /**
  12.      * 手动提交的方式，当enable-auto-commit: false时起作用
  13.      * manual:手动调用Acknowledgment.acknowledge()后立即提交
  14.      * record:当每一条记录被消费者监听器（ListenerConsumer）处理之后提交
  15.      * batch:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后提交
  16.      * time: 当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，距离上次提交时间大于TIME时提交
  17.      * count:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后，被处理record数量大于等于COUNT时提交
  18.      * count_time:当每一批poll()的数据被消费者监听器（ListenerConsumer）处理之后, 手动调用Acknowledgment.acknowledge()后提交
  19.      */
  20.     @Value("${iot.kafka.listener.ack-mode:manual_immediate}")
  21.     private String ackMode;
  22.  
  23.     /**
  24.      * 消费超时时间
  25.      */
  26.     @Value("${iot.kafka.listener.poll-timeout:3000}")
  27.     private Long pollTimeout;
  28.  
  29.     /**
  30.      * 是否开启批量处理
  31.      */
  32.     @Value("${iot.kafka.listener.batch_listener:#{true}}")
  33.     private Boolean batchListener;
  34.  
  35. }

1.4 KafkaProducerConfiguration 生产者配置

  1. @Data
  2. @Configuration
  3. public class KafkaProducerConfiguration {
  4.     /**
  5.      * 重试次数 默认值0
  6.      */
  7.     @Value("${iot.kafka.producer.retries:0}")
  8.     private Integer retries;
  9.  
  10.     /**
  11.      * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
  12.      * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
  13.      * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
  14.      */
  15.     @Value("${iot.kafka.producer.acks:all}")
  16.     private String acks;
  17.  
  18.     /**
  19.      * 指定缓存的大小，生产者缓存每个分区未发送的消息。默认 16384
  20.      */
  21.     @Value("${iot.kafka.producer.batch-size:16384}")
  22.     private Integer batchSize;
  23.  
  24.     /**
  25.      * 生产者发送请求之前等待一段时间，设置等待时间是希望更多地消息填补到未满的批中。 默认 30
  26.      */
  27.     @Value("${iot.kafka.producer.properties.linger.ms:30}")
  28.     private Integer lingerMs;
  29.  
  30.     /**
  31.      * 通过KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里，然后把很多消息收集成一个一个的Batch，再发送到Broker上去的 默认32m
  32.      */
  33.     @Value("${iot.kafka.producer.buffer-memory:33554432}")
  34.     private Integer bufferMemory;
  35.  
  36. }

2.工厂配置

2.1 ConsumerFactoryBuilder 消费者工厂

  1. @Configuration
  2. public class ConsumerFactoryBuilder {
  3.  
  4.     @Autowired
  5.     private KafkaConfiguration kafkaConfiguration;
  6.  
  7.     @Autowired
  8.     private KafkaConsumerConfiguration kafkaConsumerConfiguration;
  9.  
  10.     @Autowired
  11.     private KafkaListenerConfiguration kafkaListenerConfiguration;
  12.  
  13.     /**
  14.      * 消费者配置
  15.      *
  16.      * @return properties
  17.      */
  18.     @Bean
  19.     public Map<String, Object> consumerConfigs() {
  20.         Map<String, Object> props = new ConcurrentHashMap<>();
  21.         //配置地址
  22.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getBootstrapServers());
  23.         //消费者组 默认组id
  24.         props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerConfiguration.getGroupId());
  25.         //是否开启自动提交
  26.         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafkaConsumerConfiguration.isEnableAutoCommitConfig());
  27.         /* 消费策略
  28.          * earliest  当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
  29.          * latest 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
  30.          * none topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
  31.          */
  32.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafkaConsumerConfiguration.getAutoOffsetResetConfig());
  33.         //消费者默认等待服务响应时间(毫秒)
  34.         props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, kafkaConsumerConfiguration.getFetchMaxWait());
  35.         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafkaConsumerConfiguration.getAutoCommitIntervalMsConfig());
  36.         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafkaConsumerConfiguration.getSessionTimeoutMsConfig());
  37.         props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaConsumerConfiguration.getMaxPollRecordsConfig());
  38.         props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaConsumerConfiguration.getMaxPollIntervalConfig());
  39.         //key序列化器选择
  40.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  41.         //value序列化器选择
  42.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
  43.         //设置sasl认证
  44.         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
  45.         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
  46.         props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + kafkaConfiguration.getUserName() + "' password='" + kafkaConfiguration.getPassword() + "';");
  47.         return props;
  48.     }
  49.  
  50.     /**
  51.      * kafka消费者工厂
  52.      */
  53.     @Bean
  54.     public ConsumerFactory<Object, Object> consumerFactory() {
  55.         return new DefaultKafkaConsumerFactory(consumerConfigs());
  56.     }
  57.  
  58.  
  59.     /**
  60.      * 监听工厂
  61.      */
  62.     @Bean
  63.     KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaListenerContainerFactory() {
  64.         ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
  65.         factory.setConsumerFactory(consumerFactory());
  66.         //线程数
  67.         factory.setConcurrency(kafkaListenerConfiguration.getConcurrency());
  68.         //手动提交
  69.         factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
  70.         //开启批量处理
  71.         factory.setBatchListener(kafkaListenerConfiguration.getBatchListener());
  72.         factory.getContainerProperties().setPollTimeout(kafkaListenerConfiguration.getPollTimeout());
  73.         return factory;
  74.     }
  75.  
  76. }

2.2 ProducerFactoryBuilder 生产者工厂

  1.  
  2. @Configuration
  3. public class ProducerFactoryBuilder {
  4.  
  5.     @Autowired
  6.     private KafkaConfiguration kafkaConfiguration;
  7.  
  8.     @Autowired
  9.     private KafkaProducerConfiguration kafkaProducerConfiguration;
  10.  
  11.  
  12.     /**
  13.      * 生产者配置
  14.      *
  15.      * @return 配置
  16.      */
  17.     @Bean
  18.     public Map<String, Object> producerConfigs() {
  19.         Map<String, Object> props = new HashMap<>(11);
  20.         //kafka server地址
  21.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfiguration.getBootstrapServers());
  22.         /*
  23.          * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为 - 1。
  24.          * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
  25.          * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1 的设置。
  26.          */
  27.         props.put(ProducerConfig.ACKS_CONFIG, kafkaProducerConfiguration.getAcks());
  28.         //消息发送失败重试次数
  29.         props.put(ProducerConfig.RETRIES_CONFIG, kafkaProducerConfiguration.getRetries());
  30.         //去缓冲区中一次拉16k的数据，发送到broker
  31.         props.put(ProducerConfig.BATCH_SIZE_CONFIG, kafkaProducerConfiguration.getBatchSize());
  32.         // 批量发送，延迟为30毫秒，如果30ms内凑不够batch则强制发送，提高并发
  33.         props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaProducerConfiguration.getLingerMs());
  34.         //设置缓存区大小 32m
  35.         props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaProducerConfiguration.getBufferMemory());
  36.         //key序列化器选择
  37.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  38.         //value序列化器选择
  39.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
  40.         //设置sasl认证
  41.         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
  42.         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
  43.         props.put(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='" + kafkaConfiguration.getUserName() + "' password='" + kafkaConfiguration.getPassword() + "';");
  44.         return props;
  45.     }
  46.  
  47.  
  48.     /**
  49.      * Producer Template 配置
  50.      */
  51.     @Bean
  52.     public KafkaTemplate<String, String> kafkaTemplate() {
  53.         Map<String, Object> stringObjectMap = producerConfigs();
  54.         DefaultKafkaProducerFactory<String, String> objectObjectDefaultKafkaProducerFactory = new DefaultKafkaProducerFactory<>(stringObjectMap);
  55.         return new KafkaTemplate<>(objectObjectDefaultKafkaProducerFactory);
  56.     }
  57. }

3.配置文件

需配置 kafka.server-host 其他在代码中均已配置默认值

  1. kafka:
  2.   server-host: 192.168.1.113:9048

4.使用示例

  1. @Slf4j
  2. @RestController
  3. @RequestMapping("test")
  4. public class TestController {
  5.  
  6.     @Autowired
  7.     private KafkaTemplate<String, String> kafkaTemplate;
  8.  
  9.  
  10.     @PostMapping("test")
  11.     public RestResult test() {
  12.         kafkaTemplate.send("TOPIC_NAME", 0, "key", "this is a message");
  13.         return RestResult.wrapSuccessResponse();
  14.     }
  15.  
  16.     @KafkaListener(topics = "TOPIC_NAME", groupId = "MyGroup1", containerFactory = "kafkaListenerContainerFactory")
  17.     public void kafkaListener(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
  18.         for (ConsumerRecord item : records) {
  19.             System.out.printf("topic is %s, offset is %d,partition is %s, value is %s \n", item.topic(), item.offset(), item.partition(), item.value());
  20.             log.info("topic is : {}, offset is : {},partition is : {}, value is : {}",item.topic(), item.offset(), item.partition(), item.value());
  21.         }
  22.         ack.acknowledge();
  23.     }
  24.  
  25.  
  26. }

-- 230619更正 增加配置文件及默认值说明。

以上。

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

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

  

闽ICP备14008679号