ShankaraNarayanan ShankaraNarayanan - 6 months ago 164
Java Question

spring state machine in multi threaded environment

We have just started using spring state machine. Have a couple of questions:


  • Is the state context only once per state machine?

  • Do the events passed to the state machine run in a blocking way? Any way to make them run in parallel, like, providing a new state machine everytime an event is triggered?



Here is my code:

Configuring the states and transitions:

@Override
public void configure(
StateMachineTransitionConfigurer<WorkFlowStates, WorkFlowEvent> transitions)
throws Exception {
transitions
.withExternal()
.source(WorkFlowStates.ready)
.target(WorkFlowStates.l1Creation)
.event(WorkFlowEvent.createWorkItem)
.action(workFlowCreator.createL1())


Providing actions during state transitions:

public Action<WorkFlowStates, WorkFlowEvent> createL3() {
return new Action<WorkFlowStates, WorkFlowEvent>() {

public void execute(StateContext<WorkFlowStates, WorkFlowEvent> context) {
System.out.println("l3 creation in action");
Map<Object, Object> variables = context.getExtendedState().getVariables();
Integer counter = (Integer)variables.get("counter");
if(counter == null) counter = 1;
else counter = counter+1;
variables.put("counter", counter);
System.out.println("Counter is "+counter);
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
variables.put(Level.l3, WorkItemFactory.createFailureL1L2L3WorkItem());
variables.put("level", Level.l3);
System.out.println("l3 created");
}
};
}


Task executor:

public void configure(StateMachineConfigurationConfigurer<WorkFlowStates,
WorkFlowEvent>config)
throws Exception {
config
.withConfiguration()
.autoStartup(true)
.taskExecutor(taskExecutor())
.listener(new WorkFlowStateMachineListener());
}

@Bean(name = StateMachineSystemConstants.TASK_EXECUTOR_BEAN_NAME)
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.afterPropertiesSet();
taskExecutor.setCorePoolSize(5);
return taskExecutor;
}


And events passed to the state machine:

StateMachine<WorkFlowStates, WorkFlowEvent> stateMachine =
context.getBean(StateMachine.class);

stateMachine.sendEvent(WorkFlowEvent.createWorkItem);
stateMachine.sendEvent(WorkFlowEvent.createWorkItem);

Answer

Yes, default behaviour is blocking as underlying TaskExecutor is SyncTaskExecutor. This can be changed via common config as mentioned in docs. Also see tasks recipe where ThreadPoolTaskExecutor is used on default to execute regions parallel.

When moving away from a blocking machine you need to pay attention how machine works and when its ready to process further events as machine may then be in a state where events are discarded. This is usually when you may need to start adding deferred events so that machine can process those in future in more suitable time.