CompletableFuture не завершается по тайм-ауту

Question

У меня есть метод read, который читает из входного потока. Я хочу завершить будущее, если read не завершится после тайм-аута.

public static void main(String[] args) {
   CompletableFuture<?> future = new CompletableFuture<>();
   future
      .thenRunAsync(() -> {
         try {
            read(data);
         } catch (IOException e) {
            future.completeExceptionally(e);
         }
      })
      .orTimeout(1, TimeUnit.SECONDS);

   future.join();
} 

Но когда я запускаю этот код, он не завершается sh по таймауту и ​​все равно ожидает входной поток.

Answer 1

Есть как минимум две проблемы с вашим кодом:

1. Ничего не выполняется. Вы создаете CompletableFuture и затем вызываете thenRunAsync для него. Этап, созданный thenRunAsync, сработает только после завершения предыдущего этапа . Поскольку вы никогда не завершите оригинал CompletableFuture, этого никогда не произойдет. Вы также в конечном итоге присоединяетесь к будущему, которое никогда не завершится.

2. Вы присоединяетесь к неправильному CompletableFuture. Методы, такие как thenRunAsync и orTimeout возвращает новый экземпляр , который создает своего рода «цепочку» этапов. Каждый этап запускается завершением своего «родительского» этапа. Чтобы полностью понять это, я рекомендую прочитать документацию CompletionStage.

Вот пример вашего кода, работающего, как я подозреваю, вы хотите:

public static void main(String[] args) {
  CompletableFuture.runAsync(
          () -> {
            try {
              read(data);
            } catch (IOException ex) {
              throw new UncheckedIOException(ex);
            }
          })
      .orTimeout(1L, TimeUnit.SECONDS)
      .join();
}

Некоторые примечания:

• Используется CompletableFuture#runAsync(Runnable) для создания "изначальной" стадии. Этот этап завершится, когда Runnable завершится, и общий ForkJoinPool используется для выполнения Runnable.

• Если и когда брошено, UncheckedIOException в runAsync Этап приведет к исключительному завершению этапа.

• Метод #join() вызывается для экземпляра, возвращенного вызовом orTimeout(1L, TimeUnit.SECONDS). Теперь, если время ожидания истекло, вызов join() вызовет CompletionException обертывание TimeoutException.

Предупреждение: Вызов read не прерывается ¹ по истечении времени ожидания, что означает, что он будет продолжать выполняться в фоновом режиме. Это связано с тем, что CompletableFuture не имеет ссылки на исполняющие потоки и поэтому не может их прерывать.

---

^{1. Предполагая, что прерывание будет даже иметь эффект .}