SpringKafka发送消息并获取响应方法

本篇文章主要是结合我之前面试的各种经历和实战开发中遇到的问题解决经验整理的，希望这篇《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<SendResult<String, Student>> future = 
        this.studentKafkaTemplate.send(topicName, student);

    try {
        // 阻塞等待最多 5 秒，超时抛出 TimeoutException
        SendResult<String, Student> 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<StudentDto> 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 发送结果并返回业务对象”这一需求的标准、简洁且可控的方案，适用于大多数需要强一致响应语义的业务场景。

今天带大家了解了的相关知识，希望对你有所帮助；关于文章的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~