Kafka 消费者避免内存溢出的核心不是依赖某种自动反压协议，而是通过配置拉取批次大小、控制单次 poll 记录数，并在应用层确保处理速度不低于拉取速度。
先说结论：Kafka 消费者采用拉取模型，反压能力取决于客户端配置与业务处理逻辑的配合，需限制单次加载数据量并谨慎管理 offset 提交。
- 适合：Java/Scala 等主流客户端开发场景，尤其是处理耗时波动较大的业务。
- 先准备：评估单条消息最大体积与 JVM 堆内存大小，确认当前消费延迟状况。
- 验收：观察 Full GC 频率是否降低，消费者 Lag 指标是否稳定且无持续增长。
关键配置参数
如果没有时间深入调优，可优先调整以下两个参数来限制单次进入内存的数据量：
1. 调小 max.poll.records：减少每次 poll 方法返回的消息条数。默认通常为 500，建议根据内存情况调整为 100 左右。
2. 调小 max.partition.fetch.bytes：限制每个分区单次拉取的字节上限。默认 1MB，大消息场景需调大但需警惕内存占用。
注意：过度调小可能导致网络请求频繁，需结合吞吐量要求平衡。
代码实战：手动提交 Offset
关闭自动提交 enable.auto.commit=false。在业务逻辑处理成功后再提交 offset。这样确保只有成功处理的消息才会提交 offset，避免数据丢失，但未提交的消息重启后会重消费。
Java 客户端示例：
try {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {// 业务处理逻辑process(record);}// 处理成功后同步提交 offsetconsumer.commitSync();
} catch (Exception e) {// 捕获异常，此次不提交 offset，重启后会重消费log.error("Processing failed", e);
}代码实战：应用层背压
在代码层面，如果内部处理队列已满，暂停调用 poll() 方法，直到队列有空闲。这是应用层最直接的反压手段。
// 假设 businessQueue 为内部处理队列
if (businessQueue.size() < MAX_QUEUE_SIZE) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));// 将记录放入队列for (ConsumerRecord<String, String> record : records) {businessQueue.offer(record);}
} else {// 队列已满，暂停拉取，避免内存累积Thread.sleep(100);
}验证方法
1. 监控 JVM 内存：使用 JMX 或监控工具观察 Heap Usage 是否出现持续攀升不下降的情况。
2. 检查 GC 日志：确认 Full GC 的频率是否减少，停顿时间是否缩短。
3. 观察 Consumer Lag：使用 Kafka 自带命令或监控面板查看消费延迟，确保调整参数后延迟没有无限堆积。
常见坑
1. rebalance 风暴：如果处理时间超过 max.poll.interval.ms，消费者会被踢出组，导致重复拉取和内存压力加剧。
2. 大消息陷阱：如果单个消息超过 max.partition.fetch.bytes 或 broker 端的 message.max.bytes，可能导致无法消费或内存瞬间峰值。
3. 误用缓存：不要在消费者内部使用无界队列缓存消息，这会让反压配置失效。
参考来源
Apache Kafka Documentation, Consumer Configuration, https://kafka.apache.org/documentation/#consumerconfigs
原文链接：https://www.zjcp.cc/ask/11637.html