如何有效测试 CompletableFuture

最后修改:2024年1月8日

by baeldung

1. 引言

CompletableFuture 是 Java 中进行异步编程的利器。它提供了一种便捷的方式来链式组合异步任务并处理其结果。通常在需要执行异步操作，并在后续阶段消费或处理其结果的场景中使用。

但由于其异步特性，对 CompletableFuture 进行单元测试可能颇具挑战。传统依赖顺序执行的测试方法往往难以捕捉异步代码的细微差别。本文将探讨两种有效测试 CompletableFuture 的方法：黑盒测试和状态测试。

2. 异步代码测试的挑战

异步代码因其非阻塞和并发执行特性引入了测试难点，传统测试方法难以应对。主要挑战包括：

  • ⏱️ 时序问题：异步操作引入了时间依赖，难以控制执行流并在特定时间点验证代码行为。依赖顺序执行的传统测试方法可能不适用
  • ⚠️ 异常处理：异步操作可能抛出异常，必须确保代码能优雅处理异常且不会静默失败。单元测试应覆盖各种场景验证异常处理机制
  • 🔄 竞态条件：异步代码可能导致竞态条件，多个线程同时访问或修改共享数据时可能产生意外结果
  • 📊 测试覆盖率：由于交互复杂性和潜在的非确定性结果，实现异步代码的全面测试覆盖颇具挑战

3. 黑盒测试

黑盒测试专注于代码的外部行为而不关心内部实现。这种方法适合从用户视角验证异步代码行为。测试者只需知道代码的输入和预期输出。

使用黑盒测试 CompletableFuture 时，我们重点关注：

  • ✅ 成功完成：验证 CompletableFuture 能成功完成并返回预期结果
  • ❌ 异常处理：验证 CompletableFuture 能优雅处理异常，避免静默失败
  • ⏳ 超时处理：确保 CompletableFuture 在遇到超时时行为符合预期

可使用 Mockito 等模拟框架来模拟被测 CompletableFuture 的依赖项，从而隔离测试环境。

3.1. 被测系统

我们将测试一个名为 processAsync() 的方法，它封装了异步数据检索和组合过程。该方法接受 Microservice 对象列表作为输入，返回 CompletableFuture<String>。每个 Microservice 对象代表一个能执行异步检索操作的微服务。

processAsync() 使用两个辅助方法 fetchDataAsync() 和 combineResults() 处理异步数据检索和组合：

CompletableFuture<String> processAsync(List<Microservice> microservices) {
    List<CompletableFuture<String>> dataFetchFutures = fetchDataAsync(microservices);
    return combineResults(dataFetchFutures);
}

fetchDataAsync() 方法遍历微服务列表，为每个调用 retrieveAsync()，返回 CompletableFuture<String> 列表：

private List<CompletableFuture<String>> fetchDataAsync(List<Microservice> microservices) {
    return microservices.stream()
        .map(client -> client.retrieveAsync(""))
        .collect(Collectors.toList());
}

combineResults() 方法使用 CompletableFuture.allOf() 等待所有 Future 完成。完成后映射 Future，合并结果并返回单个字符串：

private CompletableFuture<String> combineResults(List<CompletableFuture<String>> dataFetchFutures) {
    return CompletableFuture.allOf(dataFetchFutures.toArray(new CompletableFuture[0]))
      .thenApply(v -> dataFetchFutures.stream()
        .map(future -> future.exceptionally(ex -> {
            throw new CompletionException(ex);
        })
          .join())
      .collect(Collectors.joining()));
}

3.2. 测试用例：验证成功数据检索与组合

此测试验证 processAsync() 方法能正确从多个微服务检索数据并组合结果：

@Test
public void givenAsyncTask_whenProcessingAsyncSucceed_thenReturnSuccess() 
  throws ExecutionException, InterruptedException {
    Microservice mockMicroserviceA = mock(Microservice.class);
    Microservice mockMicroserviceB = mock(Microservice.class);

    when(mockMicroserviceA.retrieveAsync(any())).thenReturn(CompletableFuture.completedFuture("Hello"));
    when(mockMicroserviceB.retrieveAsync(any())).thenReturn(CompletableFuture.completedFuture("World"));

    CompletableFuture<String> resultFuture = processAsync(List.of(mockMicroserviceA, mockMicroserviceB));

    String result = resultFuture.get();
    assertEquals("HelloWorld", result);
}

3.3. 测试用例：验证微服务抛出异常时的异常处理

此测试验证当微服务抛出异常时，processAsync() 方法会抛出 ExecutionException，并断言异常消息与微服务抛出的一致：

@Test
public void givenAsyncTask_whenProcessingAsyncWithException_thenReturnException() 
  throws ExecutionException, InterruptedException {
    Microservice mockMicroserviceA = mock(Microservice.class);
    Microservice mockMicroserviceB = mock(Microservice.class);

    when(mockMicroserviceA.retrieveAsync(any())).thenReturn(CompletableFuture.completedFuture("Hello"));
    when(mockMicroserviceB.retrieveAsync(any()))
      .thenReturn(CompletableFuture.failedFuture(new RuntimeException("Simulated Exception")));
    CompletableFuture<String> resultFuture = processAsync(List.of(mockMicroserviceA, mockMicroserviceB));

    ExecutionException exception = assertThrows(ExecutionException.class, resultFuture::get);
    assertEquals("Simulated Exception", exception.getCause().getMessage());
}

3.4. 测试用例：验证组合结果超时时的超时处理

此测试尝试在 300 毫秒超时内获取 processAsync() 的组合结果，断言超时时抛出 TimeoutException：

@Test
public void givenAsyncTask_whenProcessingAsyncWithTimeout_thenHandleTimeoutException() 
  throws ExecutionException, InterruptedException {
    Microservice mockMicroserviceA = mock(Microservice.class);
    Microservice mockMicroserviceB = mock(Microservice.class);

    Executor delayedExecutor = CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS);
    when(mockMicroserviceA.retrieveAsync(any()))
      .thenReturn(CompletableFuture.supplyAsync(() -> "Hello", delayedExecutor));
    Executor delayedExecutor2 = CompletableFuture.delayedExecutor(500, TimeUnit.MILLISECONDS);
    when(mockMicroserviceB.retrieveAsync(any()))
      .thenReturn(CompletableFuture.supplyAsync(() -> "World", delayedExecutor2));
    CompletableFuture<String> resultFuture = processAsync(List.of(mockMicroserviceA, mockMicroserviceB));

    assertThrows(TimeoutException.class, () -> resultFuture.get(300, TimeUnit.MILLISECONDS));
}

上述代码使用 CompletableFuture.delayedExecutor() 创建执行器，分别延迟 200 和 500 毫秒完成 retrieveAsync() 调用，模拟微服务延迟，验证 processAsync() 能正确处理超时。

4. 状态测试

状态测试专注于验证代码执行过程中的状态转换。这种方法特别适合测试异步代码，因为它允许测试者跟踪代码在不同状态间的进展，确保正确转换。

例如，我们可以验证当异步任务成功完成时，CompletableFuture 转换到完成状态；当发生异常或任务被取消时，转换到失败状态。

4.1. 测试用例：验证成功完成后的状态

此测试验证当所有组成 CompletableFuture 成功完成时，实例转换到完成状态：

@Test
public void givenCompletableFuture_whenCompleted_thenStateIsDone() {
    Executor delayedExecutor = CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS);
    CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "Hello", delayedExecutor);
    CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> " World");
    CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> "!");
    CompletableFuture<String>[] cfs = new CompletableFuture[] { cf1, cf2, cf3 };

    CompletableFuture<Void> allCf = CompletableFuture.allOf(cfs);

    assertFalse(allCf.isDone());
    allCf.join();
    String result = Arrays.stream(cfs)
      .map(CompletableFuture::join)
      .collect(Collectors.joining());

    assertFalse(allCf.isCancelled());
    assertTrue(allCf.isDone());
    assertFalse(allCf.isCompletedExceptionally());
}

4.2. 测试用例：验证异常完成后的状态

此测试验证当组成 CompletableFuture 中的 cf2 异常完成时，allCf 转换到异常状态：

@Test
public void givenCompletableFuture_whenCompletedWithException_thenStateIsCompletedExceptionally() 
  throws ExecutionException, InterruptedException {
    Executor delayedExecutor = CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS);
    CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "Hello", delayedExecutor);
    CompletableFuture<String> cf2 = CompletableFuture.failedFuture(new RuntimeException("Simulated Exception"));
    CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> "!");
    CompletableFuture<String>[] cfs = new CompletableFuture[] { cf1, cf2, cf3 };

    CompletableFuture<Void> allCf = CompletableFuture.allOf(cfs);

    assertFalse(allCf.isDone());
    assertFalse(allCf.isCompletedExceptionally());

    assertThrows(CompletionException.class, allCf::join);

    assertTrue(allCf.isCompletedExceptionally());
    assertTrue(allCf.isDone());
    assertFalse(allCf.isCancelled());
}

4.3. 测试用例：验证任务取消后的状态

此测试验证当使用 cancel(true) 方法取消 allCf 时，它转换到取消状态：

@Test
public void givenCompletableFuture_whenCancelled_thenStateIsCancelled() 
  throws ExecutionException, InterruptedException {
    Executor delayedExecutor = CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS);
    CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> "Hello", delayedExecutor);
    CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> " World");
    CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> "!");
    CompletableFuture<String>[] cfs = new CompletableFuture[] { cf1, cf2, cf3 };

    CompletableFuture<Void> allCf = CompletableFuture.allOf(cfs);
    assertFalse(allCf.isDone());
    assertFalse(allCf.isCompletedExceptionally());

    allCf.cancel(true);

    assertTrue(allCf.isCancelled());
    assertTrue(allCf.isDone());
}

5. 总结

总之，由于 CompletableFuture 的异步特性，对其进行单元测试可能颇具挑战。但这是编写健壮可维护异步代码的重要环节。通过黑盒测试和状态测试方法，我们可以在各种条件下评估 CompletableFuture 代码的行为，确保其按预期运行并优雅处理潜在异常。

示例代码可在 GitHub 获取。

原始标题:How to Effectively Unit Test CompletableFuture

« 上一篇： Java Weekly, Issue 519
» 下一篇： Get Index of First Element Matching Boolean Using Java Streams | Baeldung