JUnit测试CompletableFuture.supplyAsync时Mock方法调用验证失败解决方案

问题描述 #

原本通过同步逻辑可验证RuntimeService.createProcessInstanceByKey方法被调用3次，将逻辑迁移到CompletableFuture.supplyAsync异步执行后，测试失败，提示方法未被调用，且逻辑进入exceptionally分支。

错误信息：

Wanted but not invoked: runtimeService.createProcessInstanceByKey(
);
-> at
Actually, there were zero interactions with this mock.StartProcessServiceTest.startProcesses(StartProcessServiceTest.java:52)

生产代码 #

import org.camunda.bpm.engine.RuntimeService; 

public class StartProcessService {
    private final RuntimeService runtimeService;
   
    public void handleInExkassodatenRequest(MyRequest request) {
        request.getDaten().forEach(entity -> {
            CompletableFuture.supplyAsync(() -> selectRequiredProcess(entity))
                    .thenAccept(caseId -> {
                        log.info("Case started ");
                    })
                    .exceptionally(err -> {
                        return null;
                    });
        });
    }

    private CompletableFuture<String> selectRequiredProcess(InExkassodaten entity) {
        startNewProcess(entity);
        return CompletableFuture.completedFuture(entity.getCaseId());
    }

    private void startNewProcess(MyObject myObject) {
        ProcessInstance processInstance = runtimeService
                .createProcessInstanceByKey("MY_KEY")
                .setVariable("MY_VAR", myObject)
                .execute();
    }
}

原测试代码 #

@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
class MyServiceTest {

    @InjectMocks
    private StartProcessService startProcessService;

    @Mock(answer = Answers.RETURNS_DEEP_STUBS)
    private RuntimeService runtimeService;


    @Test
    void startProcesses() {

        ProcessInstantiationBuilder processInstanceByKey = mock(ProcessInstantiationBuilderImpl.class);
        when(runtimeService.createProcessInstanceByKey(anyString())).thenReturn(processInstanceByKey);
        when(processInstanceByKey.setVariable(anyString(), any())).thenReturn(processInstanceByKey);
        when(processInstanceByKey.execute()).thenReturn(mock(ProcessInstance.class));

        startProcessService.handleInExkassodatenRequest(InExkassodatenFactory.createValidTestData());
        verify(runtimeService, times(3)).createProcessInstanceByKey(anyString());
    }
}

问题原因 #

1. 主线程提前结束：测试主线程在异步任务执行前就完成了验证逻辑，异步任务还未调用RuntimeService方法，测试已终止。
2. 异步线程无等待：supplyAsync默认使用ForkJoinPool.commonPool()执行任务，测试代码未等待所有异步任务完成就进行校验。
3. 异常无日志：exceptionally分支仅返回null，未记录异常信息，无法定位具体失败原因。

解决方案 #

方案一：修改生产+测试代码，等待异步任务完成 #

生产代码修改

修改方法返回所有异步任务的CompletableFuture集合，方便测试追踪并等待任务完成：

import org.camunda.bpm.engine.RuntimeService; 
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

public class StartProcessService {
    private final RuntimeService runtimeService;
   
    public List<CompletableFuture<Void>> handleInExkassodatenRequest(MyRequest request) {
        return request.getDaten().stream()
                .map(entity -> CompletableFuture.supplyAsync(() -> selectRequiredProcess(entity))
                        .thenAccept(caseId -> {
                            log.info("Case started ");
                        })
                        .exceptionally(err -> {
                            // 添加异常日志，排查失败原因
                            log.error("Failed to start process for entity: {}", entity, err);
                            return null;
                        }))
                .collect(Collectors.toList());
    }

    private CompletableFuture<String> selectRequiredProcess(InExkassodaten entity) {
        startNewProcess(entity);
        return CompletableFuture.completedFuture(entity.getCaseId());
    }

    private void startNewProcess(MyObject myObject) {
        ProcessInstance processInstance = runtimeService
                .createProcessInstanceByKey("MY_KEY")
                .setVariable("MY_VAR", myObject)
                .execute();
    }
}

测试代码修改

获取所有异步任务的Future，等待全部完成后再执行验证：

@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
class MyServiceTest {

    @InjectMocks
    private StartProcessService startProcessService;

    @Mock(answer = Answers.RETURNS_DEEP_STUBS)
    private RuntimeService runtimeService;


    @Test
    void startProcesses() throws InterruptedException, ExecutionException {
        ProcessInstantiationBuilder processInstanceByKey = mock(ProcessInstantiationBuilderImpl.class);
        when(runtimeService.createProcessInstanceByKey(anyString())).thenReturn(processInstanceByKey);
        when(processInstanceByKey.setVariable(anyString(), any())).thenReturn(processInstanceByKey);
        when(processInstanceByKey.execute()).thenReturn(mock(ProcessInstance.class));

        // 获取所有异步任务
        List<CompletableFuture<Void>> futures = startProcessService.handleInExkassodatenRequest(InExkassodatenFactory.createValidTestData());
        
        // 等待所有异步任务执行完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).get();
        
        // 验证方法调用次数
        verify(runtimeService, times(3)).createProcessInstanceByKey(anyString());
    }
}

方案二：使用同步线程池，让异步任务在测试主线程执行 #

这种方式不需要等待，直接将异步任务转为同步执行，适合单元测试场景。

生产代码修改

添加自定义线程池支持，方便测试时替换为同步执行器：

import org.camunda.bpm.engine.RuntimeService; 
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class StartProcessService {
    private final RuntimeService runtimeService;
    private final ExecutorService executorService;

    // 生产环境构造方法
    public StartProcessService(RuntimeService runtimeService) {
        this.runtimeService = runtimeService;
        this.executorService = Executors.newFixedThreadPool(3);
    }

    // 测试专用构造方法，注入自定义线程池
    StartProcessService(RuntimeService runtimeService, ExecutorService executorService) {
        this.runtimeService = runtimeService;
        this.executorService = executorService;
    }
   
    public void handleInExkassodatenRequest(MyRequest request) {
        request.getDaten().forEach(entity -> {
            CompletableFuture.supplyAsync(() -> selectRequiredProcess(entity), executorService)
                    .thenAccept(caseId -> {
                        log.info("Case started ");
                    })
                    .exceptionally(err -> {
                        log.error("Failed to start process for entity: {}", entity, err);
                        return null;
                    });
        });
    }

    // 其他方法不变...
}

测试代码修改

使用同步执行器，让异步任务在测试主线程执行：

@ExtendWith(MockitoExtension.class)
@MockitoSettings(strictness = Strictness.LENIENT)
class MyServiceTest {

    private StartProcessService startProcessService;

    @Mock(answer = Answers.RETURNS_DEEP_STUBS)
    private RuntimeService runtimeService;

    // 同步执行器：所有任务直接在调用线程执行
    private final ExecutorService synchronousExecutor = Runnable::run;

    @BeforeEach
    void setUp() {
        // 初始化测试用服务实例，注入同步线程池
        startProcessService = new StartProcessService(runtimeService, synchronousExecutor);
    }

    @Test
    void startProcesses() {
        ProcessInstantiationBuilder processInstanceByKey = mock(ProcessInstantiationBuilderImpl.class);
        when(runtimeService.createProcessInstanceByKey(anyString())).thenReturn(processInstanceByKey);
        when(processInstanceByKey.setVariable(anyString(), any())).thenReturn(processInstanceByKey);
        when(processInstanceByKey.execute()).thenReturn(mock(ProcessInstance.class));

        startProcessService.handleInExkassodatenRequest(InExkassodatenFactory.createValidTestData());
        
        // 任务已同步执行，直接验证
        verify(runtimeService, times(3)).createProcessInstanceByKey(anyString());
    }
}

额外排查建议 #

• 确认InExkassodatenFactory.createValidTestData()返回的数据集确实包含3条数据，避免因数据量不符导致验证失败。
• 查看exceptionally分支的日志，定位具体异常原因（如Mock对象在异步线程中的访问问题、实体数据为空等）。

内容的提问来源于stack exchange，提问作者Roma Kap