Spring Boot 中 Kafka 生产者日志可见但消费者无日志，通常因消息未真正提交至 Broker 导致——KafkaTemplate 默认异步发送且缓冲未刷新，需显式调用 flush() 或启用同步模式。
spring boot 中 kafka 生产者日志可见但消费者无日志，通常因消息未真正提交至 broker 导致——kafkatemplate 默认异步发送且缓冲未刷新，需显式调用 flush() 或启用同步模式。
在 Spring Kafka 应用中，KafkaTemplate.send() 方法默认是异步非阻塞的：它仅将消息写入本地内存缓冲区（producer buffer），随后立即返回，而不会等待消息被成功写入 Kafka Broker。这意味着：
- 生产者日志（如
LOGGER.info("Message sent...")）会立刻打印； - 但消息尚未落盘或完成 Leader 副本复制，消费者线程（由
@KafkaListener驱动）无法感知该消息，自然不会触发消费逻辑，也就不会输出消费者日志。
✅ 正确做法是确保消息真正提交到 Broker 后再视为“已发送”。有两种主流方式：
✅ 方案一：启用同步发送（推荐用于调试/开发）
修改 KafkaTemplate 配置，强制每次 send() 后等待响应（即 acks=all + linger.ms=0 + 同步 get()）：
@Bean
public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
KafkaTemplate<String, String> template = new KafkaTemplate<>(producerFactory);
template.setObservationEnabled(true); // 可选：启用 Micrometer 观察
return template;
}并在 KafkaProducer.sendMessage() 中改为同步调用：
public void sendMessage(String message) {
ListenableFuture<SendResult<String, String>> future =
kafkaTemplate.send("java", message);
try {
SendResult<String, String> result = future.get(10, TimeUnit.SECONDS); // 阻塞等待，超时10秒
LOGGER.info("Message sent & acknowledged: offset={}", result.getRecordMetadata().offset());
} catch (ExecutionException | InterruptedException | TimeoutException e) {
LOGGER.error("Failed to send message", e);
Thread.currentThread().interrupt();
}
}⚠️ 注意：
future.get()会阻塞当前线程，不适用于高吞吐生产场景，仅建议在开发验证、单元测试或低频触发场景使用。
✅ 方案二：手动 flush（适用于批量/高性能场景）
若使用 send() 多次后统一提交（如批量发送），需在关键节点调用 kafkaTemplate.flush()：
public void sendBatch(List<String> messages) {
messages.forEach(msg -> kafkaTemplate.send("java", msg));
kafkaTemplate.flush(); // 强制刷出所有缓冲消息
LOGGER.info("Batch flushed to broker");
}? 验证技巧：启动应用后，可通过 Kafka 命令行工具确认消息是否真实存在：
# 查看 topic 分区最新 offset kafka-run-class.sh kafka.tools.GetOffsetShell \ --bootstrap-server localhost:9092 \ --topic java \ --time -1
? 补充检查项（避免其他干扰）
- 确保
KafkaConsumer类被 Spring 容器正确扫描（类路径含@Service且包在@ComponentScan范围内）； - 检查
application.yml中 consumer 配置是否生效：group-id必须与@KafkaListener的groupId完全一致（注意大小写和空格）； - 若使用 Spring Boot 3.x + Kafka 3.3+，确认
spring-kafka版本 ≥ 3.0.12（修复了部分 listener 初始化延迟问题）； - 日志级别确认：
logging.level.org.springframework.kafka.listener=DEBUG可查看 listener 是否注册成功及 poll 行为。
? 总结：Kafka 消息“发送” ≠ “已提交”。生产环境应结合 acks=all、retries 和幂等性（enable.idempotence=true）保障可靠性；而日志缺失的第一排查点，永远是——消息是否真的到达了 Broker？
