實(shí)戰(zhàn)互聯(lián)網(wǎng)大廠基于Spring狀態(tài)機(jī)驅(qū)動(dòng)訂單支付狀態(tài)流轉(zhuǎn)，可以直接封神！
先來解釋什么是“狀態(tài)”（ State ）?，F(xiàn)實(shí)事物是有不同狀態(tài)的，例如一個(gè)自動(dòng)門，就有 open 和 closed 兩種狀態(tài)。我們通常所說的狀態(tài)機(jī)是有限狀態(tài)機(jī)，也就是被描述的事物的狀態(tài)的數(shù)量是有限個(gè)，例如自動(dòng)門的狀態(tài)就是兩個(gè) open 和 closed 。
最近一直有小伙伴問我大廠是如何設(shè)計(jì)和實(shí)現(xiàn)訂單支付狀態(tài)流轉(zhuǎn)的，一個(gè)典型的方案就是基于Spring狀態(tài)機(jī)驅(qū)動(dòng)訂單支付狀態(tài)流轉(zhuǎn)。今天，就給大家詳細(xì)介紹下什么是狀態(tài)機(jī)、Spring狀態(tài)機(jī)是如何實(shí)現(xiàn)的，以及以Spring狀態(tài)機(jī)為例，實(shí)現(xiàn)狀態(tài)機(jī)驅(qū)動(dòng)訂單支付狀態(tài)流轉(zhuǎn)的詳細(xì)案例代碼。好了，開始今天的內(nèi)容。
1、什么是狀態(tài)機(jī)
1.1 什么是狀態(tài)
先來解釋什么是“狀態(tài)”（ State ）?，F(xiàn)實(shí)事物是有不同狀態(tài)的，例如一個(gè)自動(dòng)門，就有 open 和 closed 兩種狀態(tài)。我們通常所說的狀態(tài)機(jī)是有限狀態(tài)機(jī)，也就是被描述的事物的狀態(tài)的數(shù)量是有限個(gè)，例如自動(dòng)門的狀態(tài)就是兩個(gè) open 和 closed 。
圖片
狀態(tài)機(jī)，也就是 State Machine ，不是指一臺(tái)實(shí)際機(jī)器，而是指一個(gè)數(shù)學(xué)模型。說白了，一般就是指一張狀態(tài)轉(zhuǎn)換圖。例如，根據(jù)自動(dòng)門的運(yùn)行規(guī)則，我們可以抽象出下面這么一個(gè)圖。
自動(dòng)門有兩個(gè)狀態(tài)，open 和 closed ，closed 狀態(tài)下，如果讀取開門信號(hào)，那么狀態(tài)就會(huì)切換為 open 。open 狀態(tài)下如果讀取關(guān)門信號(hào)，狀態(tài)就會(huì)切換為 closed 。
狀態(tài)機(jī)的全稱是有限狀態(tài)自動(dòng)機(jī)，自動(dòng)兩個(gè)字也是包含重要含義的。給定一個(gè)狀態(tài)機(jī)，同時(shí)給定它的當(dāng)前狀態(tài)以及輸入，那么輸出狀態(tài)時(shí)可以明確的運(yùn)算出來的。例如對(duì)于自動(dòng)門，給定初始狀態(tài) closed ，給定輸入“開門”，那么下一個(gè)狀態(tài)時(shí)可以運(yùn)算出來的。
這樣狀態(tài)機(jī)的基本定義我們就介紹完畢了。重復(fù)一下：狀態(tài)機(jī)是有限狀態(tài)自動(dòng)機(jī)的簡稱，是現(xiàn)實(shí)事物運(yùn)行規(guī)則抽象而成的一個(gè)數(shù)學(xué)模型。
1.2 四大概念
下面來給出狀態(tài)機(jī)的四大概念。
- 第一個(gè)是 State ，狀態(tài)。一個(gè)狀態(tài)機(jī)至少要包含兩個(gè)狀態(tài)。例如上面自動(dòng)門的例子，有 open 和 closed 兩個(gè)狀態(tài)。
- 第二個(gè)是 Event ，事件。事件就是執(zhí)行某個(gè)操作的觸發(fā)條件或者口令。對(duì)于自動(dòng)門，“按下開門按鈕”就是一個(gè)事件。
- 第三個(gè)是 Action ，動(dòng)作。事件發(fā)生以后要執(zhí)行動(dòng)作。例如事件是“按開門按鈕”，動(dòng)作是“開門”。編程的時(shí)候，一個(gè) Action一般就對(duì)應(yīng)一個(gè)函數(shù)。
- 第四個(gè)是 Transition ，變換。也就是從一個(gè)狀態(tài)變化為另一個(gè)狀態(tài)。例如“開門過程”就是一個(gè)變換。
1.3 狀態(tài)機(jī)
有限狀態(tài)機(jī)（Finite-state machine,FSM），又稱有限狀態(tài)自動(dòng)機(jī)，簡稱狀態(tài)機(jī)，是表示有限個(gè)狀態(tài)以及在這些狀態(tài)之間的轉(zhuǎn)移和動(dòng)作等行為的數(shù)學(xué)模型。
FSM是一種算法思想，簡單而言，有限狀態(tài)機(jī)由一組狀態(tài)、一個(gè)初始狀態(tài)、輸入和根據(jù)輸入及現(xiàn)有狀態(tài)轉(zhuǎn)換為下一個(gè)狀態(tài)的轉(zhuǎn)換函數(shù)組成。
其作用主要是描述對(duì)象在它的生命周期內(nèi)所經(jīng)歷的狀態(tài)序列，以及如何響應(yīng)來自外界的各種事件。
2、狀態(tài)機(jī)圖
做需求時(shí)，需要了解以下六種元素：起始、終止、現(xiàn)態(tài)、次態(tài)（目標(biāo)狀態(tài)）、動(dòng)作、條件，我們就可以完成一個(gè)狀態(tài)機(jī)圖了：
以訂單為例：以從待支付狀態(tài)轉(zhuǎn)換為待發(fā)貨狀態(tài)為例
圖片
①現(xiàn)態(tài)：是指當(dāng)前所處的狀態(tài)。待支付
②條件：又稱為“事件”，當(dāng)一個(gè)條件被滿足，將會(huì)觸發(fā)一個(gè)動(dòng)作，或者執(zhí)行一次狀態(tài)的遷移。支付事件
③動(dòng)作：條件滿足后執(zhí)行的動(dòng)作。動(dòng)作執(zhí)行完畢后，可以遷移到新的狀態(tài)，也可以仍舊保持原狀態(tài)。動(dòng)作不是必需的，當(dāng)條件滿足后，也可以不執(zhí)行任何動(dòng)作，直接遷移到新狀態(tài)。狀態(tài)轉(zhuǎn)換為待發(fā)貨
④次態(tài)：條件滿足后要遷往的新狀態(tài)?！按螒B(tài)”是相對(duì)于“現(xiàn)態(tài)”而言的，“次態(tài)”一旦被激活，就轉(zhuǎn)變成新的“現(xiàn)態(tài)”了。待發(fā)貨 注意事項(xiàng)
- 避免把某個(gè)“程序動(dòng)作”當(dāng)作是一種“狀態(tài)”來處理。那么如何區(qū)分“動(dòng)作”和“狀態(tài)”？“動(dòng)作”是不穩(wěn)定的，即使沒有條件的觸發(fā)，“動(dòng)作”一旦執(zhí)行完畢就結(jié)束了；而“狀態(tài)”是相對(duì)穩(wěn)定的，如果沒有外部條件的觸發(fā)，一個(gè)狀態(tài)會(huì)一直持續(xù)下去。
- 狀態(tài)劃分時(shí)漏掉一些狀態(tài)，導(dǎo)致跳轉(zhuǎn)邏輯不完整。所以在設(shè)計(jì)狀態(tài)機(jī)時(shí)，我們需要反復(fù)的查看設(shè)計(jì)的狀態(tài)圖或者狀態(tài)表，最終達(dá)到一種牢不可破的設(shè)計(jì)方案。
3、spring statemachine
3.1 狀態(tài)機(jī)spring statemachine 概述
Spring Statemachine是應(yīng)用程序開發(fā)人員在Spring應(yīng)用程序中使用狀態(tài)機(jī)概念的框架
Spring Statemachine旨在提供以下功能：
- 易于使用的扁平單級(jí)狀態(tài)機(jī)，用于簡單的使用案例。
- 分層狀態(tài)機(jī)結(jié)構(gòu)，以簡化復(fù)雜的狀態(tài)配置。
- 狀態(tài)機(jī)區(qū)域提供更復(fù)雜的狀態(tài)配置。
- 使用觸發(fā)器，轉(zhuǎn)換，警衛(wèi)和操作。
- 鍵入安全配置適配器。
- 生成器模式，用于在Spring Application上下文之外使用的簡單實(shí)例化通常用例的食譜
- 基于Zookeeper的分布式狀態(tài)機(jī)
- 狀態(tài)機(jī)事件監(jiān)聽器。
- UML Eclipse Papyrus建模。
- 將計(jì)算機(jī)配置存儲(chǔ)在永久存儲(chǔ)中。
- Spring IOC集成將bean與狀態(tài)機(jī)關(guān)聯(lián)起來。
狀態(tài)機(jī)功能強(qiáng)大，因?yàn)樾袨槭冀K保證一致，使調(diào)試相對(duì)容易。這是因?yàn)椴僮饕?guī)則是在機(jī)器啟動(dòng)時(shí)寫成的。這個(gè)想法是你的應(yīng)用程序可能存在于有限數(shù)量的狀態(tài)中，某些預(yù)定義的觸發(fā)器可以將你的應(yīng)用程序從一個(gè)狀態(tài)轉(zhuǎn)移到另一個(gè)狀態(tài)。此類觸發(fā)器可以基于事件或計(jì)時(shí)器。
在應(yīng)用程序之外定義高級(jí)邏輯然后依靠狀態(tài)機(jī)來管理狀態(tài)要容易得多。您可以通過發(fā)送事件，偵聽更改或僅請(qǐng)求當(dāng)前狀態(tài)來與狀態(tài)機(jī)進(jìn)行交互。
3.2 快速開始
以訂單狀態(tài)扭轉(zhuǎn)的例子為例：
表結(jié)構(gòu)設(shè)計(jì)如下：
CREATE TABLE`tb_order` (
`id`bigint(20) unsignedNOTNULL AUTO_INCREMENT COMMENT'主鍵ID',
`order_code`varchar(128) COLLATE utf8mb4_bin DEFAULTNULLCOMMENT'訂單編碼',
`status`smallint(3) DEFAULTNULLCOMMENT'訂單狀態(tài)',
`name`varchar(64) COLLATE utf8mb4_bin DEFAULTNULLCOMMENT'訂單名稱',
`price`decimal(12,2) DEFAULTNULLCOMMENT'價(jià)格',
`delete_flag`tinyint(2) NOTNULLDEFAULT'0'COMMENT'刪除標(biāo)記，0未刪除 1已刪除',
`create_time`timestampNOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMPCOMMENT'創(chuàng)建時(shí)間',
`update_time`timestampNOTNULLDEFAULT'0000-00-00 00:00:00'COMMENT'更新時(shí)間',
`create_user_code`varchar(32) COLLATE utf8mb4_bin DEFAULTNULLCOMMENT'創(chuàng)建人',
`update_user_code`varchar(32) COLLATE utf8mb4_bin DEFAULTNULLCOMMENT'更新人',
`version`int(11) NOTNULLDEFAULT'0'COMMENT'版本號(hào)',
`remark`varchar(64) COLLATE utf8mb4_bin DEFAULTNULLCOMMENT'備注',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=6DEFAULTCHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='訂單表';
/*Data for the table `tb_order` */
insertinto`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)機(jī) -->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-redis</artifactId>
<version>1.2.9.RELEASE</version>
</dependency>
<!--狀態(tài)機(jī)-->
<dependency>
<groupId>org.springframework.statemachine</groupId>
<artifactId>spring-statemachine-starter</artifactId>
<version>2.0.1.RELEASE</version>
</dependency>2）定義狀態(tài)機(jī)狀態(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;
}
}
thrownew RuntimeException("enum not exists.");
}
}事件：
public enum OrderStatusChangeEvent {
// 支付，發(fā)貨，確認(rèn)收貨
PAYED, DELIVERY, RECEIVED;
}3）定義狀態(tài)機(jī)規(guī)則和配置狀態(tài)機(jī)
@Configuration
@EnableStateMachine(name = "orderStateMachine")
publicclass 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)換事件關(guā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
publicclass Persist<E, S> {
/**
* 持久化到內(nèi)存map中
*
* @return
*/
@Bean(name = "stateMachineMemPersister")
public static StateMachinePersister getPersister() {
returnnew DefaultStateMachinePersister(new StateMachinePersist() {
@Override
public void write(StateMachineContext context, Object contextObj) throws Exception {
log.info("持久化狀態(tài)機(jī),context:{},contextObj:{}", JSON.toJSONString(context), JSON.toJSONString(contextObj));
map.put(contextObj, context);
}
@Override
public StateMachineContext read(Object contextObj) throws Exception {
log.info("獲取狀態(tài)機(jī),contextObj:{}", JSON.toJSONString(contextObj));
StateMachineContext stateMachineContext = (StateMachineContext) map.get(contextObj);
log.info("獲取狀態(tài)機(jī)結(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);
returnnew RedisStateMachinePersister<>(p);
}
}4）業(yè)務(wù)系統(tǒng)
controller：
@RestController
@RequestMapping("/order")
publicclass 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";
}
/**
* 對(duì)訂單進(jìn)行支付
*
* @param id
* @return
*/
@RequestMapping("/pay")
public String pay(@RequestParam("id") Long id) {
//對(duì)訂單進(jìn)行支付
orderService.pay(id);
return"success";
}
/**
* 對(duì)訂單進(jìn)行發(fā)貨
*
* @param id
* @return
*/
@RequestMapping("/deliver")
public String deliver(@RequestParam("id") Long id) {
//對(duì)訂單進(jìn)行確認(rèn)收貨
orderService.deliver(id);
return"success";
}
/**
* 對(duì)訂單進(jìn)行確認(rèn)收貨
*
* @param id
* @return
*/
@RequestMapping("/receive")
public String receive(@RequestParam("id") Long id) {
//對(duì)訂單進(jìn)行確認(rèn)收貨
orderService.receive(id);
return"success";
}
}servie：
@Service("orderService")
@Slf4j
publicclass 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;
}
/**
* 對(duì)訂單進(jìn)行支付
*
* @param id
* @return
*/
public Order pay(Long id) {
Order order = orderMapper.selectById(id);
log.info("線程名稱：{},嘗試支付，訂單號(hào)：{}" ,Thread.currentThread().getName() , id);
if (!sendEvent(OrderStatusChangeEvent.PAYED, order)) {
log.error("線程名稱：{},支付失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
thrownew RuntimeException("支付失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 對(duì)訂單進(jìn)行發(fā)貨
*
* @param id
* @return
*/
public Order deliver(Long id) {
Order order = orderMapper.selectById(id);
log.info("線程名稱：{},嘗試發(fā)貨，訂單號(hào)：{}" ,Thread.currentThread().getName() , id);
if (!sendEvent(OrderStatusChangeEvent.DELIVERY, order)) {
log.error("線程名稱：{},發(fā)貨失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
thrownew RuntimeException("發(fā)貨失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 對(duì)訂單進(jìn)行確認(rèn)收貨
*
* @param id
* @return
*/
public Order receive(Long id) {
Order order = orderMapper.selectById(id);
log.info("線程名稱：{},嘗試收貨，訂單號(hào)：{}" ,Thread.currentThread().getName() , id);
if (!sendEvent(OrderStatusChangeEvent.RECEIVED, order)) {
log.error("線程名稱：{},收貨失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
thrownew RuntimeException("收貨失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個(gè)方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動(dòng)狀態(tài)機(jī)
orderStateMachine.start();
//嘗試恢復(fù)狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
//持久化狀態(tài)機(jī)狀態(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
publicclass 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)機(jī)反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
}
@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)機(jī)反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_RECEIVE.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
}
@OnTransition(source = "WAIT_RECEIVE", target = "FINISH")
public void receiveTransition(Message<OrderStatusChangeEvent> message) {
Order order = (Order) message.getHeaders().get("order");
log.info("確認(rèn)收貨，狀態(tài)機(jī)反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.FINISH.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
}
}3.3 測(cè)試驗(yàn)證
1）驗(yàn)證業(yè)務(wù)
- 新增一個(gè)訂單http://localhost:8084/order/create
- 對(duì)訂單進(jìn)行支付http://localhost:8084/order/pay?id=2
- 對(duì)訂單進(jìn)行發(fā)貨http://localhost:8084/order/deliver?id=2
- 對(duì)訂單進(jìn)行確認(rèn)收貨http://localhost:8084/order/receive?id=2
正常流程結(jié)束。如果對(duì)一個(gè)訂單進(jìn)行支付了，再次進(jìn)行支付，則會(huì)報(bào)錯(cuò)：http://localhost:8084/order/pay?id=2
報(bào)錯(cuò)如下：
圖片
2）驗(yàn)證持久化
使用內(nèi)存持久化類持久化：
@Resource
private StateMachinePersister<OrderStatus, OrderStatusChangeEvent, String> stateMachineMemPersister;
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個(gè)方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動(dòng)狀態(tài)機(jī)
orderStateMachine.start();
//嘗試恢復(fù)狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
//持久化狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}redis持久化
引入依賴：
<!-- redis持久化狀態(tài)機(jī) -->
<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修飾保證這個(gè)方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動(dòng)狀態(tài)機(jī)
orderStateMachine.start();
//嘗試恢復(fù)狀態(tài)機(jī)狀態(tài)
stateMachineRedisPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
//持久化狀態(tài)機(jī)狀態(tài)
stateMachineRedisPersister.persist(orderStateMachine, String.valueOf(order.getId()));
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}3.4 狀態(tài)機(jī)存在的問題
1）stateMachine無法拋出異常，異常會(huì)被狀態(tài)機(jī)給消化掉
問題現(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)機(jī)反饋信息：{}", message.getHeaders().toString());
try {
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
//模擬異常
if(Objects.equals(order.getName(),"A")){
thrownew RuntimeException("執(zhí)行業(yè)務(wù)異常");
}
} catch (Exception e) {
//如果出現(xiàn)異常，記錄異常信息，拋出異常信息進(jìn)行回滾
log.error("payTransition 出現(xiàn)異常：{}",e);
throw e;
}
}監(jiān)聽事件拋出異常，在發(fā)送事件中無法感知：
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order) {
boolean result = false;
try {
//啟動(dòng)狀態(tài)機(jī)
orderStateMachine.start();
//嘗試恢復(fù)狀態(tài)機(jī)狀態(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)機(jī)狀態(tài)，如果根據(jù)true持久化，則會(huì)出現(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)聽事件是一個(gè)線程，發(fā)送事件的結(jié)果是在監(jiān)聽操作執(zhí)行完之后才返回
監(jiān)聽線程：
解決方案：自己保存異常到數(shù)據(jù)庫或者內(nèi)存中，進(jìn)行判斷
也可以通過接口：org.springframework.statemachine.StateMachine##getExtendedState
方法把執(zhí)行狀態(tài)放入這個(gè)變量中
public interface ExtendedState {
Map<Object, Object> getVariables();
<T> T get(Object var1, Class<T> var2);
void setExtendedStateChangeListener(ExtendedState.ExtendedStateChangeListener var1);
publicinterface ExtendedStateChangeListener {
void changed(Object var1, Object var2);
}
}
org.springframework.statemachine.support.DefaultExtendedState##getVariables
privatefinal Map<Object, Object> variables;
public DefaultExtendedState() {
this.variables = new ObservableMap(new ConcurrentHashMap(), new DefaultExtendedState.LocalMapChangeListener());
}
public Map<Object, Object> getVariables() {
returnthis.variables;
}改造監(jiān)聽狀態(tài)：把業(yè)務(wù)的執(zhí)行結(jié)果進(jìn)行保存，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)機(jī)反饋信息：{}", message.getHeaders().toString());
try {
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
//模擬異常
if(Objects.equals(order.getName(),"A")){
thrownew RuntimeException("執(zhí)行業(yè)務(wù)異常");
}
//成功 則為1
orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(),1);
} catch (Exception e) {
//如果出現(xiàn)異常，則進(jìn)行回滾
log.error("payTransition 出現(xiàn)異常：{}",e);
//將異常信息變量信息中，失敗則為0
orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(), 0);
throw e;
}
}發(fā)送事件改造：如果獲取到業(yè)務(wù)執(zhí)行異常，則返回失敗，不進(jìn)行狀態(tài)機(jī)持久化 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修飾保證這個(gè)方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order){
boolean result = false;
try {
//啟動(dòng)狀態(tài)機(jī)
orderStateMachine.start();
//嘗試恢復(fù)狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
if(!result){
returnfalse;
}
//獲取到監(jiān)聽的結(jié)果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(CommonConstants.payTransition + order.getId());
//操作完成之后,刪除本次對(duì)應(yīng)的key信息
orderStateMachine.getExtendedState().getVariables().remove(CommonConstants.payTransition+order.getId());
//如果事務(wù)執(zhí)行成功，則持久化狀態(tài)機(jī)
if(Objects.equals(1,Integer.valueOf(o))){
//持久化狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
}else {
//訂單執(zhí)行業(yè)務(wù)異常
returnfalse;
}
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}代碼優(yōu)化
- 發(fā)送事件只針對(duì)了支付，如果是非支付事件呢？
//獲取到監(jiān)聽的結(jié)果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(CommonConstants.payTransition + order.getId());- 監(jiān)聽設(shè)置狀態(tài)的代碼有重復(fù)代碼，需要進(jìn)行優(yōu)化，可使用aop
try {
//TODO 其他業(yè)務(wù)
//成功 則為1
orderStateMachine.getExtendedState().getVariables().put(CommonConstants.payTransition+order.getId(),1);
} catch (Exception e) {
//如果出現(xiàn)異常，則進(jì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 payTransitinotallow="payTransition";
String deliverTransitinotallow="deliverTransition";
String receiveTransitinotallow="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;
/**
* 對(duì)訂單進(jìn)行支付
*
* @param id
* @return
*/
public Order pay(Long id) {
Order order = orderMapper.selectById(id);
log.info("線程名稱：{},嘗試支付，訂單號(hào)：{}" ,Thread.currentThread().getName() , id);
if (!sendEvent(OrderStatusChangeEvent.PAYED, order,CommonConstants.payTransition)) {
log.error("線程名稱：{},支付失敗, 狀態(tài)異常，訂單信息：{}", Thread.currentThread().getName(), order);
thrownew RuntimeException("支付失敗, 訂單狀態(tài)異常");
}
return order;
}
/**
* 發(fā)送訂單狀態(tài)轉(zhuǎn)換事件
* synchronized修飾保證這個(gè)方法是線程安全的
*
* @param changeEvent
* @param order
* @return
*/
private synchronized boolean sendEvent(OrderStatusChangeEvent changeEvent, Order order,String key){
boolean result = false;
try {
//啟動(dòng)狀態(tài)機(jī)
orderStateMachine.start();
//嘗試恢復(fù)狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.restore(orderStateMachine, String.valueOf(order.getId()));
Message message = MessageBuilder.withPayload(changeEvent).setHeader("order", order).build();
result = orderStateMachine.sendEvent(message);
if(!result){
returnfalse;
}
//獲取到監(jiān)聽的結(jié)果信息
Integer o = (Integer) orderStateMachine.getExtendedState().getVariables().get(key + order.getId());
//操作完成之后,刪除本次對(duì)應(yīng)的key信息
orderStateMachine.getExtendedState().getVariables().remove(key+order.getId());
//如果事務(wù)執(zhí)行成功，則持久化狀態(tài)機(jī)
if(Objects.equals(1,Integer.valueOf(o))){
//持久化狀態(tài)機(jī)狀態(tài)
stateMachineMemPersister.persist(orderStateMachine, String.valueOf(order.getId()));
}else {
//訂單執(zhí)行業(yè)務(wù)異常
returnfalse;
}
} catch (Exception e) {
log.error("訂單操作失敗:{}", e);
} finally {
orderStateMachine.stop();
}
return result;
}使用aop對(duì)監(jiān)聽事件切面，把業(yè)務(wù)執(zhí)行結(jié)果封裝到狀態(tài)機(jī)的變量中，注解：
@Retention(RetentionPolicy.RUNTIME)
public @interface LogResult {
/**
*執(zhí)行的業(yè)務(wù)key
*
* @return String
*/
String key();
}切面：
@Component
@Aspect
@Slf4j
publicclass LogResultAspect {
//攔截 LogHistory注解
@Pointcut("@annotation(com.zengqingfa.springboot.state.demo.aop.annotation.LogResult)")
private void logResultPointCut() {
//logResultPointCut 日志注解切點(diǎn)
}
@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è)務(wù)執(zhí)行正常，則保存信息
//成功 則為1
orderStateMachine.getExtendedState().getVariables().put(key + order.getId(), 1);
} catch (Throwable e) {
log.error("e:{}", e.getMessage());
//如果業(yè)務(wù)執(zhí)行異常，則保存信息
//將異常信息變量信息中，失敗則為0
orderStateMachine.getExtendedState().getVariables().put(key + order.getId(), 0);
throw e;
}
return returnVal;
}
}監(jiān)聽類使用注解：
@Component("orderStateListener")
@WithStateMachine(name = "orderStateMachine")
@Slf4j
publicclass 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)機(jī)反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_DELIVER.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
//模擬異常
if (Objects.equals(order.getName(), "A")) {
thrownew RuntimeException("執(zhí)行業(yè)務(wù)異常");
}
}
@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)機(jī)反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.WAIT_RECEIVE.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
}
@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("確認(rèn)收貨，狀態(tài)機(jī)反饋信息：{}", message.getHeaders().toString());
//更新訂單
order.setStatus(OrderStatus.FINISH.getKey());
orderMapper.updateById(order);
//TODO 其他業(yè)務(wù)
}
}好了，通過Spring狀態(tài)機(jī)驅(qū)動(dòng)訂單支付狀態(tài)流轉(zhuǎn)，你學(xué)會(huì)了嗎？
責(zé)任編輯：武曉燕
來源：
冰河世紀(jì)
