Spring与状态机Reactor测试中阻塞操作适配问题

一、推荐：非阻塞的Reactor原生测试方案 #

完全遵循反应式编程模型，替换原有的阻塞和轮询逻辑，用Reactor的StepVerifier和Flux实现轮询：

核心思路 #

1. 用链式反应式操作替代block()，将所有测试步骤串联成一个Mono/Flux链
2. 用Flux.interval实现定时轮询，配合filter和take(1)终止轮询
3. 用StepVerifier验证整个流程的正确性

代码示例 #

假设你有以下方法：

• startStateMachine(): Mono<ResponseDto>：启动状态机返回带唯一ID的响应
• checkFirstTableUpdated(String id): Mono<Boolean>：检查第一表是否更新
• checkSecondTableUpdated(String id): Mono<Boolean>：检查第二表是否更新

完整测试代码

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
import java.time.Duration;
import java.util.function.Predicate;

@Test
void testStateMachineFlow() {
    startStateMachine()
        .flatMap(response -> {
            String uniqueId = response.getUniqueId();
            
            // 2. 初始验证数据库值，未更新则进入轮询
            return pollUntil(
                    checkFirstTableUpdated(uniqueId),
                    updated -> updated,
                    Duration.ofSeconds(1),
                    Duration.ofSeconds(10)
                )
                // 4. 检查另一表更新
                .then(checkSecondTableUpdated(uniqueId))
                .filter(updated -> updated);
        })
        .as(StepVerifier::create)
        .expectNext(true)
        .verifyComplete();
}

// 抽通用轮询工具方法，复用性更强
private <T> Mono<T> pollUntil(Mono<T> checkMono, Predicate<T> condition, Duration interval, Duration timeout) {
    return checkMono
        .filter(condition)
        .switchIfEmpty(
            Flux.interval(interval)
                .flatMap(tick -> checkMono)
                .filter(condition)
                .take(1)
                .timeout(timeout)
                .single()
        );
}

代码说明

• 整个流程无阻塞操作，完全符合反应式模型
• pollUntil方法封装了轮询逻辑：先执行一次检查，若不符合条件则每隔指定时间重试，直到满足条件或超时
• StepVerifier用于验证整个反应式链是否正常完成，确保所有数据库检查都通过

二、临时方案：允许阻塞操作（不推荐，仅过渡用） #

如果需要快速迁移测试，可关闭Reactor的BlockHound检测，继续使用block()和原有的Awaitility轮询逻辑：

方法1：在测试类上添加配置 #

import org.springframework.test.context.TestPropertySource;
import org.junit.jupiter.api.Test;

@TestPropertySource(properties = "reactor.blockhound.enabled=false")
public class StateMachineIntegrationTest {
    @Test
    void testStateMachineFlow() {
        // 恢复原有block()逻辑
        ResponseDto response = startStateMachine().block(Duration.ofSeconds(5));
        assert response != null;
        String uniqueId = response.getUniqueId();

        // 初始验证
        assert checkFirstTableUpdated(uniqueId).block();

        // 原Awaitility轮询逻辑
        Awaitility.await().atMost(Duration.ofSeconds(10)).pollInterval(Duration.ofSeconds(1))
            .until(() -> checkFirstTableUpdated(uniqueId).block());

        // 检查另一表
        assert checkSecondTableUpdated(uniqueId).block();
    }
}

方法2：在测试配置文件中全局关闭 #

在src/test/resources/application-test.properties中添加：

reactor.blockhound.enabled=false

注意事项 #

• 该方案仅适合临时过渡，长期使用会破坏反应式编程的非阻塞特性，可能引发线程池阻塞等问题
• 建议尽快迁移到非阻塞的测试方案

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