如何为Spring Kafka的onFailure事件设置超时时间？

我来给你梳理下怎么解决这个问题哈！你遇到的情况是Kafka服务器不可用时，onFailure回调触发太慢，导致前端一直等待响应。我们可以通过给DeferredResult设置超时，再结合超时回调逻辑，确保3秒内给前端返回结果，不管消息发送成功、失败还是超时。

核心思路 #

• 给DeferredResult直接设置超时时间，让Spring MVC在超时后主动触发回调
• 在超时回调里尝试取消Kafka的发送任务（如果支持的话），避免资源浪费
• 在Kafka发送的成功/失败回调里，先判断DeferredResult是否已经超时，避免重复设置结果

完整代码实现 #

首先，我们需要注入必要的Bean，然后实现带超时的异步接口：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.concurrent.ScheduledExecutorService;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import org.springframework.util.concurrent.ListenableFuture;
import java.util.concurrent.Executors;

@RestController
public class KafkaAsyncController {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${spring.kafka.topic}")
    private String topic;

    // 配置一个定时线程池用于超时任务
    @Autowired
    private ScheduledExecutorService scheduledExecutorService;

    @RequestMapping("/test")
    public DeferredResult<ResponseEntity<?>> test(@RequestParam(value = "message", required = true) String message) {
        // 初始化DeferredResult，设置3秒超时时间
        DeferredResult<ResponseEntity<?>> deferredResult = new DeferredResult<>(3000L);

        // 发送消息到Kafka，获取ListenableFuture
        ListenableFuture<SendResult<String, String>> sendFuture = kafkaTemplate.send(topic, message);

        // 超时回调：3秒后自动触发
        deferredResult.onTimeout(() -> {
            // 尝试取消Kafka发送任务（部分Kafka版本可能不支持，不影响前端响应）
            sendFuture.cancel(true);
            // 返回超时响应给前端
            deferredResult.setResult(ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT)
                    .body("请求超时，消息发送未完成，请稍后重试"));
        });

        // Kafka发送成功的回调
        sendFuture.addCallback(
                result -> {
                    // 确保DeferredResult还未超时或设置结果，避免重复操作
                    if (!deferredResult.isSetOrExpired()) {
                        deferredResult.setResult(ResponseEntity.ok(
                                "消息发送成功，偏移量：" + result.getRecordMetadata().offset()));
                    }
                },
                // Kafka发送失败的回调
                ex -> {
                    if (!deferredResult.isSetOrExpired()) {
                        deferredResult.setResult(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
                                .body("消息发送失败：" + ex.getMessage()));
                    }
                }
        );

        return deferredResult;
    }

    // 配置定时线程池Bean
    @Bean
    public ScheduledExecutorService scheduledExecutorService() {
        return Executors.newSingleThreadScheduledExecutor();
    }
}

额外优化方案：用CompletableFuture增强超时控制 #

如果你的KafkaTemplate版本不支持取消ListenableFuture，可以用CompletableFuture来包装，实现更可靠的超时逻辑：

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.CompletableFuture;

// 把ListenableFuture转换成CompletableFuture
CompletableFuture<SendResult<String, String>> completableFuture = new CompletableFuture<>();
sendFuture.addCallback(completableFuture::complete, completableFuture::completeExceptionally);

// 创建一个3秒后的超时任务
CompletableFuture<Object> timeoutTask = CompletableFuture.delayedExecutor(3, TimeUnit.SECONDS, scheduledExecutorService)
        .submit(() -> {
            completableFuture.completeExceptionally(new TimeoutException("消息发送超时"));
            return null;
        });

// 监听任意一个任务完成（发送完成或超时）
CompletableFuture.anyOf(completableFuture, timeoutTask).whenComplete((result, ex) -> {
    if (ex instanceof TimeoutException) {
        deferredResult.setResult(ResponseEntity.status(HttpStatus.REQUEST_TIMEOUT).body("请求超时"));
    } else if (ex != null) {
        deferredResult.setResult(ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("发送失败：" + ex.getMessage()));
    } else {
        SendResult<String, String> sendResult = (SendResult<String, String>) result;
        deferredResult.setResult(ResponseEntity.ok("发送成功，偏移量：" + sendResult.getRecordMetadata().offset()));
    }
});

这样不管Kafka的发送任务是否支持取消，都能确保3秒内给前端返回结果，不会让用户一直等待。

内容的提问来源于stack exchange，提问作者V. Perfilev