項目終于用上了 Spring 狀態(tài)機，太優(yōu)雅了！
共 68686字，需瀏覽 138分鐘
·
2024-05-23 15:30
1、什么是狀態(tài)機
1.1 什么是狀態(tài)
先來解釋什么是“狀態(tài)”（ State ）?，F(xiàn)實事物是有不同狀態(tài)的，例如一個自動門，就有 open 和 closed 兩種狀態(tài)。我們通常所說的狀態(tài)機是有限狀態(tài)機，也就是被描述的事物的狀態(tài)的數(shù)量是有限個，例如自動門的狀態(tài)就是兩個 open 和 closed 。
狀態(tài)機，也就是 State Machine ，不是指一臺實際機器，而是指一個數(shù)學模型。說白了，一般就是指一張狀態(tài)轉(zhuǎn)換圖。例如，根據(jù)自動門的運行規(guī)則，我們可以抽象出下面這么一個圖。
自動門有兩個狀態(tài)，open 和 closed ，closed 狀態(tài)下，如果讀取開門信號，那么狀態(tài)就會切換為 open 。open 狀態(tài)下如果讀取關門信號，狀態(tài)就會切換為 closed 。
狀態(tài)機的全稱是有限狀態(tài)自動機，自動兩個字也是包含重要含義的。給定一個狀態(tài)機，同時給定它的當前狀態(tài)以及輸入，那么輸出狀態(tài)時可以明確的運算出來的。例如對于自動門，給定初始狀態(tài) closed ，給定輸入“開門”，那么下一個狀態(tài)時可以運算出來的。
這樣狀態(tài)機的基本定義我們就介紹完畢了。重復一下：狀態(tài)機是有限狀態(tài)自動機的簡稱，是現(xiàn)實事物運行規(guī)則抽象而成的一個數(shù)學模型。
1.2 四大概念
下面來給出狀態(tài)機的四大概念。
-
第一個是 State ，狀態(tài)。一個狀態(tài)機至少要包含兩個狀態(tài)。例如上面自動門的例子，有 open 和 closed 兩個狀態(tài)。 -
第二個是 Event ，事件。事件就是執(zhí)行某個操作的觸發(fā)條件或者口令。對于自動門，“按下開門按鈕”就是一個事件。 -
第三個是 Action ，動作。事件發(fā)生以后要執(zhí)行動作。例如事件是“按開門按鈕”，動作是“開門”。編程的時候，一個 Action一般就對應一個函數(shù)。 -
第四個是 Transition ，變換。也就是從一個狀態(tài)變化為另一個狀態(tài)。例如“開門過程”就是一個變換。
1.3 狀態(tài)機
有限狀態(tài)機（Finite-state machine,FSM），又稱有限狀態(tài)自動機，簡稱狀態(tài)機，是表示有限個狀態(tài)以及在這些狀態(tài)之間的轉(zhuǎn)移和動作等行為的數(shù)學模型。
FSM是一種算法思想，簡單而言，有限狀態(tài)機由一組狀態(tài)、一個初始狀態(tài)、輸入和根據(jù)輸入及現(xiàn)有狀態(tài)轉(zhuǎn)換為下一個狀態(tài)的轉(zhuǎn)換函數(shù)組成。
其作用主要是描述對象在它的生命周期內(nèi)所經(jīng)歷的狀態(tài)序列，以及如何響應來自外界的各種事件。
推薦一個開源免費的 Spring Boot 實戰(zhàn)項目：
https://github.com/javastacks/spring-boot-best-practice
2、狀態(tài)機圖
做需求時，需要了解以下六種元素：起始、終止、現(xiàn)態(tài)、次態(tài)（目標狀態(tài)）、動作、條件，我們就可以完成一個狀態(tài)機圖了：
以訂單為例：以從待支付狀態(tài)轉(zhuǎn)換為待發(fā)貨狀態(tài)為例
-
①現(xiàn)態(tài)：是指當前所處的狀態(tài)。待支付 -
②條件：又稱為“事件”，當一個條件被滿足，將會觸發(fā)一個動作，或者執(zhí)行一次狀態(tài)的遷移。支付事件 -
③動作：條件滿足后執(zhí)行的動作。動作執(zhí)行完畢后，可以遷移到新的狀態(tài)，也可以仍舊保持原狀態(tài)。動作不是必需的，當條件滿足后，也可以不執(zhí)行任何動作，直接遷移到新狀態(tài)。狀態(tài)轉(zhuǎn)換為待發(fā)貨 -
④次態(tài)：條件滿足后要遷往的新狀態(tài)?！按螒B(tài)”是相對于“現(xiàn)態(tài)”而言的，“次態(tài)”一旦被激活，就轉(zhuǎn)變成新的“現(xiàn)態(tài)”了。待發(fā)貨 注意事項
1、避免把某個“程序動作”當作是一種“狀態(tài)”來處理。那么如何區(qū)分“動作”和“狀態(tài)”？“動作”是不穩(wěn)定的，即使沒有條件的觸發(fā)，“動作”一旦執(zhí)行完畢就結(jié)束了；而“狀態(tài)”是相對穩(wěn)定的，如果沒有外部條件的觸發(fā)，一個狀態(tài)會一直持續(xù)下去。
2、狀態(tài)劃分時漏掉一些狀態(tài)，導致跳轉(zhuǎn)邏輯不完整。所以在設計狀態(tài)機時，我們需要反復的查看設計的狀態(tài)圖或者狀態(tài)表，最終達到一種牢不可破的設計方案。
3、spring statemachine
3.1 狀態(tài)機spring statemachine 概述
Spring Statemachine是應用程序開發(fā)人員在Spring應用程序中使用狀態(tài)機概念的框架
Spring Statemachine旨在提供以下功能：
-
易于使用的扁平單級狀態(tài)機，用于簡單的使用案例。 -
分層狀態(tài)機結(jié)構(gòu)，以簡化復雜的狀態(tài)配置。 -
狀態(tài)機區(qū)域提供更復雜的狀態(tài)配置。 -
使用觸發(fā)器，轉(zhuǎn)換，警衛(wèi)和操作。 -
鍵入安全配置適配器。 -
生成器模式，用于在Spring Application上下文之外使用的簡單實例化通常用例的食譜 -
基于Zookeeper的分布式狀態(tài)機 -
狀態(tài)機事件監(jiān)聽器。 -
UML Eclipse Papyrus建模。 -
將計算機配置存儲在永久存儲中。 -
Spring IOC集成將bean與狀態(tài)機關聯(lián)起來。
狀態(tài)機功能強大，因為行為始終保證一致，使調(diào)試相對容易。這是因為操作規(guī)則是在機器啟動時寫成的。這個想法是你的應用程序可能存在于有限數(shù)量的狀態(tài)中，某些預定義的觸發(fā)器可以將你的應用程序從一個狀態(tài)轉(zhuǎn)移到另一個狀態(tài)。此類觸發(fā)器可以基于事件或計時器。
在應用程序之外定義高級邏輯然后依靠狀態(tài)機來管理狀態(tài)要容易得多。您可以通過發(fā)送事件，偵聽更改或僅請求當前狀態(tài)來與狀態(tài)機進行交互。
官網(wǎng)：spring.io/projects/sp…
3.2 快速開始
Spring Boot 基礎就不介紹了，推薦看這個實戰(zhàn)項目：
https://github.com/javastacks/spring-boot-best-practice
以訂單狀態(tài)扭轉(zhuǎn)的例子為例：
表結(jié)構(gòu)設計如下：
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 '訂單狀態(tài)',
`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 '創(chuàng)建時間',
`update_time` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' COMMENT '更新時間',
`create_user_code` varchar(32) COLLATE utf8mb4_bin DEFAULT NULL COMMENT '創(chuàng)建人',
`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持久化狀態(tài)機 -->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-redis</artifactId>
<version>1.2.9.RELEASE</version>
</dependency>
<!--狀態(tài)機-->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-starter</artifactId>
<version>2.0.1.RELEASE</version>
</dependency>
2）定義狀態(tài)機狀態(tài)和事件
狀態(tài)枚舉：
public enum OrderStatus {
// 待支付，待發(fā)貨，待收貨，已完成
WAIT_PAYMENT(1, "待支付"),
WAIT_DELIVER(2, "待發(fā)貨"),
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 {
// 支付，發(fā)貨，確認收貨
PAYED, DELIVERY, RECEIVED;
}
3）定義狀態(tài)機規(guī)則和配置狀態(tài)機
@Configuration
@EnableStateMachine(name = "orderStateMachine")
public class OrderStateMachineConfig extends StateMachineConfigurerAdapter<OrderStatus, OrderStatusChangeEvent> {
/**
* 配置狀態(tài)
*
* @param states
* @throws Exception
*/
public void configure(StateMachineStateConfigurer<OrderStatus, OrderStatusChangeEvent> states) throws Exception {
states
.withStates()
.initial(OrderStatus.WAIT_PAYMENT)
.states(EnumSet.allOf(OrderStatus.class));
}
/**
* 配置狀態(tài)轉(zhuǎn)換事件關系
*
* @param transitions
* @throws Exception
*/
public void configure(StateMachineTransitionConfigurer<OrderStatus, OrderStatusChangeEvent> transitions) throws Exception {
transitions
//支付事件:待支付-》待發(fā)貨
.withExternal().source(OrderStatus.WAIT_PAYMENT).target(OrderStatus.WAIT_DELIVER).event(OrderStatusChangeEvent.PAYED)
.and()
//發(fā)貨事件:待發(fā)貨-》待收貨
.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> {
/**
* 持久化到內(nèi)存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("持久化狀態(tài)機,context:{},contextObj:{}", JSON.toJSONString(context), JSON.toJSONString(contextObj));
map.put(contextObj, context);
}
@Override
public StateMachineContext read(Object contextObj) throws Exception {
log.info("獲取狀態(tài)機,contextObj:{}", JSON.toJSONString(contextObj));
StateMachineContext stateMachineContext = (StateMachineContext) map.get(contextObj);
log.info("獲取狀態(tài)機結(jié)果,stateMachineContext:{}", JSON.toJSONString(stateMachineContext));
return stateMachineContext;
}
private Map map = new HashMap();
});
}
@Resource
private RedisConnectionFactory redisConnectionFactory;
/**
* 持久化到redis中，在分布式系統(tǒng)中使用
*
* @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）業(yè)務系統(tǒng)
controller：
@RestController
@RequestMapping("/order")
public class OrderController {
@Resource
private OrderService orderService;
/**
* 根據(jù)id查詢訂單
*
* @return
*/
@RequestMapping("/getById")
public Order getById(@RequestParam("id") Long id) {
//根據(jù)id查詢訂單
Order order = orderService.getById(id);
return order;
}
/**
* 創(chuàng)建訂單
*
* @return
*/
@RequestMapping("/create")
public String create(@RequestBody Order order) {
//創(chuàng)建訂單
orderService.create(order);
return "sucess";
}
/**
* 對訂單進行支付
*
* @param id
* @return
*/
@RequestMapping("/pay")
public String pay(@RequestParam("id") Long id) {
//對訂單進行支付
orderService.pay(id);
return "success";
}
/**
* 對訂單進行發(fā)貨
*
* @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;
/**
* 創(chuàng)建訂單
*
* @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("線程名稱：{},支付失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
throw new RuntimeException("支付失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 對訂單進行發(fā)貨
*
* @param id
* @return
*/
public Order deliver(Long id) {
Order order = orderMapper.selectById(id);
log.info("線程名稱：{},嘗試發(fā)貨，訂單號：{}" ,Thread.currentThread().getName() , id);
if (!sendEvent(OrderStatusChangeEvent.DELIVERY, order)) {
log.error("線程名稱：{},發(fā)貨失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
throw new RuntimeException("發(fā)貨失敗, 訂單狀態(tài)異常");
}
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("線程名稱：{},收貨失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
throw new RuntimeException("收貨失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動狀態(tài)機
orderStateMachine.start();
//嘗試恢復狀態(tài)機狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
//持久化狀態(tài)機狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}
}
監(jiān)聽狀態(tài)的變化：
@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("支付，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
}
@OnTransition(source = "WAIT_DELIVER", target = "WAIT_RECEIVE")
public void deliverTransition(Message<OrderStatusChangeEvent> message) {
Order order = (Order) message.getHeaders().get("order");
log.info("發(fā)貨，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_RECEIVE.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
}
@OnTransition(source = "WAIT_RECEIVE", target = "FINISH")
public void receiveTransition(Message<OrderStatusChangeEvent> message) {
Order order = (Order) message.getHeaders().get("order");
log.info("確認收貨，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.FINISH.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
}
}
3.3 測試驗證
1）驗證業(yè)務
-
新增一個訂單
http://localhost:8084/order/create
-
對訂單進行支付
http://localhost:8084/order/pay?id=2
-
對訂單進行發(fā)貨
http://localhost:8084/order/deliver?id=2
-
對訂單進行確認收貨
http://localhost:8084/order/receive?id=2
正常流程結(jié)束。如果對一個訂單進行支付了，再次進行支付，則會報錯：http://localhost:8084/order/pay?id=2
報錯如下：
2）驗證持久化
內(nèi)存
使用內(nèi)存持久化類持久化：
@Resource
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineMemPersister;
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動狀態(tài)機
orderStateMachine.start();
//嘗試恢復狀態(tài)機狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
//持久化狀態(tài)機狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}
redis持久化
引入依賴：
<!-- redis持久化狀態(tài)機 -->
<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;
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動狀態(tài)機
orderStateMachine.start();
//嘗試恢復狀態(tài)機狀態(tài)
stateMachineRedisPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
//持久化狀態(tài)機狀態(tài)
stateMachineRedisPersister.persist(orderStateMachine, String.valueOf(order.getId()));
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}
3.4 狀態(tài)機存在的問題
1）stateMachine無法拋出異常，異常會被狀態(tài)機給消化掉
問題現(xiàn)象
從orderStateMachine.sendEvent(message);獲取的結(jié)果無法感知到。無論執(zhí)行正常還是拋出異常，都返回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("支付，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
try {
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
//模擬異常
if(Objects.equals(order.getName(),"A")){
throw new RuntimeException("執(zhí)行業(yè)務異常");
}
} catch (Exception e) {
//如果出現(xiàn)異常，記錄異常信息，拋出異常信息進行回滾
log.error("payTransition 出現(xiàn)異常：{}",e);
throw e;
}
}
監(jiān)聽事件拋出異常，在發(fā)送事件中無法感知：
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動狀態(tài)機
orderStateMachine.start();
//嘗試恢復狀態(tài)機狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
//事件執(zhí)行異常了，依然返回true，無法感知異常
result = orderStateMachine.sendEvent(message);
if(result){
//持久化狀態(tài)機狀態(tài)，如果根據(jù)true持久化，則會出現(xiàn)問題
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
}
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}
調(diào)試發(fā)現(xiàn)：發(fā)送事件和監(jiān)聽事件是一個線程，發(fā)送事件的結(jié)果是在監(jiān)聽操作執(zhí)行完之后才返回
監(jiān)聽線程：
解決方案：自己保存異常到數(shù)據(jù)庫或者內(nèi)存中，進行判斷
也可以通過接口：org.springframework.statemachine.StateMachine##getExtendedState
方法把執(zhí)行狀態(tài)放入這個變量中
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;
}
改造監(jiān)聽狀態(tài)：把業(yè)務的執(zhí)行結(jié)果進行保存，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("支付，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
try {
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
//模擬異常
if(Objects.equals(order.getName(),"A")){
throw new RuntimeException("執(zhí)行業(yè)務異常");
}
//成功 則為1
orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(),1);
} catch (Exception e) {
//如果出現(xiàn)異常，則進行回滾
log.error("payTransition 出現(xiàn)異常：{}",e);
//將異常信息變量信息中，失敗則為0
orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(), 0);
throw e;
}
}
發(fā)送事件改造：如果獲取到業(yè)務執(zhí)行異常，則返回失敗，不進行狀態(tài)機持久化 com.zengqingfa.springboot.state.demo.service.impl.OrderServiceImpl##sendEvent
@Resource
private StateMachine<OrderStatus, OrderStatusChangeEvent> orderStateMachine;
@Resource
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineMemPersister;
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order){
boolean result = false;
try {
//啟動狀態(tài)機
orderStateMachine.start();
//嘗試恢復狀態(tài)機狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
if(!result){
return false;
}
//獲取到監(jiān)聽的結(jié)果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(CommonConstants.payTransition + order.getId());
//操作完成之后,刪除本次對應的key信息
orderStateMachine.getExtendedState().getVariables().remove(CommonConstants.payTransition+order.getId());
//如果事務執(zhí)行成功，則持久化狀態(tài)機
if(Objects.equals(1,Integer.valueOf(o))){
//持久化狀態(tài)機狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
}else {
//訂單執(zhí)行業(yè)務異常
return false;
}
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}
代碼優(yōu)化
-
發(fā)送事件只針對了支付，如果是非支付事件呢？
//獲取到監(jiān)聽的結(jié)果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(CommonConstants.payTransition + order.getId());
-
監(jiān)聽設置狀態(tài)的代碼有重復代碼，需要進行優(yōu)化，可使用aop
try {
//TODO 其他業(yè)務
//成功 則為1
orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(),1);
} catch (Exception e) {
//如果出現(xiàn)異常，則進行回滾
log.error("payTransition 出現(xiàn)異常：{}",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";
}
支付發(fā)送事件：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("線程名稱：{},支付失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
throw new RuntimeException("支付失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order,String key){
boolean result = false;
try {
//啟動狀態(tài)機
orderStateMachine.start();
//嘗試恢復狀態(tài)機狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
if(!result){
return false;
}
//獲取到監(jiān)聽的結(jié)果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(key + order.getId());
//操作完成之后,刪除本次對應的key信息
orderStateMachine.getExtendedState().getVariables().remove(key+order.getId());
//如果事務執(zhí)行成功，則持久化狀態(tài)機
if(Objects.equals(1,Integer.valueOf(o))){
//持久化狀態(tài)機狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
}else {
//訂單執(zhí)行業(yè)務異常
return false;
}
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}
使用aop對監(jiān)聽事件切面，把業(yè)務執(zhí)行結(jié)果封裝到狀態(tài)機的變量中，注解：
@Retention(RetentionPolicy.RUNTIME)
public @interface LogResult {
/**
*執(zhí)行的業(yè)務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 {
//獲取參數(shù)
Object[] args = pjp.getArgs();
log.info("參數(shù)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 {
//執(zhí)行方法
returnVal = pjp.proceed();
//如果業(yè)務執(zhí)行正常，則保存信息
//成功 則為1
orderStateMachine.getExtendedState().getVariables().put(key + order.getId(), 1);
} catch (Throwable e) {
log.error("e:{}", e.getMessage());
//如果業(yè)務執(zhí)行異常，則保存信息
//將異常信息變量信息中，失敗則為0
orderStateMachine.getExtendedState().getVariables().put(key + order.getId(), 0);
throw e;
}
return returnVal;
}
}
監(jiān)聽類使用注解：
@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("支付，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
//模擬異常
if (Objects.equals(order.getName(), "A")) {
throw new RuntimeException("執(zhí)行業(yè)務異常");
}
}
@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("發(fā)貨，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_RECEIVE.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
}
@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("確認收貨，狀態(tài)機反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.FINISH.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務
}
}
評論
圖片
表情