如何配置@KafkaListener单次调用接收的消息数量？

可以通过配置Kafka消费者的核心参数，调整单个@KafkaListener方法单次调用时接收的消息数量。结合你提到的消息大小（500KB~20MB），需要同时调整条数限制和字节限制两类参数，否则会因单条消息过大导致实际拉取条数远低于预期。

关键配置参数 #

• max.poll.records：控制消费者单次poll()操作拉取的最大消息条数，是你期望的单次接收条数上限。
• fetch.max.bytes：消费者单次拉取请求能获取的总字节数（所有分区合计），默认值50MB。如果单条消息接近50MB，最多只能拉取2条。
• max.partition.fetch.bytes：单个分区单次拉取的最大字节数，必须大于业务中最大的单条消息大小（比如20MB的消息，这个值要设为25MB以上），否则消费者无法拉取这条超大消息。

配置方式 #

方式1：Spring Boot配置文件（application.yml） #

spring:
  kafka:
    consumer:
      max-poll-records: 10  # 按需设置单次期望接收的最大条数
      fetch-max-bytes: 104857600  # 100MB，足够容纳多条大消息
      max-partition-fetch-bytes: 26214400  # 25MB，大于20MB的单条消息上限
      group-id: your-consumer-group-id  # 替换为你的消费者组ID
      bootstrap-servers: your-kafka-servers  # 替换为你的Kafka地址

方式2：直接在@KafkaListener中指定properties #

@KafkaListener(
    topics = "${kafkalistener.topic}", 
    batch = "true",
    concurrency = "1",
    properties = {
        "max.poll.records=10",
        "fetch.max.bytes=104857600",
        "max.partition.fetch.bytes=26214400",
        "group.id=your-consumer-group-id"
    }
)
public void listen(List<ConsumerRecord<?,String>> records,Acknowledgment ack) {
    // 处理消息逻辑
    ack.acknowledge(); // 手动提交偏移量
}

注意事项 #

• max.poll.records是“软限制”：实际拉取的消息条数会受fetch.max.bytes和max.partition.fetch.bytes的约束，当累计字节数达到限制时，即使未达到条数上限也会停止拉取。
• 必须保证max.partition.fetch.bytes > 单条消息最大大小：否则消费者会抛出FetchOffsetOutOfRangeException或类似错误，无法正常拉取消息。
• 手动提交偏移量时，要确保所有消息处理完成后再调用ack.acknowledge()，避免消息重复消费。

内容的提问来源于stack exchange，提问作者user3001150