kafka 生产者默认使用异步发送+内存缓冲，若未显式 flush 或启用同步模式，消息可能滞留在客户端缓冲区而未真正提交至 broker，导致消费者无法消费、日志也不触发。
kafka 生产者默认使用异步发送+内存缓冲，若未显式 flush 或启用同步模式，消息可能滞留在客户端缓冲区而未真正提交至 broker，导致消费者无法消费、日志也不触发。
在 Spring Kafka 应用中，KafkaTemplate.send() 方法本质上是异步非阻塞的：它仅将消息写入生产者的本地缓冲区（RecordAccumulator），随后由后台 Sender 线程批量拉取并发送至 Kafka Broker。这意味着：
- ✅
LOGGER.info(...)在send()调用后立即执行（仅表示“已提交到缓冲区”）； - ❌ 消费者
@KafkaListener不会触发，因为消息尚未落盘或被 Broker 提交（log.flush.interval.ms/linger.ms/batch.size等参数共同影响实际发送时机）； - ?
kafka-console-consumer也看不到消息，进一步证实消息未真正到达 Broker。
✅ 正确做法：强制刷新或启用同步发送
方案一：调用 flush()（推荐用于调试/低频场景）
@Service
public class KafkaProducer {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducer.class);
private final KafkaTemplate<string string> kafkaTemplate;
public KafkaProducer(KafkaTemplate<string string> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void sendMessage(String message) {
LOGGER.info("Message sent -> {}", message);
kafkaTemplate.send("java", message).get(); // 阻塞等待发送完成（含 flush）
// 或显式 flush（适用于批量发送后统一提交）：
// kafkaTemplate.flush();
}
}</string></string>
⚠️ 注意：
.get()会抛出ExecutionException/InterruptedException，需捕获处理；生产环境慎用高频.get()，因会显著降低吞吐。
方案二：配置 spring.kafka.producer.properties.acks=all + 合理 linger.ms
确保消息真正被 Broker 提交（而非仅接收）：
Apache Superset Dashboard and SQL Exploration Skill
下载
Apache Superset 是一个广泛采用的开源 BI 平台，用于 SQL 探索、图表构建和仪表板交付。当代理需要查询仓库数据、组装仪表板或使用成熟的分析界面解释指标而不是临时笔记本代码时，此技能非常有用。
spring:
kafka:
bootstrap-servers: localhost:9092
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
properties:
acks: all # 等待所有 ISR 副本确认（强一致性）
linger.ms: 5 # 最大等待 5ms 合并批次（降低延迟）
batch.size: 16384 # 16KB 批次（平衡吞吐与延迟）
方案三：启用 idempotence（幂等性生产者，隐式保障 flush 安全性）
spring:
kafka:
producer:
properties:
enable.idempotence: true # 自动启用幂等性，要求 acks=all & max.in.flight.requests.per.connection=1
? 验证消费者是否正常工作
确保 KafkaConsumer 类被 Spring 容器正确扫描并注册监听器：
- 检查是否添加了
@EnableKafka（通常 Spring Boot 2.3+ 自动启用）； - 确认
application.yml中 consumer 配置生效（spring.kafka.consumer.*前缀）； - 查看启动日志中是否有类似
Setting newly created group id或Subscribed to topic(s): [java]的提示。
? 总结
| 问题现象 | 根本原因 | 解决关键 |
|---|---|---|
| Producer 日志有，Consumer 无响应 | 消息滞留于生产者缓冲区，未提交至 Broker | 强制同步（.get()）、显式 flush() 或优化 acks/linger.ms |
| 控制台消费者也收不到 | Broker 侧无数据写入 | 检查 bootstrap-servers 地址、Topic 是否自动创建、ACL 权限（如适用） |
只要消息未真正写入 Kafka Broker 的日志（LogSegment），任何消费者（包括 @KafkaListener 和 kafka-console-consumer）都不可见——这并非日志配置错误，而是 Kafka 异步架构的必然行为。理解并合理驾驭缓冲与确认机制，是构建可靠消息链路的第一课。
