每次用到的时候新创建一个状态机,太奢侈了,官方文档里面也提到过这点。
而且创建出来的实例,其状态也跟当前订单的不符;spring statemachine暂时不支持每次创建时指定当前状态,所以对状态机引擎实例的持久化,就成了必须要考虑的问题。(不过在后续版本有直接指定状态的方式,这个后面会写)
这里扩展说明一下,状态机引擎的持久化一直是比较容易引起讨论的,因为很多场景并不希望再多存储一些中间非业务数据,之前在淘宝工作时,淘宝的订单系统tradeplatform自己实现了一套workflowEngine,其实说白了也就是一套状态机引擎,所有的配置都放在xml中,每次每个环节的请求过来,都会重新创建一个状态机引擎实例,并根据当前的订单状态来设置引擎实例的状态。
workflowEngine没有做持久化,私下里猜测下这样实现的原因:1、淘系数据量太大,一天几千万笔订单,额外的信息存储就要耗费很多存储资源;2、完全自主开发的状态机引擎,可定制化比较强,根据自己的业务需要可以按自己的需要处理。
而反过来,spring statemachine并不支持随意指定初始状态,每次创建都是固定的初始化状态,其实也只是有好处的,标准版流程,而且可以保证安全,每个节点都是按照事先定义好的流程跑下来,而不是随意指定。所以,状态机引擎实例的持久化,我们这次的主题,那就继续聊下去吧。
spring statemachine 本身支持了内存、redis及db的持久化,内存持久化就不说了,看源码实现就是放在了hashmap里,平时也没谁项目中可以这么奢侈,啥啥都放在内存中,而且一旦重启…..:sweat:。下面详细说下利用redis进行的持久化操作。
spring statemachine 本身是提供了一个redis存储的组件的,在1.2.10.RELEASE版本中,这个组件需要通过依赖引入,同时需要引入的还有序列化的组件kyro、data-common:
gradle引入依赖 (build.gradle 或者 libraries.gradle,由自己项目的gradle组织方式来定):
compile 'org.springframework.statemachine:spring-statemachine-core:1.2.10.RELEASE'
compile 'org.springframework.statemachine:spring-statemachine-data-common:1.2.10.RELEASE'
compile 'org.springframework.statemachine:spring-statemachine-kyro:1.2.10.RELEASE'
compile 'org.springframework.statemachine:spring-statemachine-redis:1.2.10.RELEASE'复制代码
当然如果是maven的话,一样的,pom.xml如下:
<dependencies>
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-core</artifactId>
<version>1.2.10.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-data-common</artifactId>
<version>1.2.10.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-kyro</artifactId>
<version>1.2.10.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-redis</artifactId>
<version>1.2.10.RELEASE</version>
</dependency>
</dependencies>复制代码
说明:
spring statemachine持久化时,采用了三层结构设计,persister —>persist —>repository。
import org.springframework.beans.factory.annotation.Autowire;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.statemachine.StateMachinePersist;
import org.springframework.statemachine.persist.StateMachinePersister;
import org.springframework.statemachine.redis.RedisStateMachinePersister;
@Configuration
public class BizOrderRedisStateMachinePersisterConfig {
@Autowired
private StateMachinePersist bizOrderRedisStateMachineContextPersist;
@Bean(name = "bizOrderRedisStateMachinePersister",autowire = Autowire.BY_TYPE)
public StateMachinePersister<BizOrderStatusEnum, BizOrderStatusChangeEventEnum,String> bizOrderRedisStateMachinePersister() {
return new RedisStateMachinePersister<>(bizOrderRedisStateMachineContextPersist);
}
}复制代码
这里采用官方samples中初始化的方式,通过@Bean注解来创建一个RedisStateMachinePersister实例,注意其中传递进去的Persist为自定义的bizOrderRedisStateMachineContextPersist
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.MessageHeaders;
import org.springframework.statemachine.StateMachineContext;
import org.springframework.statemachine.StateMachinePersist;
import org.springframework.statemachine.kryo.MessageHeadersSerializer;
import org.springframework.statemachine.kryo.StateMachineContextSerializer;
import org.springframework.statemachine.kryo.UUIDSerializer;
import org.springframework.statemachine.redis.RedisStateMachineContextRepository;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.UnsupportedEncodingException;
import java.util.Base64;
import java.util.UUID;
@Component("bizOrderRedisStateMachineContextPersist")
public class BizOrderRedisStateMachineContextPersist implements StateMachinePersist<BizOrderStatusEnum, BizOrderStatusChangeEventEnum, String> {
@Autowired
@Qualifier("redisStateMachineContextRepository")
private RedisStateMachineContextRepository<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> redisStateMachineContextRepository;
@Autowired
private BizOrderStateMachineContextRepository bizOrderStateMachineContextRepository;
// 加入存储到DB的数据repository, biz_order_state_machine_context表结构:
// bizOrderId
// contextStr
// curStatus
// updateTime
/**
* Write a {@link StateMachineContext} into a persistent store
* with a context object {@code T}.
*
* @param context the context
* @param contextObj the context ojb
* @throws Exception the exception
*/
@Override
@Transactional
public void write(StateMachineContext<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> context, String contextObj) throws Exception {
redisStateMachineContextRepository.save(context, contextObj);
// save to db
BizOrderStateMachineContext queryResult = bizOrderStateMachineContextRepository.selectByOrderId(contextObj);
if (null == queryResult) {
BizOrderStateMachineContext bosmContext = new BizOrderStateMachineContext(contextObj,
context.getState().getStatus(), serialize(context));
bizOrderStateMachineContextRepository.insertSelective(bosmContext);
} else {
queryResult.setCurOrderStatus(context.getState().getStatus());
queryResult.setContext(serialize(context));
bizOrderStateMachineContextRepository.updateByPrimaryKeySelective(queryResult);
}
}
/**
* Read a {@link StateMachineContext} from a persistent store
* with a context object {@code T}.
*
* @param contextObj the context ojb
* @return the state machine context
* @throws Exception the exception
*/
@Override
public StateMachineContext<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> read(String contextObj) throws Exception {
StateMachineContext<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> context = redisStateMachineContextRepository.getContext(contextObj);
//redis 访缓存击穿
if (null != context && BizOrderConstants.STATE_MACHINE_CONTEXT_ISNULL.equalsIgnoreCase(context.getId())) {
return null;
}
//redis 为空走db
if (null == context) {
BizOrderStateMachineContext boSMContext = bizOrderStateMachineContextRepository.selectByOrderId(contextObj);
if (null != boSMContext) {
context = deserialize(boSMContext.getContext());
redisStateMachineContextRepository.save(context, contextObj);
} else {
context = new StateMachineContextIsNull();
redisStateMachineContextRepository.save(context, contextObj);
}
}
return context;
}
private String serialize(StateMachineContext<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> context) throws UnsupportedEncodingException {
Kryo kryo = kryoThreadLocal.get();
ByteArrayOutputStream out = new ByteArrayOutputStream();
Output output = new Output(out);
kryo.writeObject(output, context);
output.close();
return Base64.getEncoder().encodeToString(out.toByteArray());
}
@SuppressWarnings("unchecked")
private StateMachineContext<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> deserialize(String data) throws UnsupportedEncodingException {
if (StringUtils.isEmpty(data)) {
return null;
}
Kryo kryo = kryoThreadLocal.get();
ByteArrayInputStream in = new ByteArrayInputStream(Base64.getDecoder().decode(data));
Input input = new Input(in);
return kryo.readObject(input, StateMachineContext.class);
}
private static final ThreadLocal<Kryo> kryoThreadLocal = new ThreadLocal<Kryo>() {
@SuppressWarnings("rawtypes")
@Override
protected Kryo initialValue() {
Kryo kryo = new Kryo();
kryo.addDefaultSerializer(StateMachineContext.class, new StateMachineContextSerializer());
kryo.addDefaultSerializer(MessageHeaders.class, new MessageHeadersSerializer());
kryo.addDefaultSerializer(UUID.class, new UUIDSerializer());
return kryo;
}
};
}复制代码
说明:
这里需要额外创建5张表,分别存储ActionGuardStateStateMachineTransition,比较复杂。
有两个repository,一个是spring statemachine提供的redisRepo,另一个则是项目中基于mybatis的repo,先是db-repo:
import org.apache.ibatis.annotations.Param;
import org.springframework.data.domain.Pageable;
import org.springframework.stereotype.Repository;
import java.util.List;
@Repository
public interface BizOrderStateMachineContextRepository {
int deleteByPrimaryKey(Long id);
BizOrderStateMachineContext selectByOrderId(String bizOrderId);
int updateByPrimaryKey(BizOrderStateMachineContext BizOrderStateMachineContext);
int updateByPrimaryKeySelective(BizOrderStateMachineContext BizOrderStateMachineContext);
int insertSelective(BizOrderStateMachineContext BizOrderStateMachineContext);
int selectCount(BizOrderStateMachineContext BizOrderStateMachineContext);
List<BizOrderStateMachineContext> selectPage(@Param("BizOrderStateMachineContext") BizOrderStateMachineContext BizOrderStateMachineContext, @Param("pageable") Pageable pageable);
}复制代码
然后是redisRepo
import org.springframework.beans.factory.annotation.Autowire;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.statemachine.redis.RedisStateMachineContextRepository;
@Configuration
public class BizOrderRedisStateMachineRepositoryConfig {
/**
* 接入asgard后,redis的connectionFactory可以通过serviceName + InnerConnectionFactory来注入
*/
@Autowired
private RedisConnectionFactory finOrderRedisInnerConnectionFactory;
@Bean(name = "redisStateMachineContextRepository", autowire = Autowire.BY_TYPE)
public RedisStateMachineContextRepository<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> redisStateMachineContextRepository() {
return new RedisStateMachineContextRepository<>(finOrderRedisInnerConnectionFactory);
}
}
复制代码
@Autowired
@Qualifier("bizOrderRedisStateMachinePersister")
private StateMachinePersister<BizOrderStatusEnum,BizOrderStatusChangeEventEnum,String> bizOrderRedisStateMachinePersister;
......
bizOrderRedisStateMachinePersister.persist(stateMachine, request.getBizCode());
......
StateMachine<BizOrderStatusEnum, BizOrderStatusChangeEventEnum> stateMachine
= bizOrderRedisStateMachinePersister.restore(srcStateMachine,statusRequest.getBizCode());
......复制代码
支持,关于spring statemachine的持久化就交代完了,下面就是最关键的,怎么利用状态机来串联业务,下一节将会详细描述。