如何在 Spring Kafka 中同步等待消息发送完成并返回响应对象

在 spring kafka 中，若需确保消息成功发送后再向客户端返回结果，应避免使用异步回调（如 `addcallback`），而改用 `listenablefuture.get()` 阻塞等待发送结果，并据此决定是否返回业务对象（如 `studentdto`）。

Spring 的 KafkaTemplate.send() 方法返回的是 ListenableFuture>，它本质上是异步的。但正如问题中所指出的：addCallback 中的 onSuccess 是纯回调，执行时机不可控，无法用于同步返回值给上层控制器（如 REST Controller）。此时，最直接、可靠且符合 Spring 生态实践的方式是主动阻塞获取结果。

✅ 正确做法：使用 future.get() 同步等待

public StudentDto publishStudentDto(Student student, String topicName) throws ExecutionException, InterruptedException {
    ListenableFuture> future = 
        thi
s.studentKafkaTemplate.send(topicName, student);

    try {
        // 阻塞等待最多 5 秒，超时抛出 TimeoutException
        SendResult result = future.get(5, TimeUnit.SECONDS);

        // 发送成功：记录日志并构造响应 DTO
        logger.info("Student published to topic: {} at offset {} partition {}", 
            topicName, 
            result.getRecordMetadata().offset(), 
            result.getRecordMetadata().partition());

        return StudentDto.from(student); // 假设提供静态工厂方法

    } catch (TimeoutException e) {
        logger.error("Kafka send timed out for student: {}", student, e);
        throw new RuntimeException("Failed to publish student: timeout waiting for Kafka response", e);
    } catch (ExecutionException e) {
        Throwable cause = e.getCause();
        logger.error("Kafka send failed for student: {}", student, cause);
        throw new RuntimeException("Failed to publish student to Kafka", cause);
    }
}

⚠️ 注意事项：务必设置超时时间（推荐 get(timeout, unit)）：避免线程无限阻塞，影响服务可用性；必须处理 InterruptedException 和 ExecutionException：前者表示线程被中断，后者包装了 Kafka 发送失败的真实异常（如网络错误、序列化失败、Broker 不可达等）；不要在高并发、低延迟场景滥用此方式：同步等待会占用 Web 容器线程（如 Tomcat 线程），若 Kafka 延迟高或不稳定，可能引发线程池耗尽；此时应考虑异步响应（如 WebSocket、轮询、或返回 202 Accepted + 异步任务 ID）；若项目已升级至 Spring Kafka 2.7+，推荐迁移到 CompletableFuture + @Async 或 WebFlux 的 Mono 风格，以实现真正的非阻塞流式处理。

✅ 补充：Controller 层调用示例

@PostMapping("/students")
public ResponseEntity createStudent(@RequestBody Student student) {
    try {
        StudentDto dto = publishStudentDto(student, "student-topic");
        return ResponseEntity.ok(dto);
    } catch (Exception e) {
        return ResponseEntity.status(500).build();
    }
}

综上，future.get() 是解决“同步获取 Kafka 发送结果并返回业务对象”这一需求的标准、简洁且可控的方案，适用于大多数需要强一致响应语义的业务场景。