kafka批量消费配置 刺骨的言语ヽ痛彻心扉 2022-09-02 11:50 127阅读 0赞 spring.kafka.bootstrap-servers=127.0.0.1:9092 spring.kafka.consumer.group-id=order_ck_group spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=false #最多拉取1000条数据 spring.kafka.consumer.max-poll-records=1000 #最长1000毫秒 spring.kafka.listener.type=batch spring.kafka.listener.idle-between-polls=1000 spring.kafka.listener.ack-mode=manual_immediate #topic名称 spring.kafka.topic=order_topic 消费者代码 @Component public class OrderConsumer { @KafkaListener(topics = "#{'${spring.kafka.topic}'.split(',')}") public void orderConsumer(List<ConsumerRecord> data, Acknowledgment acknowledgment) { /*Iterator<ConsumerRecord> iterator = data.iterator(); List<JSONObject> list = new ArrayList<>(); JSONObject jsonObject = new JSONObject(); while (iterator.hasNext()) { jsonObject = JSON.parseObject(iterator.next().value().toString()); list.add(jsonObject); }*/ System.out.println(data.size()); acknowledgment.acknowledge(); } }
还没有评论,来说两句吧...