上一章对状态机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