凡科小程序制作_品牌网线和普通网线有什么区别_花钱推广的网络平台_百度搜索引擎优化方案

1、批量消费消息

1.1 配置参数

fetch.min.bytes：拉取的最小字节数

fetch.max.bytes：拉取的最大字节数

max.partition.fetch.bytes：分区拉取的最大字节数

fetch.max.wait.ms：拉取操作的最大等待时间

max.poll.records：拉取操作最大获取的记录数量

以上参数决定了一次消费能消费多少条消息。

    @Beanpublic KafkaListenerContainerFactory<?> kafkaBatchConsumerFactory() {return batchFactory(consumerConfig(KafkaConsumerGroupIdConstant.KAFKA_CONSUMER_BATCH_GROUP, KafkaConsumerResetOffsetEnum.EARLIEST.getType()));}

 1.2 消费代码

    @KafkaListener(groupId = KafkaConsumerGroupIdConstant.KAFKA_CONSUMER_BATCH_GROUP,topics = KafkaTopicConstant.KAFKA_TOPIC,containerFactory = "kafkaBatchConsumerFactory")public void batchConsume(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {String topicName = KafkaTopicConstant.KAFKA_TOPIC;try {List<String> valueList = records.stream().map(ConsumerRecord::value).collect(Collectors.toList());for (String value : valueList) {log.info("批量消费topic:{}, value:{}", topicName, value);}ack.acknowledge();} catch (Exception e) {log.error("kafka消费{}:", topicName, e);}}

1.3 消费结果 

2、单个消费消息

2.1 配置参数

    @Beanpublic KafkaListenerContainerFactory<?> kafkaSingleConsumerFactory() {ConcurrentKafkaListenerContainerFactory<Integer, String> resultFactory = batchFactory(consumerConfig(KafkaConsumerGroupIdConstant.KAFKA_CONSUMER_SINGLE_GROUP, KafkaConsumerResetOffsetEnum.EARLIEST.getType()));resultFactory.setBatchListener(false);return resultFactory;}

2.2 消费代码

@KafkaListener(groupId = KafkaConsumerGroupIdConstant.KAFKA_CONSUMER_SINGLE_GROUP,topics = KafkaTopicConstant.KAFKA_TOPIC,containerFactory = "kafkaSingleConsumerFactory")public void singleConsume(ConsumerRecord<String, String> records, Acknowledgment ack) {String topicName = KafkaTopicConstant.KAFKA_TOPIC;try {log.info("单个消费 topic:{}, value:{}", topicName, records.value());ack.acknowledge();} catch (Exception e) {log.error("kafka消费{}:", topicName, e);}}

2.3 消费结果 

注：

部分代码未能在博客内体现，请点击以下链接跳转至Gitee的xudongbase项目的kafka分支。

xudongbase: 主要是项目中可以用到的共通方法，现有easyexcel分支在持续更新中。欢迎大家Star和提交Issues。easyexcel分支：批量设置样式，批量添加批注，批量合并单元格，设置冻结行和列，设置行高列宽，隐藏行和列，绑定下拉框数据，设置水印，插入图片 - Gitee.comhttps://gitee.com/xudong_master/xudongbase/tree/kafka/