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();
    }
}