如何設計一個狀態機！
1、什麼是狀態機
1.1 什麼是狀態
先來解釋什麼是 “狀態”（ State ）。現實事物是有不同狀態的，例如一個自動門，就有 open 和 closed 兩種狀態。我們通常所說的狀態機是有限狀態機，也就是被描述的事物的狀態的數量是有限個，例如自動門的狀態就是兩個 open 和 closed 。
狀態機，也就是 State Machine ，不是指一臺實際機器，而是指一個數學模型。說白了，一般就是指一張狀態轉換圖。例如，根據自動門的運行規則，我們可以抽象出下面這麼一個圖。
自動門有兩個狀態，open 和 closed ，closed 狀態下，如果讀取開門信號，那麼狀態就會切換爲 open 。open 狀態下如果讀取關門信號，狀態就會切換爲 closed 。
狀態機的全稱是有限狀態自動機，自動兩個字也是包含重要含義的。給定一個狀態機，同時給定它的當前狀態以及輸入，那麼輸出狀態時可以明確的運算出來的。例如對於自動門，給定初始狀態 closed ，給定輸入 “開門”，那麼下一個狀態時可以運算出來的。
這樣狀態機的基本定義我們就介紹完畢了。重複一下：狀態機是有限狀態自動機的簡稱，是現實事物運行規則抽象而成的一個數學模型。
1.2 四大概念
下面來給出狀態機的四大概念。
-
第一個是 State ，狀態。一個狀態機至少要包含兩個狀態。例如上面自動門的例子，有 open 和 closed 兩個狀態。
-
第二個是 Event ，事件。事件就是執行某個操作的觸發條件或者口令。對於自動門，“按下開門按鈕” 就是一個事件。
-
第三個是 Action ，動作。事件發生以後要執行動作。例如事件是 “按開門按鈕”，動作是 “開門”。編程的時候，一個 Action 一般就對應一個函數。
-
第四個是 Transition ，變換。也就是從一個狀態變化爲另一個狀態。例如 “開門過程” 就是一個變換。
1.3 狀態機
有限狀態機（Finite-state machine,FSM），又稱有限狀態自動機，簡稱狀態機，是表示有限個狀態以及在這些狀態之間的轉移和動作等行爲的數學模型。
FSM 是一種算法思想，簡單而言，有限狀態機由一組狀態、一個初始狀態、輸入和根據輸入及現有狀態轉換爲下一個狀態的轉換函數組成。
其作用主要是描述對象在它的生命週期內所經歷的狀態序列，以及如何響應來自外界的各種事件。
2、狀態機圖
做需求時，需要了解以下六種元素：起始、終止、現態、次態（目標狀態）、動作、條件，我們就可以完成一個狀態機圖了：
以訂單爲例：以從待支付狀態轉換爲待發貨狀態爲例
-
①現態：是指當前所處的狀態。待支付
-
②條件：又稱爲 “事件”，當一個條件被滿足，將會觸發一個動作，或者執行一次狀態的遷移。支付事件
-
③動作：條件滿足後執行的動作。動作執行完畢後，可以遷移到新的狀態，也可以仍舊保持原狀態。動作不是必需的，當條件滿足後，也可以不執行任何動作，直接遷移到新狀態。狀態轉換爲待發貨
-
④次態：條件滿足後要遷往的新狀態。“次態”是相對於 “現態” 而言的，“次態”一旦被激活，就轉變成新的 “現態” 了。
1、避免把某個 “程序動作” 當作是一種 “狀態” 來處理。那麼如何區分 “動作” 和“狀態”？“動作”是不穩定的，即使沒有條件的觸發，“動作”一旦執行完畢就結束了；而 “狀態” 是相對穩定的，如果沒有外部條件的觸發，一個狀態會一直持續下去。
2、狀態劃分時漏掉一些狀態，導致跳轉邏輯不完整。所以在設計狀態機時，我們需要反覆的查看設計的狀態圖或者狀態表，最終達到一種牢不可破的設計方案。
3、spring statemachine
3.1 狀態機 spring statemachine 概述
Spring Statemachine 是應用程序開發人員在 Spring 應用程序中使用狀態機概念的框架
Spring Statemachine 旨在提供以下功能：
-
易於使用的扁平單級狀態機，用於簡單的使用案例。
-
分層狀態機結構，以簡化複雜的狀態配置。
-
狀態機區域提供更復雜的狀態配置。
-
使用觸發器，轉換，警衛和操作。
-
鍵入安全配置適配器。
-
生成器模式，用於在 Spring Application 上下文之外使用的簡單實例化通常用例的食譜
-
基於 Zookeeper 的分佈式狀態機
-
狀態機事件監聽器。
-
UML Eclipse Papyrus 建模。
-
將計算機配置存儲在永久存儲中。
-
Spring IOC 集成將 bean 與狀態機關聯起來。
狀態機功能強大，因爲行爲始終保證一致，使調試相對容易。這是因爲操作規則是在機器啓動時寫成的。這個想法是你的應用程序可能存在於有限數量的狀態中，某些預定義的觸發器可以將你的應用程序從一個狀態轉移到另一個狀態。此類觸發器可以基於事件或計時器。
在應用程序之外定義高級邏輯然後依靠狀態機來管理狀態要容易得多。您可以通過發送事件，偵聽更改或僅請求當前狀態來與狀態機進行交互。
3.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 NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '創建時間',
`update_time` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' COMMENT '更新時間',
`create_user_code` varchar(32) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '創建人',
`update_user_code` varchar(32) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '更新人',
`version` int(11) NOT NULL DEFAULT '0' 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`(`id`,`order_code`,`status`,`name`,`price`,`delete_flag`,`create_time`,`update_time`,`create_user_code`,`update_user_code`,`version`,`remark`) values
(2,'A111',1,'A','22.00',0,'2022-10-15 16:14:11','2022-10-02 21:29:14','zhangsan','zhangsan',0,NULL),
(3,'A111',1,'訂單A','22.00',0,'2022-10-02 21:53:13','2022-10-02 21:29:14','zhangsan','zhangsan',0,NULL),
(4,'A111',1,'訂單A','22.00',0,'2022-10-02 21:53:13','2022-10-02 21:29:14','zhangsan','zhangsan',0,NULL),
(5,'A111',1,'訂單A','22.00',0,'2022-10-03 09:08:30','2022-10-02 21:29:14','zhangsan','zhangsan',0,NULL);1）引入依賴
<!-- redis持久化狀態機 -->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-redis</artifactId>
<version>1.2.9.RELEASE</version>
</dependency>
<!--狀態機-->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-starter</artifactId>
<version>2.0.1.RELEASE</version>
</dependency>2）定義狀態機狀態和事件
狀態枚舉：
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;
}3）定義狀態機規則和配置狀態機
@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);
}
}配置持久化：
@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();
});
}
@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);
}
}4）業務系統
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";
}
}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;
}
}監聽狀態的變化：
@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 其他業務
}
}3.3 測試驗證
1）驗證業務
-
新增一個訂單
http://localhost:8084/order/create
-
對訂單進行支付
http://localhost:8084/order/pay?id=2
-
對訂單進行發貨
http://localhost:8084/order/deliver?id=2
-
對訂單進行確認收貨
http://localhost:8084/order/receive?id=2
正常流程結束。如果對一個訂單進行支付了，再次進行支付，則會報錯：http://localhost:8084/order/pay?id=2
報錯如下：
2）驗證持久化
內存
使用內存持久化類持久化：
@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);
//持久化狀態機狀態
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}redis 持久化
引入依賴：
<!-- redis持久化狀態機 -->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-redis</artifactId>
<version>1.2.9.RELEASE</version>
</dependency>配置 yaml：
spring:
redis:
database: 0
host: localhost
jedis:
pool:
max-active: 8
max-idle: 8
max-wait: ''
min-idle: 0
password: ''
port: 6379
timeout: 0使用 redis 持久化類持久化：
@Resource
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineRedisPersister;
/**
* 發送訂單狀態轉換事件
* synchronized修飾保證這個方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
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;
}3.4 狀態機存在的問題
1）stateMachine 無法拋出異常，異常會被狀態機給消化掉
問題現象
從 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;
}調試發現：發送事件和監聽事件是一個線程，發送事件的結果是在監聽操作執行完之後才返回
監聽線程：
解決方案：自己保存異常到數據庫或者內存中，進行判斷
也可以通過接口：org.springframework.statemachine.StateMachine##getExtendedState
方法把執行狀態放入這個變量中
public interface ExtendedState {
Map<Object, Object> getVariables();
<T> T get(Object var1, Class<T> var2);
void setExtendedStateChangeListener(ExtendedState.ExtendedStateChangeListener var1);
public interface ExtendedStateChangeListener {
void changed(Object var1, Object var2);
}
}org.springframework.statemachine.support.DefaultExtendedState##getVariables
private final Map<Object, Object> variables;
public DefaultExtendedState() {
this.variables = new ObservableMap(new ConcurrentHashMap(), new DefaultExtendedState.LocalMapChangeListener());
}
public Map<Object, Object> getVariables() {
return this.variables;
}改造監聽狀態：把業務的執行結果進行保存，1 成功，0 失敗
@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;
}
}發送事件改造：如果獲取到業務執行異常，則返回失敗，不進行狀態機持久化 com.zengqingfa.springboot.state.demo.service.impl.OrderServiceImpl##sendEvent
@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;
}代碼優化
- 發送事件只針對了支付，如果是非支付事件呢？
//獲取到監聽的結果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(CommonConstants.payTransition + order.getId());- 監聽設置狀態的代碼有重複代碼，需要進行優化，可使用 aop
try {
//TODO 其他業務
//成功 則爲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;
}常量類：
public interface CommonConstants {
String orderHeader="order";
String payTransition="payTransition";
String deliverTransition="deliverTransition";
String receiveTransition="receiveTransition";
}支付發送事件：com.zengqingfa.springboot.state.demo.service.impl.OrderServiceImpl##pay
@Resource
private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;
@Resource
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineMemPersister;
@Resource
private OrderMapper orderMapper;
/**
* 對訂單進行支付
*
* @param id
* @return
*/
public Order pay(Long id) {
Order order = orderMapper.selectById(id);
log.info("線程名稱：{},嘗試支付，訂單號：{}" ,Thread.currentThread().getName() , id);
if (!sendEvent(OrderStatusChangeEvent.PAYED, order,CommonConstants.payTransition)) {
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,String key){
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(key + order.getId());
//操作完成之後,刪除本次對應的key信息
orderStateMachine.getExtendedState().getVariables().remove(key+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;
}使用 aop 對監聽事件切面，把業務執行結果封裝到狀態機的變量中，註解：
@Retention(RetentionPolicy.RUNTIME)
public @interface LogResult {
/**
*執行的業務key
*
* @return String
*/
String key();
}切面：
@Component
@Aspect
@Slf4j
public class LogResultAspect {
//攔截 LogHistory註解
@Pointcut("@annotation(com.zengqingfa.springboot.state.demo.aop.annotation.LogResult)")
private void logResultPointCut() {
//logResultPointCut 日誌註解切點
}
@Resource
private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;
@Around("logResultPointCut()")
public Object logResultAround(ProceedingJoinPoint pjp) throws Throwable {
//獲取參數
Object[] args = pjp.getArgs();
log.info("參數args:{}", args);
Message message = (Message) args[0];
Order order = (Order) message.getHeaders().get("order");
//獲取方法
Method method = ((MethodSignature) pjp.getSignature()).getMethod();
// 獲取LogHistory註解
LogResult logResult = method.getAnnotation(LogResult.class);
String key = logResult.key();
Object returnVal = null;
try {
//執行方法
returnVal = pjp.proceed();
//如果業務執行正常，則保存信息
//成功 則爲1
orderStateMachine.getExtendedState().getVariables().put(key + order.getId(), 1);
} catch (Throwable e) {
log.error("e:{}", e.getMessage());
//如果業務執行異常，則保存信息
//將異常信息變量信息中，失敗則爲0
orderStateMachine.getExtendedState().getVariables().put(key + order.getId(), 0);
throw e;
}
return returnVal;
}
}監聽類使用註解：
@Component("orderStateListener")
@WithStateMachine(name = "orderStateMachine")
@Slf4j
public class OrderStateListenerImpl {
@Resource
private OrderMapper orderMapper;
@OnTransition(source = "WAIT_PAYMENT", target = "WAIT_DELIVER")
@Transactional(rollbackFor = Exception.class)
@LogResult(key = CommonConstants.payTransition)
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 其他業務
//模擬異常
if (Objects.equals(order.getName(), "A")) {
throw new RuntimeException("執行業務異常");
}
}
@OnTransition(source = "WAIT_DELIVER", target = "WAIT_RECEIVE")
@LogResult(key = CommonConstants.deliverTransition)
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")
@LogResult(key = CommonConstants.receiveTransition)
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 其他業務
}
}如有錯誤之處，還望大家指正。
作者：https://duidaima.com/Group/Topic/JAVA/11942
本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/8lZ1jpvOTb2cd4-IhclaMg