springboot kafka批量消费
作者:互联网
配置
/** * @ClassName KafkaConsumerConfig * @Description 这里描述 * @Author admin * @Date 2021/2/25 15:02 */ @Configuration public class KafkaConsumerConfig { @Value("${spring.kafka.consumer.bootstrap-servers}") private String servers; @Value("${spring.kafka.consumer.enable-auto-commit}") private boolean enableAutoCommit; @Value("${spring.kafka.consumer.group-id}") private String groupId; @Value("${spring.kafka.consumer.auto-offset-reset}") private String autoOffsetReset; @Value("${spring.kafka.consumer.concurrency}") private int concurrency; // 批量接受数据 @Bean KafkaListenerContainerFactory batchFactory() { ConcurrentKafkaListenerContainerFactoryfactory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs())); factory.setBatchListener(true); // 开启批量监听 factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);//设置手动提交ackMode return factory; } @Bean public MapconsumerConfigs() { Mapprops = new HashMap<>(); props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10); //设置每次接收Message的数量 props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100"); props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000); props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);//开启幂等性。 return props; } }
监听消费
@KafkaListener(topics = "first_top0105",containerFactory="batchFactory") public void listen(List<ConsumerRecord> list, Acknowledgment ack) { Listmessages = new ArrayList<>(); for (ConsumerRecord record : list) { //判断是否为null Optional kafkaMessage = Optional.ofNullable(record.value()); //判断是否包含值 if(kafkaMessage.isPresent()){ //得到Optional实例中的值 Object message = kafkaMessage.get(); } log.info("收到消息的topic名称:" + record.topic() + ",key:" + record.key() + ",分区位置:" + record.partition() + ", 下标" + record.offset()+",内容value:"+kafkaMessage); } log.info("线程:{} consumer-id6 消费->{}",Thread.currentThread(), messages.toString()); ack.acknowledge(); }
标签:ConsumerConfig,springboot,批量,kafka,record,props,put,CONFIG 来源: https://blog.51cto.com/u_15193673/2847850