Kafka批量消费实现

批量消费指的是一次性拉取一批消息，然后批量处理
依赖spring-kafka

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.2.4.RELEASE</version></dependency>

配置消费者工厂

@ConfigurationpublicclassKafkaConsumerConfig{@BeanpublicConsumerFactory<String,String>consumerFactory(){Map<String,Object>props=newHashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");props.put(ConsumerConfig.GROUP_ID_CONFIG,"batch-consumer-group");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class);// 批量消费配置props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,100);// 每次最多拉取100条props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG,10240);// 至少10KB才返回returnnewDefaultKafkaConsumerFactory<>(props);}@BeanpublicConcurrentKafkaListenerContainerFactory<String,String>batchFactory(ConsumerFactory<String,String>consumerFactory){ConcurrentKafkaListenerContainerFactory<String,String>factory=newConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory);factory.setBatchListener(true);// 启用批量监听// 手动提交模式factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);returnfactory;}}

实现批量消费监听器

@ServicepublicclassOrderBatchConsumer{@KafkaListener(topics="order-topic",containerFactory="batchFactory")publicvoidconsumeBatch(List<ConsumerRecord<String,String>>records,Acknowledgmentack){log.info("收到批量消息: {} 条",records.size());try{// 业务处理for(ConsumerRecord<String,String>record:records){processOrder(record.value());}// 全部成功才提交offsetack.acknowledge();log.info("批次处理成功,已提交offset");}catch(Exceptione){log.error("批量处理失败",e);// 不提交offset,等待重新投递}}}

批量消费可靠性保障
不能在finally中提交offset,因为不管消费是否成功都会提交offset

@KafkaListener(topics="topic",containerFactory="batchFactory")publicvoidconsume(List<ConsumerRecord<String,String>>records,Acknowledgmentack){try{processBatch(records);}finally{ack.acknowledge();// 无论成败都提交,会丢消息!}}

不能设置自动提交，自动提交模式下，批量消费有可能会丢消息。要改为手动提交

props.put("enable.auto.commit","true");// 自动提交// 处理过程中失败,但offset已自动提交,消息丢失

需要用到线程池的时候，需要等到全部消息处理成功才能提交。使用CompletionService实现

@ServicepublicclassReliableBatchConsumer{privatefinalExecutorServiceexecutor=Executors.newFixedThreadPool(10);@KafkaListener(topics="payment-topic",containerFactory="batchFactory")publicvoidconsumeBatch(List<ConsumerRecord<String,String>>records,Acknowledgmentack){CompletionService<Boolean>completionService=newExecutorCompletionService<>(executor);List<Future<Boolean>>futures=newArrayList<>();// 1. 提交所有任务到线程池并发处理for(ConsumerRecord<String,String>record:records){Callable<Boolean>task=()->{try{processPayment(record.value());returntrue;}catch(Exceptione){log.error("支付处理失败: {}",record.value(),e);returnfalse;}};futures.add(completionService.submit(task));}// 2. 等待所有任务完成并检查结果booleanallSuccess=true;try{for(inti=0;i<records.size();i++){Future<Boolean>future=completionService.take();if(!future.get()){allSuccess=false;break;// 发现失败立即终止}}}catch(Exceptione){allSuccess=false;log.error("任务执行异常",e);}// 3. 全部成功才提交offsetif(allSuccess){ack.acknowledge();log.info("批次全部成功,已提交offset");}else{log.warn("批次中有失败消息,不提交offset,等待重新投递");// 重新投递会导致重复消费,需要业务保证幂等性}}}

当有失败消息的时候，不提交offset，等待消息重投或者重新拉取，但是会有消息重复的情况，业务上要做好幂等。