如何为 onFailure 事件(Spring、Kafka)设置超时? [英] How to set timeout for onFailure event (Spring, Kafka)?

问题描述

我正在尝试在 Spring MVC 中实现向 Kafka 发送消息的异步 REST 方法.一切正常，但是当服务器不可用时，onFailure 事件会被处理很长时间.例如，如何将 ListenableFuture 中的响应时间限制为三秒.

I'm trying to implement an asynchronous REST method of sending a message to Kafka in Spring MVC. Everything works, but when the server is unavailable, the onFailure event is processed for a long time. How to limit the response time in ListenableFuture for example to three seconds.

这是我的代码:

@Autowired
KafkaTemplate<String, String> kafkaTemplate;

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

@RequestMapping("/test")
DeferredResult<ResponseEntity<?>> test(
        @RequestParam(value = "message") String message
) {

    DeferredResult<ResponseEntity<?>> deferredResult = new DeferredResult<>();
    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, "testKey", message);

    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

        @Override
        public void onSuccess(SendResult<String, String> sendResult) {
            ResponseEntity<String> responseEntity = new ResponseEntity<>("SUCCESS", HttpStatus.OK);
            deferredResult.setResult(responseEntity);
        }

        @Override
        public void onFailure(Throwable ex) {
            ResponseEntity<String> responseEntity = new ResponseEntity<>("FAILURE", HttpStatus.OK);
            deferredResult.setResult(responseEntity);
        }

    });

    return deferredResult;
}

我尝试使用 Kafka 的 REQUEST_TIMEOUT_MS_CONFIG 属性和 ListenableFuture 的 .get(long timeout, TimeUnit unit) 方法，但没有得到想要的结果.

I tried to use a REQUEST_TIMEOUT_MS_CONFIG property of Kafka and .get(long timeout, TimeUnit unit) method of ListenableFuture but havn't got desired result.

推荐答案

那是因为生产者阻塞了 60 秒(默认).

That's because the producer blocks for 60 seconds (by default).

请参阅 KafkaDocumentation 中的 max.block.ms 以了解生产者配置.

See max.block.ms in the KafkaDocumentation for producer configuration.

max.block.ms 配置控制 KafkaProducer.send() 和 KafkaProducer.partitionsFor() 将阻塞多长时间.这些方法可以因为缓冲区已满或元数据不可用而被阻塞.在用户提供的序列化程序或分区程序中将不计入此超时.

max.block.ms The configuration controls how long KafkaProducer.send() and KafkaProducer.partitionsFor() will block.These methods can be blocked either because the buffer is full or metadata unavailable.Blocking in the user-supplied serializers or partitioner will not be counted against this timeout.

这篇关于如何为 onFailure 事件(Spring、Kafka)设置超时?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！