栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 前沿技术 > 大数据 > 大数据系统

SpringBoot + Kafka 使用@KafkaListener注解批量消费

SpringBoot + Kafka 使用@KafkaListener注解批量消费

使用@KafkaListener 注解进行消费时,出现如下报错:

Cannot convert from [java.lang.String] to [org.apache.kafka.clients.consumer.ConsumerRecord]

原因是默认没有开启批量监听的,解决办法是设置注解的 containerFactory 属性。

完整代码如下

1)、批量 消费监听工厂类

@Configuration
public class KafkaConfiguration {

    
    @Bean
    public KafkaListenerContainerFactory batchFactory(KafkaProperties properties) {
        Map consumerProperties = properties.buildConsumerProperties();
        ConcurrentKafkaListenerContainerFactory factory = new
                ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProperties));
        factory.setBatchListener(true); // 开启批量监听
        return factory;
    }
}

2)、消费监听

@Component
@RequiredArgsConstructor
public class GxdcKafkaConsumer {

    private final GxdcService gxdcService;

    
    @KafkaListener(topics = {"${spring.kafka.consumer.topics.ods_dc_count_result}"}, containerFactory = "batchFactory")
    public void listen(ConsumerRecords records, Consumer consumer) {
        if (records.isEmpty()) {
            return;
        }
        // 消息逻辑处理
        for (ConsumerRecord record : records) {
            switch (GxdcKeyEnum.getInstance(record.key())) {
                case DC_POINT:
                    BikePointInfo pointInfo = JSON.parseObject(record.value(), BikePointInfo.class);
                    gxdcService.saveBikePointInfo(pointInfo);
                    break;
                case DC_ORDER:
                    OrderSummaryInfo summaryInfo = JSON.parseObject(record.value(), OrderSummaryInfo.class);
                    gxdcService.saveOrderSummaryInfo(summaryInfo);
                    break;
                default:
            }
        }

    }

}

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/687941.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号