Spring Kafka生产者不能异步工作

Question

我正在使用非阻塞（异步）向 Kafka 发送消息：

    ListenableFuture<SendResult<Integer, String>> future = template.send(record);
    future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {

        @Override
        public void onSuccess(SendResult<Integer, String> result) {
            handleSuccess(data);
        }

        @Override
        public void onFailure(Throwable ex) {
            handleFailure(data, record, ex);
        }

    });

当发送操作完成它的工作时，这完美地工作。

但是当出现连接问题时（例如服务器宕机），结果变成非异步的，并且该方法保持阻塞状态直到 max.block.ms 的持续时间结束。

Answer 1

这在 Async Kafka producer 中很自然。你有两个选择

1. 要么减少 max.block.ms，但不要减少太多。
2. 您可以等待确认

也可以为 onCompletion() 创建回调函数