响应式编程：混合阻塞MySQL查询与异步WebClient请求的最优方案

这确实是Reactor开发中常见的痛点——MySQL的JDBC驱动是阻塞式的，但我们又想和WebClient的非阻塞HTTP请求并行执行，把整体响应延迟降到最低。核心思路是把阻塞的MySQL操作隔离到专用线程池，同时让WebClient请求在Reactor的异步线程中运行，最后通过Reactor的组合操作让两者并行执行，总耗时就取两者中较长的那个，而不是相加。

下面一步步拆解具体实现：

1. 将阻塞MySQL查询包装为非阻塞Mono #

JDBC操作是阻塞的，如果直接在Reactor的IO线程（比如Netty的EventLoop）上执行，会阻塞整个线程池，拖垮系统性能。所以我们需要用Mono.fromCallable()把阻塞逻辑包装起来，再通过subscribeOn()指定一个专门处理阻塞操作的线程池。

Reactor提供了Schedulers.boundedElastic()，这是专门为阻塞操作设计的线程池，会根据负载动态创建线程，同时避免资源耗尽。如果你的并发量很高，也可以自定义线程池来更精准地控制资源：

// 自定义MySQL专用线程池，适配你的MySQL连接池大小
private static final Scheduler MYSQL_SCHEDULER = Schedulers.newBoundedElastic(
    10,  // 核心线程数，建议和MySQL连接池核心数一致
    100, // 最大线程数
    "mysql-executor" // 线程池名称，方便排查问题
);

// 包装JDBC查询为Mono
private Mono<User> fetchUserFromDb(Long userId) {
    return Mono.fromCallable(() -> 
        // 这里是你的阻塞JDBC查询逻辑，比如用JdbcTemplate
        jdbcTemplate.queryForObject(
            "SELECT id, name, email FROM users WHERE id = ?",
            new Object[]{userId},
            (rs, rowNum) -> new User(rs.getLong("id"), rs.getString("name"), rs.getString("email"))
        )
    ).subscribeOn(MYSQL_SCHEDULER); // 指定线程池
}

2. 并行执行MySQL查询与WebClient请求 #

Reactor的Mono.zip()方法可以让多个Mono并行执行，等待所有操作完成后合并结果。WebClient的请求本身就是非阻塞的，会在Netty的EventLoop线程上异步执行，和MySQL的阻塞操作（在专用线程池）互不干扰，真正实现并行。

示例代码：

// WebClient非阻塞HTTP请求
private Mono<ExternalUserData> fetchExternalData(Long userId) {
    return webClient.get()
            .uri("/api/user-preferences/{userId}", userId)
            .retrieve()
            .bodyToMono(ExternalUserData.class)
            .onErrorResume(e -> Mono.just(new ExternalUserData())); // 优雅处理API请求失败
}

// 并行执行并合并结果
public Mono<CombinedUserResult> getCombinedUserInfo(Long userId) {
    // zip会同时触发两个Mono，等待两者都完成后返回Tuple
    return Mono.zip(fetchUserFromDb(userId), fetchExternalData(userId))
            .map(tuple -> new CombinedUserResult(tuple.getT1(), tuple.getT2()));
}

这里的关键是：fetchUserFromDb在mysql-executor线程池执行阻塞查询，fetchExternalData在Netty EventLoop线程执行非阻塞HTTP请求，两者同时启动，总耗时等于两个操作中耗时较长的那个，完美降低整体延迟。

3. 关键注意事项 #

• 绝对不要在EventLoop线程执行阻塞操作：EventLoop线程是Reactor处理异步IO的核心线程，数量有限，阻塞一个就会影响其他请求的处理。
• 线程池与MySQL连接池匹配：自定义MySQL线程池的最大线程数不要超过MySQL连接池的最大连接数，否则会出现线程等待连接的情况，反而增加延迟。
• 错误处理要到位：并行操作中任何一个失败都会导致整个zip失败，所以要用onErrorResume、onErrorReturn等操作符处理单个操作的异常，保证整体流程的健壮性。

完整示例代码 #

@Service
public class UserDataService {
    private final WebClient webClient;
    private final JdbcTemplate jdbcTemplate;
    private static final Scheduler MYSQL_SCHEDULER = Schedulers.newBoundedElastic(10, 100, "mysql-executor");

    public UserDataService(WebClient.Builder webClientBuilder, JdbcTemplate jdbcTemplate) {
        this.webClient = webClientBuilder.baseUrl("https://external-service.com").build();
        this.jdbcTemplate = jdbcTemplate;
    }

    private Mono<User> fetchUserFromDb(Long userId) {
        return Mono.fromCallable(() -> 
            jdbcTemplate.queryForObject(
                "SELECT id, name, email FROM users WHERE id = ?",
                new Object[]{userId},
                (rs, rowNum) -> new User(rs.getLong("id"), rs.getString("name"), rs.getString("email"))
            )
        ).subscribeOn(MYSQL_SCHEDULER);
    }

    private Mono<ExternalUserData> fetchExternalData(Long userId) {
        return webClient.get()
                .uri("/user-preferences/{userId}", userId)
                .retrieve()
                .bodyToMono(ExternalUserData.class)
                .onErrorResume(e -> Mono.just(new ExternalUserData()));
    }

    public Mono<CombinedUserResult> getCombinedUserInfo(Long userId) {
        return Mono.zip(fetchUserFromDb(userId), fetchExternalData(userId))
                .map(tuple -> new CombinedUserResult(tuple.getT1(), tuple.getT2()));
    }

    // 辅助记录类
    private record User(Long id, String name, String email) {}
    private record ExternalUserData(String favoriteTheme, Integer notificationFrequency) {}
    private record CombinedUserResult(User user, ExternalUserData externalData) {}
}

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