Spring之状态机讲解网站首页 其他

Spring之状态机讲解

爱吃牛肉的大老虎 2024-07-09 10:33:21

简介Spring之状态机讲解

1 状态机

1.1 什么是状态

先来解释什么是状态（ State ）。现实事物是有不同状态的，例如一个自动门，就有 open 和 closed 两种状态。我们通常所说的状态机是有限状态机，也就是被描述的事物的状态的数量是有限个，例如自动门的状态就是两个 open 和 closed 。

状态机，也就是 State Machine ，不是指一台实际机器，而是指一个数学模型。说白了，一般就是指一张状态转换图。

状态机的全称是有限状态自动机，自动两个字也是包含重要含义的。给定一个状态机，同时给定它的当前状态以及输入，那么输出状态时可以明确的运算出来的。例如对于自动门，给定初始状态 closed ，给定输入“开门”，那么下一个状态时可以运算出来的。

1.2 四大概念

下面来给出状态机的四大概念：

State：状态，一个状态机至少要包含两个状态。
Event：事件，事件就是执行某个操作的触发条件或者口令。对于自动门，“按开门按钮”就是一个事件。
Action：动作，事件发生以后要执行动作。例如事件是“按开门按钮”，动作是“开门”。编程的时候，一个 Action一般就对应一个函数。
Transition：变换，也就是从一个状态变化为另一个状态。例如“开门过程”就是一个变换。

1.3 状态机

有限状态机（Finite-state machine,FSM），又称有限状态自动机，简称状态机，是表示有限个状态以及在这些状态之间的转移和动作等行为的数学模型。
FSM是一种算法思想，简单而言，有限状态机由一组状态、一个初始状态、输入和根据输入及现有状态转换为下一个状态的转换函数组成。
其作用主要是描述对象在它的生命周期内所经历的状态序列，以及如何响应来自外界的各种事件。

1.4 spring statemachine

Spring Statemachine是应用程序开发人员在Spring应用程序中使用状态机概念的框架
Spring Statemachine旨在提供以下功能：

易于使用的扁平单级状态机，用于简单的使用案例。
分层状态机结构，以简化复杂的状态配置。
状态机区域提供更复杂的状态配置。
使用触发器，转换，警卫和操作。
键入安全配置适配器。
生成器模式，用于在Spring Application上下文之外使用的简单实例化通常用例的食谱
基于Zookeeper的分布式状态机
状态机事件监听器。
UML Eclipse Papyrus建模。
将计算机配置存储在永久存储中。
Spring IOC集成将bean与状态机关联起来。

状态机功能强大，因为行为始终保证一致，使调试相对容易。这是因为操作规则是在机器启动时写成的。这个想法是你的应用程序可能存在于有限数量的状态中，某些预定义的触发器可以将你的应用程序从一个状态转移到另一个状态。此类触发器可以基于事件或计时器。

2 示例Demo

2.1 订单状态图

做需求时，需要了解以下六种元素：起始、终止、现态、次态（目标状态）、动作、条件，我们就可以完成一个状态机图了：
在这里插入图片描述

以订单为例：以从待支付状态转换为待发货状态为例

现态：是指当前所处的状态。待支付
条件：又称为“事件”，当一个条件被满足，将会触发一个动作，或者执行一次状态的迁移。支付事件
动作：条件满足后执行的动作。动作执行完毕后，可以迁移到新的状态，也可以仍旧保持原状态。动作不是必需的，当条件满足后，也可以不执行任何动作，直接迁移到新状态。状态转换为待发货
次态：条件满足后要迁往的新状态。“次态”是相对于“现态”而言的，“次态”一旦被激活，就转变成新的“现态”了。

注意：

避免把某个“程序动作”当作是一种“状态”来处理。那么如何区分“动作”和“状态”？“动作”是不稳定的，即使没有条件的触发，“动作”一旦执行完毕就结束了；而“状态”是相对稳定的，如果没有外部条件的触发，一个状态会一直持续下去。
状态划分时漏掉一些状态，导致跳转逻辑不完整。所以在设计状态机时，我们需要反复的查看设计的状态图或者状态表，最终达到一种牢不可破的设计方案。

2.2 建表

表结构设计如下：

CREATE TABLE `tb_order` (
      `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键ID',
      `order_code` varchar(128) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '订单编码',
      `status` smallint(3) DEFAULT NULL COMMENT '订单状态',
      `name` varchar(64) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '订单名称',
      `price` decimal(12,2) DEFAULT NULL COMMENT '价格',
      `delete_flag` tinyint(2) NOT NULL DEFAULT '0' COMMENT '删除标记，0未删除  1已删除',
      `create_time` timestamp DEFAULT CURRENT_TIMESTAMP  COMMENT '创建时间',
      `update_time` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
      
      `remark` varchar(64) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '备注',
      PRIMARY KEY (`id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=6 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='订单表';

    /*Data for the table `tb_order` */

   insert  into `tb_order`(`order_code`,`status`,`name`,`price`,`delete_flag`) values
    ('A111',1,'A','22.00',0),
    ('A111',1,'订单A','22.00',0),
    ('A111',1,'订单A','22.00',0),
    ('A111',1,'订单A','22.00',0);

2.3 依赖和配置

2.3.1 pom.xml

<parent>
	<groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-dependencies</artifactId>
    <version>2.6.11</version>
</parent>

<properties>
    <maven.compiler.source>17</maven.compiler.source>
    <maven.compiler.target>17</maven.compiler.target>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.83</version>
    </dependency>

    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.5.0</version>
    </dependency>
    <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.2.8</version>
        </dependency>

   
    <!--状态机-->
    <dependency>
        <groupId>org.springframework.statemachine</groupId>
        <artifactId>spring-statemachine-starter</artifactId>
        <version>2.0.1.RELEASE</version>
    </dependency>
</dependencies>

2.3.2 application.yml

spring:
  application:
    name: state-machine-demo
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    driver-class-name: com.mysql.cj.jdbc.Driver
    username: root
    password: root
    url: jdbc:mysql://127.0.0.1:3306/test?useSSL=false&serverTimezone=GMT&characterEncoding=utf8
mybatis-plus:
  mapper-locations: classpath:mapper/*.xml
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

2.4 状态机配置

2.4.1 定义状态机状态和事件

订单状态

public enum OrderStatus {
        // 待支付，待发货，待收货，已完成
        WAIT_PAYMENT(1, "待支付"),
        WAIT_DELIVER(2, "待发货"),
        WAIT_RECEIVE(3, "待收货"),
        FINISH(4, "已完成");
        private Integer key;
        private String desc;
        OrderStatus(Integer key, String desc) {
            this.key = key;
            this.desc = desc;
        }
        public Integer getKey() {
            return key;
        }
        public String getDesc() {
            return desc;
        }
        public static OrderStatus getByKey(Integer key) {
            for (OrderStatus e : values()) {
                if (e.getKey().equals(key)) {
                    return e;
                }
            }
            throw new RuntimeException("enum not exists.");
        }
    }

事件：

public enum OrderStatusChangeEvent {
        // 支付，发货，确认收货
        PAYED, DELIVERY, RECEIVED;
}

2.4.2 定义状态机规则

@Configuration
@EnableStateMachine(name = "orderStateMachine")
public class OrderStateMachineConfig extends StateMachineConfigurerAdapter<OrderStatus, OrderStatusChangeEvent> {
        /**
         * 配置状态
         *
         * @param states
         * @throws Exception
         */
        public void configure(StateMachineStateConfigurer<OrderStatus, OrderStatusChangeEvent> states) throws Exception {
            states
                    .withStates()
                    .initial(OrderStatus.WAIT_PAYMENT)
                    .states(EnumSet.allOf(OrderStatus.class));
        }
        /**
         * 配置状态转换事件关系
         *
         * @param transitions
         * @throws Exception
         */
        public void configure(StateMachineTransitionConfigurer<OrderStatus, OrderStatusChangeEvent> transitions) throws Exception {
            transitions
                    //支付事件:待支付-》待发货
                    .withExternal().source(OrderStatus.WAIT_PAYMENT).target(OrderStatus.WAIT_DELIVER).event(OrderStatusChangeEvent.PAYED)
                    .and()
                    //发货事件:待发货-》待收货
                    .withExternal().source(OrderStatus.WAIT_DELIVER).target(OrderStatus.WAIT_RECEIVE).event(OrderStatusChangeEvent.DELIVERY)
                    .and()
                    //收货事件:待收货-》已完成
                    .withExternal().source(OrderStatus.WAIT_RECEIVE).target(OrderStatus.FINISH).event(OrderStatusChangeEvent.RECEIVED);
        }
    }

2.4.3 配置持久化

2.4.3.1 持久化到内存

@Configuration
@Slf4j
public class Persist<E, S> {
    /**
     * 持久化到内存map中
     *
     * @return
     */
    @Bean(name = "stateMachineMemPersister")
    public static StateMachinePersister getPersister() {
        return new DefaultStateMachinePersister(new StateMachinePersist() {
            @Override
            public void write(StateMachineContext context, Object contextObj) throws Exception {
                log.info("持久化状态机,context:{},contextObj:{}", JSON.toJSONString(context), JSON.toJSONString(contextObj));
                map.put(contextObj, context);
            }
            @Override
            public StateMachineContext read(Object contextObj) throws Exception {
                log.info("获取状态机,contextObj:{}", JSON.toJSONString(contextObj));
                StateMachineContext stateMachineContext = (StateMachineContext) map.get(contextObj);
                log.info("获取状态机结果,stateMachineContext:{}", JSON.toJSONString(stateMachineContext));
                return stateMachineContext;
            }
            private Map map = new HashMap();
        });
    }   
}

2.4.3.2 持久化到redis

<!-- redis持久化状态机 -->
<dependency>
    <groupId>org.springframework.statemachine</groupId>
    <artifactId>spring-statemachine-redis</artifactId>
    <version>1.2.9.RELEASE</version>
</dependency>

在application.yml中添加

spring:
  redis:
    database: 0
    host: localhost
    jedis:
      pool:
        max-active: 8
        max-idle: 8
        max-wait: ''
        min-idle: 0
    password: ''
    port: 6379
    timeout: 0

在Persist中添加如下：

@Resource
private RedisConnectionFactory redisConnectionFactory;
 /**
  * 持久化到redis中，在分布式系统中使用
  *
  * @return
  */
 @Bean(name = "stateMachineRedisPersister")
 public RedisStateMachinePersister<E, S> getRedisPersister() {
     RedisStateMachineContextRepository<E, S> repository = new RedisStateMachineContextRepository<>(redisConnectionFactory);
     RepositoryStateMachinePersist p = new RepositoryStateMachinePersist<>(repository);
     return new RedisStateMachinePersister<>(p);
 }

使用redis持久化：

//使用name 指定redis持久化
@Resource(name="stateMachineRedisPersister")
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineRedisPersister;

private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
    boolean result = false;
    try {
        //启动状态机
        orderStateMachine.start();
        //尝试恢复状态机状态
        stateMachineRedisPersister.restore(orderStateMachine, String.valueOf(order.getId()));
        Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
        result = orderStateMachine.sendEvent(message);
        //持久化状态机状态
        stateMachineRedisPersister.persist(orderStateMachine, String.valueOf(order.getId()));
    } catch (Exception e) {
        log.error("订单操作失败:{}", e);
    } finally {
        orderStateMachine.stop();
    }
    return result;
}

2.5 业务系统

2.5.1 controller

@RestController
@RequestMapping("/order")
public class OrderController {
    @Resource
    private OrderService orderService;
    /**
     * 根据id查询订单
     *
     * @return
     */
    @RequestMapping("/getById")
    public Order getById(@RequestParam("id") Long id) {
        //根据id查询订单
        Order order = orderService.getById(id);
        return order;
    }
    /**
     * 创建订单
     *
     * @return
     */
    @RequestMapping("/create")
    public String create(@RequestBody Order order) {
        //创建订单
        orderService.create(order);
        return "sucess";
    }
    /**
     * 对订单进行支付
     *
     * @param id
     * @return
     */
    @RequestMapping("/pay")
    public String pay(@RequestParam("id") Long id) {
        //对订单进行支付
        orderService.pay(id);
        return "success";
    }

    /**
     * 对订单进行发货
     *
     * @param id
     * @return
     */
    @RequestMapping("/deliver")
    public String deliver(@RequestParam("id") Long id) {
        //对订单进行确认收货
        orderService.deliver(id);
        return "success";
    }
    /**
     * 对订单进行确认收货
     *
     * @param id
     * @return
     */
    @RequestMapping("/receive")
    public String receive(@RequestParam("id") Long id) {
        //对订单进行确认收货
        orderService.receive(id);
        return "success";
    }
}

2.5.2 servie

@Service("orderService")
@Slf4j
public class OrderServiceImpl extends ServiceImpl<OrderMapper, Order> implements OrderService {
   @Resource
   private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;
   @Resource
   private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineMemPersister;
   @Resource
   private OrderMapper orderMapper;
   /**
    * 创建订单
    *
    * @param order
    * @return
    */
   public Order create(Order order) {
       order.setStatus(OrderStatus.WAIT_PAYMENT.getKey());
       orderMapper.insert(order);
       return order;
   }
   /**
    * 对订单进行支付
    *
    * @param id
    * @return
    */
   public Order pay(Long id) {
       Order order = orderMapper.selectById(id);
       log.info("线程名称：{},尝试支付，订单号：{}" ,Thread.currentThread().getName() , id);
       if (!sendEvent(OrderStatusChangeEvent.PAYED, order)) {
           log.error("线程名称：{},支付失败, 状态异常，订单信息：{}", Thread.currentThread().getName(), order);
           throw new RuntimeException("支付失败, 订单状态异常");
       }
       return order;
   }
   /**
    * 对订单进行发货
    *
    * @param id
    * @return
    */
   public Order deliver(Long id) {
       Order order = orderMapper.selectById(id);
       log.info("线程名称：{},尝试发货，订单号：{}" ,Thread.currentThread().getName() , id);
       if (!sendEvent(OrderStatusChangeEvent.DELIVERY, order)) {
           log.error("线程名称：{},发货失败, 状态异常，订单信息：{}", Thread.currentThread().getName(), order);
           throw new RuntimeException("发货失败, 订单状态异常");
       }
       return order;
   }
   /**
    * 对订单进行确认收货
    *
    * @param id
    * @return
    */
   public Order receive(Long id) {
       Order order = orderMapper.selectById(id);
       log.info("线程名称：{},尝试收货，订单号：{}" ,Thread.currentThread().getName() , id);
       if (!sendEvent(OrderStatusChangeEvent.RECEIVED, order)) {
           log.error("线程名称：{},收货失败, 状态异常，订单信息：{}", Thread.currentThread().getName(), order);
           throw new RuntimeException("收货失败, 订单状态异常");
       }
       return order;
   }
   /**
    * 发送订单状态转换事件
    * synchronized修饰保证这个方法是线程安全的
    *
    * @param changeEvent
    * @param order
    * @return
    */
   private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
       boolean result = false;
       try {
           //启动状态机
           orderStateMachine.start();
           //尝试恢复状态机状态
           stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
           Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
           result = orderStateMachine.sendEvent(message);
           //持久化状态机状态
           stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
       } catch (Exception e) {
           log.error("订单操作失败:{}", e);
       } finally {
           orderStateMachine.stop();
       }
       return result;
   }
}

2.5.3 监听状态变化

@Component("orderStateListener")
@WithStateMachine(name = "orderStateMachine")
@Slf4j
public class OrderStateListenerImpl {
    @Resource
    private OrderMapper orderMapper;

    @OnTransition(source = "WAIT_PAYMENT", target = "WAIT_DELIVER")
    public void payTransition(Message<OrderStatusChangeEvent> message) {
        Order order = (Order) message.getHeaders().get("order");
        log.info("支付，状态机反馈信息：{}",  message.getHeaders().toString());
        //更新订单
        order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
        orderMapper.updateById(order);
        //TODO 其他业务
    }
    @OnTransition(source = "WAIT_DELIVER", target = "WAIT_RECEIVE")
    public void deliverTransition(Message<OrderStatusChangeEvent> message) {
        Order order = (Order) message.getHeaders().get("order");
        log.info("发货，状态机反馈信息：{}",  message.getHeaders().toString());
        //更新订单
        order.setStatus(OrderStatus.WAIT_RECEIVE.getKey());
        orderMapper.updateById(order);
        //TODO 其他业务
    }
    @OnTransition(source = "WAIT_RECEIVE", target = "FINISH")
    public void receiveTransition(Message<OrderStatusChangeEvent> message) {
        Order order = (Order) message.getHeaders().get("order");
        log.info("确认收货，状态机反馈信息：{}",  message.getHeaders().toString());
        //更新订单
        order.setStatus(OrderStatus.FINISH.getKey());
        orderMapper.updateById(order);
        //TODO 其他业务
    }
}

2.6 状态机存在的问题

stateMachine 无法抛出异常，异常会被状态机给消化掉

2.6.1 问题现象

从 orderStateMachine.sendEvent(message); 获取的结果无法感知到。无论执行正常还是抛出异常，都返回true。

@Resource
private OrderMapper orderMapper;

@Resource
private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;

@OnTransition(source = "WAIT_PAYMENT", target = "WAIT_DELIVER")
@Transactional(rollbackFor = Exception.class)
public void payTransition(Message<OrderStatusChangeEvent> message) {
    Order order = (Order) message.getHeaders().get("order");
    log.info("支付，状态机反馈信息：{}",  message.getHeaders().toString());
    try {
        //更新订单
        order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
        orderMapper.updateById(order);
        //TODO 其他业务
        //模拟异常
        if(Objects.equals(order.getName(),"A")){
            throw new RuntimeException("执行业务异常");
        }
    } catch (Exception e) {
        //如果出现异常，记录异常信息，抛出异常信息进行回滚
        log.error("payTransition 出现异常：{}",e);
        throw e;
    }
}

监听事件抛出异常，在发送事件中无法感知：

 private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
     boolean result = false;
     try {
         //启动状态机
         orderStateMachine.start();
         //尝试恢复状态机状态
         stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
         Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
          //事件执行异常了，依然返回true，无法感知异常
         result = orderStateMachine.sendEvent(message);
         if(result){
             //持久化状态机状态，如果根据true持久化，则会出现问题
             stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
         }
     } catch (Exception e) {
         log.error("订单操作失败:{}", e);
     } finally {
         orderStateMachine.stop();
     }
     return result;
 }

调试发现：发送事件和监听事件是一个线程，发送事件的结果是在监听操作执行完之后才返回

2.6.2 解决方案

保存异常到数据库或者内存中，进行判断
也可以通过接口：org.springframework.statemachine.StateMachine##getExtendedState##getVariables，方法把执行状态放入这个变量中

改造监听状态：把业务的执行结果进行保存，1成功，0失败

常量类：

public interface CommonConstants {
	String orderHeader="order";
    String payTransition="payTransition";
    String deliverTransition="deliverTransition";
    String receiveTransition="receiveTransition";
}

@Resource
private OrderMapper orderMapper;
@Resource
private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;

@OnTransition(source = "WAIT_PAYMENT", target = "WAIT_DELIVER")
@Transactional(rollbackFor = Exception.class)
public void payTransition(Message<OrderStatusChangeEvent> message) {
    Order order = (Order) message.getHeaders().get("order");
    log.info("支付，状态机反馈信息：{}",  message.getHeaders().toString());
    try {
        //更新订单
        order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
        orderMapper.updateById(order);
        //TODO 其他业务
        //模拟异常
        if(Objects.equals(order.getName(),"A")){
            throw new RuntimeException("执行业务异常");
        }
        //成功 则为1
        orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(),1);
    } catch (Exception e) {
        //如果出现异常，则进行回滚
        log.error("payTransition 出现异常：{}",e);
        //将异常信息变量信息中，失败则为0
        orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(), 0);
        throw e;
    }
}

发送事件改造：如果获取到业务执行异常，则返回失败，不进行状态机持久化

@Resource
private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;
@Resource
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineMemPersister;

/**
 * 发送订单状态转换事件
 * synchronized修饰保证这个方法是线程安全的
 *
 * @param changeEvent
 * @param order
 * @return
 */
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order){
    boolean result = false;
    try {
        //启动状态机
        orderStateMachine.start();
        //尝试恢复状态机状态
        stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
        Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
        result = orderStateMachine.sendEvent(message);
        if(!result){
            return false;
        }
        //获取到监听的结果信息
        Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(CommonConstants.payTransition + order.getId());
        //操作完成之后,删除本次对应的key信息
        orderStateMachine.getExtendedState().getVariables().remove(CommonConstants.payTransition+order.getId());
        //如果事务执行成功，则持久化状态机
        if(Objects.equals(1,Integer.valueOf(o))){
            //持久化状态机状态
            stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
        }else {
            //订单执行业务异常
            return false;
        }
    } catch (Exception e) {
        log.error("订单操作失败:{}", e);
    } finally {
        orderStateMachine.stop();
    }
    return result;
}