java - 多线程环境中的Spring状态机

我有一个n个消息监听器的事件队列。消息到达时，一个消息监听器将其接收并执行状态机的新实例。
我面临的问题是，尽管多个消息是并行处理的，但是状态机会按顺序开始执行操作，尽管它们是由不同的状态机实例调用的，如您在此处看到的:

2017-10-18 16:11:03.740  INFO 30282 --- [lTaskExecutor-1] o.s.s.support.LifecycleObjectSupport     : started org.springframework.statemachine.support.DefaultStateMachineExecutor@6ddb1ea6
2017-10-18 16:11:03.741  INFO 30282 --- [lTaskExecutor-1] o.s.s.support.LifecycleObjectSupport     : started EVALUATE_IS_WALKTHROUGH SAVE END START  / START / uuid=b922b6b1-a441-4924-8531-d45e0e0c9c40 / id=null
2017-10-18 16:11:03.740  INFO 30282 --- [TaskExecutor-10] o.s.s.support.LifecycleObjectSupport     : started org.springframework.statemachine.support.DefaultStateMachineExecutor@13b6ace4
2017-10-18 16:11:03.741  INFO 30282 --- [TaskExecutor-10] o.s.s.support.LifecycleObjectSupport     : started EVALUATE_IS_WALKTHROUGH SAVE END START  / START / uuid=e06a8c1d-beed-41c6-bc63-d8c1a3a56169 / id=null
2017-10-18 16:11:03.759  INFO 30282 --- [pool-5-thread-1] i.b.b.e.processors.actions.SaveAction    : [io.botbit.backend.events.processors.actions.SaveAction@607e4071] Saving event Event[id=null,
2017-10-18 16:11:24.046  INFO 30282 --- [pool-5-thread-1] i.b.b.e.processors.actions.SaveAction    : [io.botbit.backend.events.processors.actions.SaveAction@607e4071] Saving event Event[id=null,
2017-10-18 16:11:44.058  INFO 30282 --- [pool-5-thread-1] i.b.b.e.p.a.EvaluateIsWalkthroughAction  : Evaluation is WT,,,
2017-10-18 16:11:44.059  INFO 30282 --- [pool-5-thread-1] o.s.s.support.LifecycleObjectSupport     : stopped org.springframework.statemachine.support.DefaultStateMachineExecutor@6ddb1ea6
2017-10-18 16:11:44.060  INFO 30282 --- [pool-5-thread-1] o.s.s.support.LifecycleObjectSupport     : stopped EVALUATE_IS_WALKTHROUGH SAVE END START  /  / uuid=b922b6b1-a441-4924-8531-d45e0e0c9c40 / id=null
2017-10-18 16:11:44.060  INFO 30282 --- [pool-5-thread-1] i.b.b.e.p.a.EvaluateIsWalkthroughAction  : Evaluation is WT,,,
2017-10-18 16:11:44.061  INFO 30282 --- [pool-5-thread-1] o.s.s.support.LifecycleObjectSupport     : stopped org.springframework.statemachine.support.DefaultStateMachineExecutor@13b6ace4
2017-10-18 16:11:44.061  INFO 30282 --- [pool-5-thread-1] o.s.s.support.LifecycleObjectSupport     : stopped EVALUATE_IS_WALKTHROUGH SAVE END START  /  / uuid=e06a8c1d-beed-41c6-bc63-d8c1a3a56169 / id=nul

我认为发生这种情况是因为每个 Action 都在同一线程([pool-5-thread-1])中运行。
我需要每个实例都完全并行运行，这意味着状态机是并行执行的，而且它的 Action 也是并行执行的。

任何帮助将不胜感激，谢谢!

@Component
public class EventConsumer {
    private final static Logger logger = Logger.getLogger(EventConsumer.class);
    @Autowired
    private StateMachineFactory<String, String> eventProcessorFactory;

    public void consume(Event event) {
        logger.info("Received <" + event + ">");
        StateMachine<String, String> eventProcessor = eventProcessorFactory.getStateMachine();
        eventProcessor.getExtendedState().getVariables().put("event", event);
        eventProcessor.start();
        eventProcessor.sendEvent(CommonEvents.SUCCESS);
    }

    public void consume(Collection<Event> events) {
        for (Event event : events) {
            this.consume(event);
        }
    }
}

这是状态机配置

@Configuration
@EnableStateMachineFactory
public class WiFiConnectEventProcessorConfig extends StateMachineConfigurerAdapter<String, String> {

    @Autowired
    SaveAction saveAction;
    @Autowired
    DeprecateAddVisitationAction addVisitation;
    @Autowired
    EvaluateIsWalkthroughAction isWTAction;

    @Override
    public void configure(StateMachineStateConfigurer<String, String> states) throws Exception {
        states.withStates().initial(WiFiConnectStates.START).state(WiFiConnectStates.SAVE, saveAction)
                .state(WiFiConnectStates.DEPRECATE_ADD_VISITATION, addVisitation)
                .state(WiFiConnectStates.EVALUATE_IS_WALKTHROUGH, isWTAction).end(WiFiConnectStates.END);
    }

    @Override
    public void configure(StateMachineTransitionConfigurer<String, String> transitions) throws Exception {
        transitions.withExternal().source(WiFiConnectStates.START).target(WiFiConnectStates.SAVE)
                .event(CommonEvents.SUCCESS)
                .and().withExternal().source(WiFiConnectStates.SAVE)
                .target(WiFiConnectStates.DEPRECATE_ADD_VISITATION).event(CommonEvents.SUCCESS)
                .and().withExternal()
                .source(WiFiConnectStates.DEPRECATE_ADD_VISITATION).target(WiFiConnectStates.EVALUATE_IS_WALKTHROUGH)
                .event(CommonEvents.SUCCESS)
                .and().withExternal().source(WiFiConnectStates.EVALUATE_IS_WALKTHROUGH)
                .target(WiFiConnectStates.END).event(CommonEvents.SUCCESS);
    }
}

最佳答案

您需要创建自己的TaskScheduler，并在StateMachineConfig @Configuration文件中进行如下配置。选择适合您需要的PoolSize。

    @Override
    public void configure(StateMachineConfigurationConfigurer<TaskState, TaskEvent> config) throws Exception {
        config.withConfiguration()
                .taskScheduler(myTaskScheduler());
    }

    @Bean
    public TaskScheduler myTaskScheduler() {
        final ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(10);
        return scheduler;
    }