Spring StateMachine 状态机引擎在项目中的应用(二)--持久化

背景

每次用到的时候新创建一个状态机,太奢侈了,官方文档里面也提到过这点。

而且创建出来的实例,其状态也跟当前订单的不符;spring statemachine暂时不支持每次创建时指定当前状态,所以对状态机引擎实例的持久化,就成了必须要考虑的问题。(不过在后续版本有直接指定状态的方式,这个后面会写)

扩展一下

这里扩展说明一下,状态机引擎的持久化一直是比较容易引起讨论的,因为很多场景并不希望再多存储一些中间非业务数据,之前在淘宝工作时,淘宝的订单系统tradeplatform自己实现了一套workflowEngine,其实说白了也就是一套状态机引擎,所有的配置都放在xml中,每次每个环节的请求过来,都会重新创建一个状态机引擎实例,并根据当前的订单状态来设置引擎实例的状态。

workflowEngine没有做持久化,私下里猜测下这样实现的原因:
1、淘系数据量太大,一天几千万笔订单,额外的信息存储就要耗费很多存储资源;
2、完全自主开发的状态机引擎,可定制化比较强,根据自己的业务需要可以按自己的需要处理。

而反过来,spring statemachine并不支持随意指定初始状态,每次创建都是固定的初始化状态,其实也只是有好处的,标准版流程,而且可以保证安全,每个节点都是按照事先定义好的流程跑下来,而不是随意指定。所以,状态机引擎实例的持久化,我们这次的主题,那就继续聊下去吧。

持久化

spring statemachine 本身支持了内存、redis及db的持久化,内存持久化就不说了,看源码实现就是放在了hashmap里,平时也没谁项目中可以这么奢侈,啥啥都放在内存中,而且一旦重启…..😓。下面详细说下利用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-持久化.png

说明:

spring statemachine持久化时,采用了三层结构设计,persister —>persist —>repository。

  • 其中persister中封装了write和restore两个方法,分别用于持久化写及反序列化读出。
  • persist只是一层皮,主要还是调用repository中的实际实现;但是在这里,由于redis存储不保证百分百数据安全,所以我实现了一个自定义的persist,其中封装了数据写入db、从db中读取的逻辑。
  • repository中做了两件事儿
    • 序列化/反序列化数据,将引擎实例与二进制数组互相转换
    • 读、写redis
详细的实现
Persister
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

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

说明:

  1. 如果只是持久化到redis中,那么BizOrderStateMachineContextRepository相关的所有内容均可删除。不过由于redis无法承诺百分百的数据安全,所以我这里做了两层持久化,redis+db

  2. 存入redis中的数据默认采用kryo来序列化及反序列化,RedisStateMachineContextRepository中实现了对应代码。但是spring statemachine默认的db存储比较复杂,需要创建多张表,参加下图:

    jpa-table.png

    这里需要额外创建5张表,分别存储Action\Guard\State\StateMachine\Transition,比较复杂。

  3. 所以这里创建了一张表biz_order_state_machine_context,结构很简单:bizOrderId,contextStr,curStatus,updateTime,其中关键是contextStr,用于存储与redis中相同的内容

    Repository

    有两个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的持久化就交代完了,下面就是最关键的,怎么利用状态机来串联业务,下一节将会详细描述。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 214,313评论 6 496
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 91,369评论 3 389
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 159,916评论 0 349
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 57,333评论 1 288
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 66,425评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 50,481评论 1 292
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,491评论 3 412
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 38,268评论 0 269
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,719评论 1 307
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,004评论 2 328
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,179评论 1 342
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,832评论 4 337
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,510评论 3 322
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,153评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,402评论 1 268
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 47,045评论 2 365
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,071评论 2 352

推荐阅读更多精彩内容