如何在Spring状态机中使用Spring Reactor(Flux)及从订阅者返回结果

嘿，这个问题问得相当实用！Spring Reactor的异步流能力搭配Spring State Machine的状态管理，确实能让复杂异步流程的管控变得清晰很多。我来一步步给你拆解怎么实现，以及怎么解决动作里返回结果给状态机的问题。

Spring State Machine本身默认是同步的，但我们可以很容易地把Reactor的异步逻辑融入进去，核心有两个方向：

• 用Reactor的事件流来驱动状态机的状态转换
• 在状态机的动作（Action）中执行Reactor异步操作

比如，你可以把一个Flux<Events>作为状态机的事件源，让流里的事件自动触发状态转换：

// 假设已经拿到了状态机实例stateMachine
Flux<Events> eventStream = Flux.just(Events.START)
    .delayElements(Duration.ofSeconds(2)); // 模拟异步产生的事件

// 订阅事件流，自动发送给状态机
eventStream.subscribe(event -> stateMachine.sendEvent(event));

这里的关键是：状态机的动作默认是同步执行的，但我们可以在动作内部启动Reactor异步任务，然后通过更新状态机上下文或者发送事件的方式把结果反馈给状态机。

方案1：用ExtendedState存储结果 + 发送事件触发状态转换 #

状态机的ExtendedState是一个可以存储任意数据的上下文，适合存放异步操作的结果。当Reactor异步任务完成后，我们可以把结果存入这个上下文，再发送一个事件让状态机切换到下一个状态。

举个具体的代码例子：

首先定义状态和事件：

enum States {
    INIT, PROCESSING, COMPLETED, FAILED
}

enum Events {
    START, PROCESS_SUCCESS, PROCESS_ERROR
}

然后写一个包含Reactor逻辑的动作类：

@Component
public class AsyncProcessAction implements Action<States, Events> {

    // 模拟一个返回Flux的Reactive服务
    private final SomeReactiveService reactiveService = new SomeReactiveService();

    @Override
    public void execute(StateContext<States, Events> context) {
        // 执行Reactor异步逻辑
        reactiveService.processData()
            .subscribe(
                // 异步成功时：存结果到上下文，发送成功事件
                result -> {
                    context.getExtendedState().getVariables().put("processResult", result);
                    context.getStateMachine().sendEvent(Events.PROCESS_SUCCESS);
                },
                // 异步失败时：存错误信息，发送失败事件
                error -> {
                    context.getExtendedState().getVariables().put("errorMsg", error.getMessage());
                    context.getStateMachine().sendEvent(Events.PROCESS_ERROR);
                }
            );
    }

    static class SomeReactiveService {
        public Flux<String> processData() {
            // 模拟耗时的异步操作
            return Flux.just("processed_success_data")
                .delayElements(Duration.ofSeconds(3));
        }
    }
}

接下来配置状态机，把这个动作绑定到状态转换上：

@Configuration
@EnableStateMachine
public class StateMachineConfig extends StateMachineConfigurerAdapter<States, Events> {

    @Autowired
    private AsyncProcessAction asyncProcessAction;

    @Override
    public void configure(StateMachineStateConfigurer<States, Events> states) throws Exception {
        states
            .withStates()
                .initial(States.INIT)
                .state(States.PROCESSING)
                .end(States.COMPLETED)
                .end(States.FAILED);
    }

    @Override
    public void configure(StateMachineTransitionConfigurer<States, Events> transitions) throws Exception {
        transitions
            // 从INIT到PROCESSING，触发START事件时执行异步动作
            .withExternal()
                .source(States.INIT)
                .target(States.PROCESSING)
                .event(Events.START)
                .action(asyncProcessAction)
            .and()
            // 异步成功后转到COMPLETED
            .withExternal()
                .source(States.PROCESSING)
                .target(States.COMPLETED)
                .event(Events.PROCESS_SUCCESS)
            .and()
            // 异步失败后转到FAILED
            .withExternal()
                .source(States.PROCESSING)
                .target(States.FAILED)
                .event(Events.PROCESS_ERROR);
    }

    // 配置异步任务执行器，避免主线程阻塞
    @Override
    public void configure(StateMachineConfigurationConfigurer<States, Events> config) throws Exception {
        config
            .withConfiguration()
                .autoStartup(false)
                .taskExecutor(Executors.newCachedThreadPool());
    }
}

方案2：用Mono/Flux包装动作逻辑（更优雅的Reactor风格） #

如果你想让动作完全遵循Reactor的链式写法，可以用Mono来包装异步逻辑，然后在订阅时处理状态机的更新：

@Component
public class ReactiveStyleAction implements Action<States, Events> {

    private final SomeReactiveService reactiveService = new SomeReactiveService();

    @Override
    public void execute(StateContext<States, Events> context) {
        reactiveService.processData()
            .doOnNext(result -> context.getExtendedState().getVariables().put("processResult", result))
            .then(Mono.fromRunnable(() -> context.getStateMachine().sendEvent(Events.PROCESS_SUCCESS)))
            .onErrorResume(error -> {
                context.getExtendedState().getVariables().put("errorMsg", error.getMessage());
                return Mono.fromRunnable(() -> context.getStateMachine().sendEvent(Events.PROCESS_ERROR));
            })
            .subscribeOn(Schedulers.boundedElastic())
            .subscribe();
    }
}

这种写法更符合Reactor的编程习惯，而且能更好地处理错误场景。

• 不要阻塞异步流：绝对不要在动作里调用block()来等待Reactor结果，那样会完全失去异步的意义，还可能导致线程阻塞问题。
• 利用ExtendedState传递数据：状态机的各个状态和动作之间共享数据，优先用ExtendedState，它是线程安全的。
• 异步任务执行器：一定要给状态机配置异步任务执行器，避免动作里的异步逻辑阻塞状态机的主线程。

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