Spring Boot Webflux中Reactor Netty如何保障请求线性一致性

原有代码失效的根本原因 #

你之前尝试用subscribeOn(Schedulers.single())实现顺序执行不生效，核心问题有三个：

• Schedulers.single()是Reactor内置的全局共享单线程调度器，框架内部、其他依赖的响应式组件都可能往这个调度器提交任务，会出现业务任务被插队的情况
• 你在doOnNext回调里对每个请求的业务流做独立subscribe()，不同请求的流是完全独立的，加上Mongo响应式驱动自身会用独立线程池做IO操作，不同请求的查询、写入回调返回顺序不受你指定的调度器约束，自然会出现请求C比请求B先写入的乱序问题
• Reactor Netty本身的NIO线程组是多线程设计，不同请求会被分配到不同的EventLoop线程处理，网络层本身不提供全局请求顺序保证的能力

可行实现方案 #

要实现「收到请求立刻返回ACCEPTED状态、后台严格按请求接收顺序串行执行读写逻辑」，不需要修改Reactor Netty底层配置，直接基于Reactor Core自带的单线程调度器即可实现，该调度器内部默认搭载无界FIFO任务队列，所有提交的任务严格按提交顺序执行。
核心实现原则：

• 单独创建一个业务独占的全局单线程调度器，不要和框架其他任务共用
• 所有请求的全链路业务逻辑（查询、计算、写入）都绑定到这个独占调度器上，保证前一个请求的全流程执行完成后，才会启动下一个请求的处理
• 任务提交到调度队列成功后，立刻给客户端返回响应，不需要等待业务逻辑执行完成

代码改造示例 #

1. 定义全局独占的串行调度器，全局只初始化一次：

// 业务独享的单线程调度器，内部为无界FIFO队列，任务严格按提交顺序执行
private final Scheduler strictOrderScheduler = Schedulers.newSingle("biz-strict-order-worker", true);

2. 改造控制器逻辑，移除doOnNext里的独立订阅逻辑，保证所有业务任务都提交到同一个串行队列：

@RestController
@RequestMapping(value = "/api/event")
@RequiredArgsConstructor
public class EventHandler {
    private final EventRepository repo;
    private final Scheduler strictOrderScheduler = Schedulers.newSingle("biz-strict-order-worker", true);

    @PostMapping
    @ResponseStatus(HttpStatus.ACCEPTED)
    public Mono<String> create(Event event) {
        String newEventId = event.withNewId().getId();
        // 包装当前请求的全量业务逻辑
        Mono.just(someQuery)
                .flatMap(query -> repo.find(query)
                        .map(existEvent -> {
                            event.setData(event.getData() + existEvent.getData());
                            return event;
                        })
                )
                .switchIfEmpty(Mono.just(event))
                .flatMap(repo::save)
                // 绑定全链路到独占串行调度器，保证所有操作（含IO回调）都在该线程排队执行
                .publishOn(strictOrderScheduler)
                .subscribeOn(strictOrderScheduler)
                // 提交任务到队列，异步执行
                .subscribe();
        // 入队成功立刻返回结果
        return Mono.just(newEventId);
    }
}

注意事项 #

• 上述方案是单实例进程内的顺序保证，如果服务是多实例部署，需要在前置网关层按业务分片键做请求路由，或者引入顺序消息队列、分布式锁机制实现全局顺序保证
• 如果担心无界队列导致OOM，可以在创建调度器时自定义队列大小，配置任务拒绝策略做流量降级
• 不要在业务逻辑里手动切换其他调度器（比如自定义publishOn到其他线程池），否则会打破串行约束，导致乱序
• 该方案下的线性一致性是严格可保障的：前一个请求的repo.save()操作没有完成触发回调，后一个请求的repo.find()逻辑根本不会被调度执行，完全符合「后续写操作必须依赖前序写操作完成才可执行」的要求。

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