本文介绍通过手动管理 kafkaconsumer 替代 @kafkalistener，实现对消息消费速率（如每秒 10 万条）和内存队列容量（如 100 万条）的精准控制，避免 oom 风险。
本文介绍通过手动管理 kafkaconsumer 替代 @kafkalistener，实现对消息消费速率（如每秒 10 万条）和内存队列容量（如 100 万条）的精准控制，避免 oom 风险。
在 Spring Batch 集成 Kafka 场景中，直接使用 @KafkaListener + 无界队列（如 ConcurrentLinkedQueue）极易导致内存溢出——因为监听器默认以最大吞吐优先，持续拉取消息并堆积至 JVM 堆内存。根本解法是放弃被动监听，转为显式、可控的主动轮询机制，即用原生 KafkaConsumer 替代注解驱动模型。
✅ 正确实践：手动轮询 + 容量感知
首先，构建带限流参数的 KafkaConsumer 实例：
Map<String, Object> consumerConfig = Map.of(
"bootstrap.servers", "localhost:9092",
"key.deserializer", StringDeserializer.class.getName(),
"value.deserializer", StringDeserializer.class.getName(),
"group.id", "batch-processor",
"max.poll.records", 10000, // 单次 poll 最多返回 1 万条（防单次过载）
"fetch.max.wait.ms", 500, // 若无足够数据，最多等待 500ms
"fetch.min.bytes", 1024 // 至少累积 1KB 数据才返回（提升吞吐效率）
);
KafkaConsumer<String, Message> kafkaConsumer = new KafkaConsumer<>(consumerConfig);
kafkaConsumer.subscribe(List.of("topic"));⚠️ 注意：
max.poll.records是核心限流参数，需结合业务吞吐与内存预算设定（例如设为 10,000，则每轮最多入队 1 万条）。
接着，在消费逻辑中加入队列容量守门机制：
private final ConcurrentLinkedQueue<Message> queue = new ConcurrentLinkedQueue<>();
private final int MAX_QUEUE_SIZE = 1_000_000; // 100 万条硬上限
public void receive() {
// 【关键】先检查队列是否已满，满则跳过本次 poll（暂停消费）
if (queue.size() >= MAX_QUEUE_SIZE) {
return; // 或记录日志、触发告警
}
ConsumerRecords<String, Message> records = kafkaConsumer.poll(Duration.ofMillis(100));
records.forEach(record -> {
if (queue.size() < MAX_QUEUE_SIZE) { // 双重校验，避免竞态
queue.add(record.value());
}
});
}该设计实现了真正的“背压”（backpressure）：当队列达阈值时，receive() 不执行 poll()，Kafka 消费器自然暂停拉取，待下游批处理清空队列后自动恢复。
? 进阶：实现动态速率控制（如 10 万条/秒）
若需精确控速（非仅靠 max.poll.records），可引入令牌桶或滑动窗口计数器：
private final RateLimiter rateLimiter = RateLimiter.create(100_000.0); // 10 万 tokens/sec
public void receive() {
if (queue.size() >= MAX_QUEUE_SIZE || !rateLimiter.tryAcquire()) {
return;
}
ConsumerRecords<String, Message> records = kafkaConsumer.poll(Duration.ofMillis(10));
// ... 同上入队逻辑
}? 提示：
RateLimiter来自 Guava，需引入com.google.guava:guava。也可用Resilience4j的RateLimiter实现更细粒度熔断。
? 总结与最佳实践
- ❌ 避免
@KafkaListener+ 无界队列：无法实现反压，OOM 高风险； - ✅ 用
KafkaConsumer.poll()主动控制：配合max.poll.records和队列 size 校验，实现安全背压； - ✅ 设置合理
fetch.min.bytes/fetch.max.wait.ms：平衡延迟与吞吐； - ✅ 批处理侧需保证
queue.poll()高效消费（建议用BlockingQueue+ 独立线程池）； - ✅ 生产环境务必监控
queue.size()和consumer lag，设置告警阈值。
通过以上改造，你将获得一个内存可控、速率可调、故障可溯的健壮 Kafka 批处理流水线。
