多线程环境中的弹簧状态机

Question

我有一个包含 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

我认为发生这种情况是因为每个操作都在同一个线程（[pool-5-thread-1]）中运行。我需要每个实例完全并行运行，我的意思是状态机是并行执行的，但它的操作也是如此。

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

@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);
    }
}

Answer 1

您需要创建自己的 TaskScheduler，并在 StateMachineConfig @Configuration 文件中按如下方式对其进行配置。根据您的需要选择合适的池大小。

    @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;
    }

Answer 2

如评论中所述，您的状态机将共享相同的资源，包括 TaskExecutor 和 TaskScheduler。

的默认行为是在调度程序中使用单线程执行器，这就是为什么您在应用程序中看到瓶颈的原因。

https://github.com/spring-projects/spring-framework/blob/master/spring-context/src/main/java/org/springframework/scheduling/concurrent/ConcurrentTaskExecutor.java#L90

这解释了您可以使用的各种类型的执行器和调度程序：https://docs.spring.io/spring/docs/4.2.x/spring-framework-reference/html/scheduling.html

然后，您可以升级您的配置：

  @Override
  public void configure(StateMachineConfigurationConfigurer<RfqState, RfqEvent> config)
      throws Exception {
    SyncTaskExecutor executor = new SyncTaskExecutor();
    config.withConfiguration()
        .taskExecutor( my executor )
        .taskScheduler( my scheduler );
  }