Spring StateMachine-加強版

上一章對狀態機Spring StateMachine做了基礎介紹，這次重點說明一下Spring StateMachine缺點。然后針對這個做具體優化

目標：

1.提高代碼復用率

2.修復一些bug

3.讓使用姿勢更加舒服（本人很懶，不想重復勞動^_^）

4.單據密等

5.單據加鎖

1.缺點：

　　1. Spring StateMachine是一個“重量級”狀態機框架，說他重是不框架本身比較龐大，而是說他的創建比較笨重，生產上我們一般都是用工廠模式來創建狀態機，這樣一來每一個請求都會創建一個StateMachine對象，這樣一來在高並發場景下，會創建N個StateMachine對象，對內存壓力就大了。（做過測試2分鍾內60w個並發請求，內存消耗算太大幾百兆而已）

　　2.StateMachine對spring事務不支持。（關鍵很致命）

　　3.stateMachine無法拋出異常，異常會被狀態機給消化掉。

　　4.出生時間較短，文檔和相關資料不是很全，且還存在一些bug。

（別問我為啥選它，問了也白問^_^）

針對上面問題，具體解決方案；

 

2.解決方案：

　1.“重量級" 級問題

　　a.第一種方式：利用Threadlocal來儲存stateMachine,這樣可以達到stateNachine局部復用。(簡單粗暴，注意手動釋放Threadlocal，不然會出現內存溢出，意義不大)

　　b.自己定義一個對象池，將每次利用完成的stateMachine初始化后放回池里，每次取對象都從對象池中獲取stateMachine。（相對復雜，處理不好容易出現並發性問題）

   2.StateMachine對spring事務不支持

　　利用@Transactional 注解特性將這個注解加到類上，這樣可以保證事務可以傳遞。

    3. stateMachine無法拋出異常，異常會被狀態機給消化掉

　　利用反射，重寫源碼就可以解決。

 

3.源代碼解決：

這里不解決“重量級" 級問題，原因是如果真的有那么高的並發量，建議不要是用stateMachine,而去使用squirrel-foundation，這個更加輕量級而且沒有上面的問題。

這里主要是針對stateMachine做二次包裝，目的是為了提高代碼復用率和方便開發

基礎配置類：狀態機工廠配置

 /**
　* 狀態機工廠基礎配置類
　*/
@Configuration
@EnableStateMachineFactory(name = "OneStateMachineFactory")
public class OneStateMachineConfig extends StateMachineConfigurerAdapter<StateEnum, EventEnum> {
　　
　　/**
　　　* 狀態機配置類
     **/
    @Resource
    private OutBoundBuilderMachineConfig outBoundBuilderMachineConfig;

    @Override
    public void configure(StateMachineStateConfigurer<StateEnum, EventEnum> states)
            throws Exception {
        outBoundBuilderMachineConfig.configureState(states);
    }

    @Override
    public void configure(StateMachineTransitionConfigurer<StateEnum, EventEnum> transitions)
            throws Exception {
        outBoundBuilderMachineConfig.configureStateBindEvent(transitions);

    }
　　
　　/**
　　 * 擴展StateMachinePersist
　　 */
    @Bean(name = "oneFactoryPersist")
    public StateMachinePersister<StateEnum, EventEnum, EventObj<StateEnum, EventEnum>> factoryPersister(){
        return new DefaultStateMachinePersister(new ExtStateMachinePersist<String, String, EventObj<String, String>>());
    }
　　
　　/**
　　　* 擴展StateMachineFactory
　　　*/
    @Bean(name = "oneExtStateMachineFactory")
    public ExtStateMachineFactory<StateEnum, EventEnum> extStateMachineFactory(@Autowired @Qualifier("OneStateMachineFactory") StateMachineFactory stateMachineFactory) {
        return new ExtStateMachineFactory(stateMachineFactory);
    }

　　/**
     * 擴展StateMachineBuilder
     */
    @Bean(name = "oneStateMachineBuilder")
    public ExtStateMachineBuilder<StateEnum, EventEnum> extStateMachineBuilder(@Autowired @Qualifier("outBoundExtStateMachineFactory") ExtStateMachineFactory extStateMachineFactory) {
        return new ExtStateMachineBuilder(extStateMachineFactory, factoryPersister());
    }

}

狀態配置類：業務邏輯狀態機配置

/**
 * 狀態機配置類
 **/
@Service
public class OutBoundBuilderMachineConfig {
　　
　　/**
　　 * 業務代碼實現service
　　 */　
    @Resource
    privateCreateAction createAction;

    @Resource
    private wareHouseAction wareHouseAction;

    @Resource
    private CancelAction cancelAction;

    @Resource
    private ShelfAction shelfAction;

    @Resource
    private StockAction stockAction;


    /**
     * 構建出庫狀態綁定關系
     *
     * @param states 狀態機狀態
     */
    public void configureState(StateMachineStateConfigurer<StateEnum, EventEnum> states) throws Exception {
        states.withStates()
                .initial(OutBoundStateEnum.INIT)
                .states(EnumSet.allOf(OutBoundStateEnum.class));
    }

    /**
     * 構建狀態機狀態綁定事件關聯關系
     *
     * @param transitions 參數
     */
    public void configureStateBindEvent(StateMachineTransitionConfigurer<StateEnum, EventEnum> transitions) throws Exception {

        transitions
                //創建
                .withExternal().source(OutBoundStateEnum.INIT).target(StateEnum.WAREHOUSE).action(createAction).event(EventEnum.obd_create).and()
                //倉操作
                .withExternal().source(OutBoundStateEnum.WAREHOUSE).target(StateEnum.STOCK).action(wareHouseAction).event(EventEnum.obd_warehouse).and()//揀貨
                .withExternal().source(OutBoundStateEnum.SHELF).target(StateEnum.STOCK).action(shelfAction).event(EventEnum.obd_shelf).and()
                //庫操作
                .withExternal().source(OutBoundStateEnum.STOCK).target(StateEnum.OUT_STOCK).action(stockAction).event(EventEnum.obd_stock).and()//取消
                .withExternal().source(OutBoundStateEnum.WAREHOUSE).target(StateEnum.CANCEL).action(cancelAction).event(EventEnum.obd_cancel);
    }

}

擴展StateMachinePersist：Statemachine通過實現StateMachinePersist接口，write和read當前狀態機的狀態，這里可以做定制化處理

package com.ext.statemachine.core;

import org.springframework.statemachine.StateMachineContext;
import org.springframework.statemachine.StateMachinePersist;
import org.springframework.statemachine.support.DefaultStateMachineContext;


public class ExtStateMachinePersist<S, E, T extends EventObj<S, E>> implements StateMachinePersist<S, E, T> {

    @Override
    public void write(StateMachineContext<S, E> context, T contextObj) throws Exception {

    }

    @Override
    public StateMachineContext<S, E> read(T contextObj) throws Exception {
        StateMachineContext<S, E> result =new DefaultStateMachineContext<S, E>(contextObj.getState(), null, null, null, null, contextObj.getBizNo());
        return result;
    }
}

擴展ExtStateMachineFactory，可以不做，為了定制化作准備

import org.springframework.statemachine.config.StateMachineFactory;
import org.springframework.statemachine.support.AbstractStateMachine;

import java.util.UUID;

public class ExtStateMachineFactory<S, E> implements StateMachineFactory<S, E>{
    private StateMachineFactory<S, E> stateMachineFactory;

    public ExtStateMachineFactory(StateMachineFactory<S, E> stateMachineFactory) {
        this.stateMachineFactory = stateMachineFactory;
    }


    @Override
    public ExtStateMachine<S, E> getStateMachine() {
        return new ExtStateMachine<>((AbstractStateMachine<S, E>) stateMachineFactory.getStateMachine());
    }

    @Override
    public ExtStateMachine<S, E> getStateMachine(String machineId) {
        return new ExtStateMachine<>((AbstractStateMachine<S, E>) stateMachineFactory.getStateMachine(machineId));
    }

    @Override
    public ExtStateMachine<S, E> getStateMachine(UUID uuid) {
        return new ExtStateMachine<>((AbstractStateMachine<S, E>)stateMachineFactory.getStateMachine(uuid));
    }

}

擴展ExtStateMachineBuilder，這個包裝是為了簡化代碼，可以不做

import org.springframework.statemachine.persist.StateMachinePersister;


public class ExtStateMachineBuilder<S,E> {
    private ExtStateMachineFactory<S, E> extStateMachineFactory;
    private StateMachinePersister<S, E, EventObj<S, E>>factoryPersister;

    public ExtStateMachineBuilder(ExtStateMachineFactory<S, E> extStateMachineFactory, StateMachinePersister<S, E, EventObj<S, E>> factoryPersister) {
        this.extStateMachineFactory = extStateMachineFactory;
        this.factoryPersister = factoryPersister;
    }

    public ExtStateMachine build(EventObj<S, E> eventObj){
        ExtStateMachine machine =  extStateMachineFactory.getStateMachine(eventObj.getBizNo());
        try {
            factoryPersister.restore(machine, eventObj);
            machine.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return machine;
    }
}

擴展Action: 解決stateMachine無法拋出異常，異常會被狀態機給消化掉，定制化一下

public class ExtStateMachine<S, E> implements StateMachine<S, E> {
    private AbstractStateMachine stateMachine;

    public ExtStateMachine(AbstractStateMachine<S, E> stateMachine) {
        this.stateMachine = stateMachine;
    }

　　/**
　　 * 解決stateMachine異常無法拋出異常問題
    **/
    public Exception getException() {
        Exception exception = null;
        try {
            Field field = AbstractStateMachine.class.getDeclaredField("currentError");
            field.setAccessible(true);
            Object ex = field.get(stateMachine);
            if (ex != null) {
                exception = (Exception)ex;
            }
        } catch (NoSuchFieldException | IllegalAccessException e) {
            e.printStackTrace();
        }
        return exception;
    }

    @Override
    public State<S, E> getInitialState() {
        return stateMachine.getInitialState();
    }

    @Override
    public ExtendedState getExtendedState() {
        return stateMachine.getExtendedState();
    }

    @Override
    public StateMachineAccessor<S, E> getStateMachineAccessor() {
        return stateMachine.getStateMachineAccessor();
    }

    @Override
    public void setStateMachineError(Exception exception) {
        stateMachine.setStateMachineError(exception);
    }

    @Override
    public boolean hasStateMachineError() {
        return stateMachine.hasStateMachineError();
    }

    @Override
    public UUID getUuid() {
        return stateMachine.getUuid();
    }

    @Override
    public String getId() {
        return stateMachine.getId();
    }

    @Override
    public void start() {
        stateMachine.start();
    }

    @Override
    public void stop() {
        stateMachine.stop();
    }

    @Override
    @Deprecated
    public boolean sendEvent(Message<E> event) {
        throw new UnsupportedOperationException("unsupported");
    }

    @Override
    @Deprecated
    public boolean sendEvent(E event) {
        throw new UnsupportedOperationException("unsupported");
    }


    public boolean sendEvent(EventObj eventObj) {
        boolean result = stateMachine.sendEvent(MessageBuilder
                .withPayload(eventObj.getEvent())
                .setHeader(EventObj.PARAM_NAME, eventObj.getParam())
                .setHeader(EventObj.BIZ_NO_NAME, eventObj.getBizNo())
                .build());
        Exception ex = getException();
        if (ex != null) {
            if (ex instanceof RuntimeException) {
                throw (RuntimeException) ex;
            } else {
                throw new RuntimeException(ex);
            }
        }
        if (!result) {
            throw new StateErrorException(ExceptionCodeParam.NOT_FIND_ACTION, String.format("狀態機事件未匹配到action, source=%s, event=%s, bizNo=%s", eventObj.getState().toString(), eventObj.getEvent(), eventObj.getBizNo()));
        }
        return result;
    }

    @Override
    public State<S, E> getState() {
        return stateMachine.getState();
    }

    @Override
    public Collection<State<S, E>> getStates() {
        return stateMachine.getStates();
    }

    @Override
    public Collection<Transition<S, E>> getTransitions() {
        return stateMachine.getTransitions();
    }

    @Override
    public boolean isComplete() {
        return stateMachine.isComplete();
    }

    @Override
    public void addStateListener(StateMachineListener<S, E> listener) {
        stateMachine.addStateListener(listener);
    }

    @Override
    public void removeStateListener(StateMachineListener<S, E> listener) {
        stateMachine.removeStateListener(listener);
    }
}

EventObj：定制化參數對象, 簡化狀態機參數傳遞

public class EventObj<S, E> {
    public static String BIZ_NO_NAME = "bizNo";
    public static String PARAM_NAME = "param"; //heard key

    private S state;   //當前狀態source狀態
    private E event;  //當時狀態機事件
    private String bizNo; //redis 單據鎖key
    private Object param; //參數對象


    public EventObj(S state, E event, String bizNo, Object param) {
        this.state = state;
        this.event = event;
        this.bizNo = bizNo;
        this.param = param;
    }

    public EventObj() {}

    public S getState() {
        return state;
    }

    public void setState(S state) {
        this.state = state;
    }

    public E getEvent() {
        return event;
    }

    public void setEvent(E event) {
        this.event = event;
    }

    public String getBizNo() {
        return bizNo;
    }

    public void setBizNo(String bizNo) {
        this.bizNo = bizNo;
    }

    public Object getParam() {
        return param;
    }

    public void setParam(Object param) {
        this.param = param;
    }

    public static Builder builder() {
        return new Builder();
    }


    public static class Builder<S, E> {
        private EventObj<S, E> eventObj;
        public Builder() {
            eventObj = new EventObj<S, E>();
        }

        public EventObj build() {
            return eventObj;
        }

        public EventObj state(S state) {
            eventObj.setState(state);
            return eventObj;
        }

        public EventObj event(E event) {
            eventObj.setEvent(event);
            return eventObj;
        }

        public EventObj event(String bizNo) {
            eventObj.setBizNo(bizNo);
            return eventObj;
        }

        public EventObj param(Object param) {
            eventObj.setParam(param);
            return eventObj;
        }

    }
}

擴展stateContext: 通過stateMachine上下文獲取參數，更加方便處理業務

public class ExtStateContext<S, E, P, Et> {
    private transient StateContext<S, E> oriStateContext;
    private String bizNo; //redis 單據唯一標識
    private S state; //狀態機state source
    private E event; //事件
    private Et entity; //單據對象
    private S target; //狀態機 state target
    private P param; //參數對象

    public ExtStateContext(StateContext<S, E> stateContext) {
        this.oriStateContext = stateContext;
        this.state = stateContext.getStateMachine().getState().getId();
        this.event = stateContext.getEvent();
        this.bizNo = stateContext.getMessageHeaders().get(EventObj.BIZ_NO_NAME, String.class);
        this.param = (P)stateContext.getMessageHeaders().get(EventObj.PARAM_NAME);
        this.target = stateContext.getTarget().getId();
    }

    public S getState() {
        return state;
    }

    public E getEvent() {
        return event;
    }

    public String getBizNo() {
        return bizNo;
    }

    public P getParam() {
        return param;
    }

    public Et getEntity() {
        return entity;
    }

    public S getTarget() {
        return target;
    }

    protected void setEntity(Et entity) {
        this.entity = entity;
    }

    public StateContext<S, E> getOriStateContext() {
        return oriStateContext;
    }
}

StateMachineSendService：包裝狀態機發送類（公共類）

@Service
@Slf4j
public class StateMachineSendService {

    @Resource(name = "outBoundStateMachineBuilder")
    @Lazy
    private ExtStateMachineBuilder extStateMachineBuilder;

    /**
     * @param state 狀態
     * @param event 事假
     * @param bizNo 唯一建
     * @param param 參數
     */
    public void sendEvent(StateEnum state, EventEnum event, String bizNo, Object param) {
        EventObj<StateEnum, EventEnum> eventObj = new EventObj<>(state, event, bizNo, param);
        ExtStateMachine machine = extStateMachineBuilder.build(eventObj);
        machine.sendEvent(eventObj);
    }
}

Active擴展

@Transactional(value = "ofcTransactionManager", rollbackFor = Exception.class)
@Slf4j
public abstract class ExtAction<S, E, P, Et> implements Action<S, E> {

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Value("${ofc.try.lock.time:2}")
    public Long tryLockTime;

    @Override
    public void execute(StateContext<S, E> context) {
        ExtStateContext<S, E, P, Et> extContext = new ExtStateContext<>(context);
        //構建鎖的屬性
        boolean isLock = false;
        try {
            // 鎖檢查
            if (needLock()) {
                isLock = tryLock(getRedisKey(extContext));
                if (!isLock) {
                    throw new StateErrorException(ExceptionCodeParam.NOT_GET_LOCK, "ofc 沒有獲取到鎖資源");
                }
            }
            // 期望狀態與實際狀態檢查
            if (checkState(extContext)) {
                log.info("狀態機 命中冪等 end: context={}", JSON.toJSONString(extContext));
                return;
            }
            String logMsg = JSON.toJSONString(extContext);
            log.info("狀態機 start: context={}", logMsg);
            handle(extContext);
            log.info("狀態機 end: context={}", logMsg);

        } catch (Exception e) {
            StateMachine stateMachine = context.getStateMachine();
            stateMachine.setStateMachineError(e);
            log.error("狀態機 error: error={}, context={}", e.getMessage(), JSON.toJSONString(extContext));
            throw e;
        } finally {
            if (isLock) {
                unlock(extContext);
            }
        }
    }

    /**
     * 獲取鎖資源
     *
     * @param key key
     * @return true/false
     */
    private boolean tryLock(String key) {
        String lockValue = UUID.randomUUID().toString() + System.currentTimeMillis();
        Boolean isLock = redisTemplate.opsForValue().setIfAbsent(key, lockValue, tryLockTime, TimeUnit.MINUTES);
        return Objects.isNull(isLock) ? false : isLock;
    }

    /**
     * 釋放鎖
     */
    private void unlock(ExtStateContext<S, E, P, Et> extContext) {
        String redisKey = getRedisKey(extContext);
        redisTemplate.opsForValue().getOperations().delete(redisKey);
    }

    private boolean checkState(ExtStateContext<S, E, P, Et> extContext) {
        Et entity = getEntity(extContext);
        extContext.setEntity(entity);
        S currState = getCurrState(entity);
        //密等
        if (isIdempotent(extContext, currState)) {
            return true;
        }
        if (!extContext.getState().equals(currState)) {
            throw new StateErrorException(ExceptionCodeParam.STATE_ERROR, String.format("狀態不一致, 期望狀態=%s, 實際狀態=%s", extContext.getState(), currState));
        }
        return false;
    }

    public abstract void handle(ExtStateContext<S, E, P, Et> context);

    public abstract Et getEntity(ExtStateContext<S, E, P, Et> context);

    /**
     * 獲取當前節狀態
     *
     * @param entity 對象
     * @return
     */
    public abstract S getCurrState(Et entity);

    /**
     * 是否需要加redis鎖
     *
     * @return boolean
     */
    public boolean needLock() {
        return true;
    }

    public String getRedisKey(ExtStateContext<S, E, P, Et> context) {
        if (StringUtils.isEmpty(context.getBizNo())) {
            throw new StateErrorException(ExceptionCodeParam.LOCK_KEY_EMPTY, "沒有獲取到鎖key");
        }
        log.info("ofc reids lock key: {}","action-lock-" + context.getBizNo());
        return "action-lock-" + context.getBizNo();
    }

    /**
     * 冪等判斷
     *
     * @param context   參數
     * @param currState 當前狀態
     * @return true/false
     */
    public boolean isIdempotent(ExtStateContext<S, E, P, Et> context, S currState) {
        if (context.getTarget() != null && context.getTarget().equals(currState)) {
            if (context.getState() != null && context.getState().equals(context.getTarget())) {
                throw new StateErrorException(ExceptionCodeParam.STATE_IDEMPOTENT_ERROR, "狀態機不符合冪等規則（初態和次態配置相同）");
            }
            return true;
        } else {
            return false;
        }
    }

 

使用姿勢

/**
     * 創建單據流程
     *
     * @param outboundBody 參數
     */
    @Override
    public void createbound(Body body) {
        log.info("創建單據流程參數:{}", JSON.toJSONString(body));
        //構建key
        String key = body.getBizNo;
        try {
　　　　　　　//調用狀態機
            stateMachineSendService.sendEvent(StateEnum.INIT, EventEnum.obd_create, key, body);
        } catch (OutboundException e) {
            if (OutboundException.REPEAT_ERROR.equals(e.getCode())) {
                log.warn("重復消費，param:{}", JSON.toJSONString(body));
            } else {
                throw e;
            }
        }
    }

@Service
@Slf4j
public class OutboundCreateOutboundAction extends ExtAction<StateEnum, EventEnum, Body, OutHeaderEntity> {

    @Override
    public void handle(ExtStateContext<StateEnum, EventEnum, Body, OutHeaderEntity> context) {
        //參數
        Body bodyParam = context.getParam();
        //單據
         OutHeaderEntity entity = context.getEntity();
　　　　　//業務代碼
    }

}        

到這里基本可以使用狀態機，可以滿足基本需求，而且使用更加方便和安全了。我們的目標已經全部實現。源代碼相對簡單，這里不做源碼分析了。

 

 

引用相關文章：

https://segmentfault.com/a/1190000009906317

https://www.jianshu.com/p/9ee887e045dd