如何在開源項目Cadence中實現輪詢？

作者：仇凱 2022-07-15 19:57:18

Cadence項目對所有開發人員來說都是非常有用的新工具，因此了解如何通過Cadence項目實現輪詢很有價值

本指南適用于所有希望了解Cadence中輪詢工作原理的開發人員和工程師。Cadence是相對較新（且完全開源）的容錯狀態代碼平臺，最初由Uber開發（現在得到了包括Instaclustr在內的更多公司的支持）。

Cadence的優勢

大量用例遍及單個請求-回復、復雜的狀態追蹤和異步事件響應，并與外部的不可靠依賴項進行通訊。構建此類應用程序的常用方法是將無狀態服務、數據庫、定時任務和隊列系統像大雜燴一樣整合在一起。

然而，這會對開發人員產生負面影響，因為大部分代碼都是用于管道的——這掩蓋了大量底層細節背后的實際業務邏輯。Cadence是一個完全開源的編排框架，可以幫助開發人員編寫高容錯且能夠長時間運行的應用程序，這通常也被稱為工作流。

從本質上講，它提供了一個與特定進程無關聯的虛擬內存，并保留了完整的應用程序狀態，包括函數堆棧以及兼容各種主機和軟件故障的局部變量。這使得開發人員在編寫代碼時能夠充分利用編程語言的功能特性，Cadence則負責應用程序的持久性、可用性和可擴展性。由于繁忙等待通常會消耗大量非必要的CPU周期，因而應盡可能避免使用輪詢，而是使用由事件觸發的中斷來進行實現，除非以下兩者情況：

只需要進行短時間的輪詢
能夠接受在輪詢過程中出現合理的等待

對于計算機而言，這相當于在長途旅行中每5分鐘詢問一次距離目的地還有多遠。盡管如此，在很多情況下，這是唯一可用的選擇。Cadence為持久計時器、長時間運行的活動和無限制重試提供強大的支持，使得其非常適合此類功能的實現。

使用Cadence輪詢外部服務

實現輪詢機制有很多種方法。本文主要講實現對外部服務的輪詢，并分析這樣做會從Cadence中獲得怎樣的收益。首先，我們來簡單的解釋一下Cadence的概念。Cadence的核心理念是一個無故障狀態的工作流。這意味著工作流代碼的狀態，包括局部變量和它創建的任何線程，不受進程和Cadence服務故障的影響。這是一個非常強大的理念，因為它封裝了狀態、線程處理、持久計時器和事件處理程序。為了滿足確定性的執行要求，工作流不允許直接調用任何外部API。相反，它們負責對活動的執行進行調度。活動是用來實現業務級功能的應用程序邏輯，例如調用服務或對媒體文件進行轉碼。當活動出現故障時，Cadence并不會恢復其運行狀態。因此，活動函數可以包含任何代碼，且不會受到任何限制。

輪詢的實現

代碼本身非常簡單——我們將逐行解釋代碼的作用：

State polledState = externalServiceActivities.getState();   while(!expectedState.equals(polledState)) {
  Workflow.sleep(Duration.ofSeconds(30));  
polledState = externalServiceActivities.getState();  
}

左右滑動查看完整代碼

我們首先調用一個活動，在這種情況下，外部服務可能是REST API。然后我們就需要進行條件判斷。如果未達到所需的狀態，會有10秒的等待。

這不是通常意義上的等待，而是一個持久的計時器。在這種情況下，輪詢會執行周期性的等待，但時間可能會更長；而且，如果執行失敗，我們一定不會希望浪費整個時間周期。Cadence通過將計時器以事件的方式進行持久化，并在完成后通知相應的工作服務（即管理工作流和活動實施的服務）來解決此問題。

這些計時器可以對從幾秒到幾分鐘、幾小時、幾天甚至幾個月或幾年的時間間隔進行管理。最后，通過再次調用外部服務來刷新狀態。在繼續進行操作之前，我們先快速了解一下Cadence究竟在后臺做了哪些工作來避免潛在的問題。

重要提醒：Cadence歷史記錄和輪詢注意事項

Cadence是如何實現無故障狀態工作流的呢？關鍵在于Cadence是如何堅持用工作流程執行實現的。工作流狀態恢復利用事件溯源，而事件溯源對代碼的編寫方式施加了一些限制。事件溯源將一系列不斷變化的事件轉為持久化的狀態。

每當工作流狀態發生變化時，都會有一個新的事件追加到該工作流的事件歷史記錄中。然后，Cadence通過歷史記錄來進行操作重放，以重新建立工作流的當前狀態。這就是為什么與外部環境的所有通信都應該通過活動進行，并且必須使用Cadence API來獲取當前時間、等待和創建新線程。

1、謹慎使用輪詢?

輪詢需要根據判斷條件不斷地循環。由于每個活動調用和計時器事件都是持久的，因此即使是短的輪詢間隔也可能會演變成不可接受的時間消耗。現在我們來研究輪詢片段的歷史記錄會以怎樣的方式呈現。

首先建立輪詢外部服務所需的活動。
活動由工作服務啟動。
活動完成后返回其結果。
如果條件尚未滿足，則啟動計時器。
一旦超時，就會觸發一個事件來喚醒工作流。
重復上面5個步驟，直到條件滿足。
最終的輪詢確認條件滿足（無需設置定時器）。
工作流被標記為完成。

Cadence中輪詢代碼片段的事件歷史記錄

如果工作流在中間某個地方失敗，且必須重放其歷史操作記錄，這可能會導致大量的事件清單被執行。有一些方法可以避免這些操作脫離掌控：避免使用較短的輪詢周期，在工作流中設置合理的超時時間，限制輪詢的次數。

記住所有操作都是持久的，可能需要由人重放操作。

2、配置活動重試次數?

如果外部服務由于某些原因失敗了怎么辦？我們需要嘗試，嘗試，再嘗試！Cadence存在一種機制，可以讓Cadence記錄活動結果并能夠完美地恢復工作流狀態，同時還提供了對類似重試邏輯等額外功能的支持。下面是啟用重試選項的活動配置示例：

private final ExternalServiceActivities externalServiceActivities =    Workflow.newActivityStub(ExternalServiceActivities.class,   new ActivityOptions.Builder()
.setRetryOptions(new RetryOptions.Builder()   .setInitialInterval(Duration.ofSeconds(10))    .setMaximumAttempts(3)
.build())
.setScheduleToCloseTimeout(Duration.ofMinutes(5))    .build());

??左右滑動查看完整代碼

通過這樣的操作，我們告訴Cadence，在ExternalServiceActivities中的操作最多可以重試3次，且每次重試的間隔為10秒。這樣，每個對外部服務活動的調用都可以輕松的實現重試功能，且無需編寫任何重試邏輯。

用例示例：Instafood和MegaBurgers

為了展示這種模式的實際效果，我們將在示例項目中集成一個虛構的輪詢。

1、Instafood簡介?

Instafood是一個基于在線應用的送餐服務。客戶可以通過Instafood的移動應用從他們當地最喜歡的餐廳中訂購食物。訂單可以是自取或外賣。

如果選擇外賣，Instafood將通知其外賣司機從餐廳取餐并將其送到客戶手中。Instafood為每個餐廳提供一個展示屏或平板電腦，用于Instafood和餐廳之間的通信。客戶下單后，Instafood就會通知餐廳，然后餐廳可以接受訂單、提供預計完成時間或將其標記為已完成等。對于外送訂單，Instafood將根據預計完成時間協調外賣司機取餐。

2、輪詢"MegaBurgers"?

MegaBurgers是一家大型跨國快餐漢堡連鎖店。他們有自己的移動應用程序和網站，并使用REST API作為后端為客戶提供訂單服務。Instafood和MegaBurgers已達成協議，Instafood客戶可以通過Instafood的應用程序在MegaBurger下單，并可選擇自取和外賣。與通用方案不同的是，MegaBurger并未選擇在所有店面安裝Instafood展示屏，而是同意將Instafood的訂餐系統以集成的方式與自身基于REST的訂餐系統進行對接，以完成下單和接收更新。

MegaBurger的訂單狀態機

MegaBurger的REST API沒有推送機制（WebSockets、WebHooks等），無法接收訂單狀態更新。

相反，其需要定期發送GET請求來確定訂單狀態，這些輪詢可能會導致訂單工作流在Instafood端反復執行（例如安排外賣司機取餐）。

建立Instafood項目

你需要配置一個Cadence集群來運行示例項目。在此示例中，我們將使用Instaclustr平臺來執行操作。

第1步：創建Instaclustr托管集群?

Cadence集群需要連接Apache Cassandra集群作為持久層。為了成功配置Cadence和Cassandra集群，我們將遵循“創建Cadence集群”文檔的操作指導。

以下操作會自動進行初始化，無需人為干預：
防火墻規則將在Cassandra集群上為Cadence節點自動生成配置。
Cadence和Cassandra之間的身份驗證（包括客戶端加密）將會自動生成配置。
Cadence的默認配置和鍵空間的可見性將在Cassandra中自動創建。
兩個集群之間會建立關聯關系，以確保不會在停止Cadence集群之前因意外而刪除Cassandra集群。
將創建一個負載均衡器。建議通過負載均衡器地址來連接和訪問集群。

第2步：配置Cadence域?

Cadence的后端由多租戶服務提供支持，其中隔離單元被稱為域。為了讓Instafood應用程序運行，我們首先需要為它注冊一個域。

1、為了與Cadence集群進行交互，我們需要安裝其命令行界面客戶端。

macOS?

如果使用macOS，可以通過Homebrew安裝Cadence CLI，如下所示：

brew install cadence-workflow
# run command line client
cadence <command> <arguments>

其他操作系統?

可以通過Docker Hub鏡像倉庫ubercadence/cli來運行和使用CLI：

# run command line client
docker run --network=host --rm ubercadence/cli:master <command>  <arguments>

左右滑動查看完整代碼

在以后的步驟中，我們將使用cadence來指代客戶端。

2、為了連接的穩定性，建議通過負載均衡器地址來連接和訪問集群。可以在“連接信息”選項卡的頂部找到負載均衡器地址，如下所示：

“ab-cd12ef23-45gh-4baf-ad99-df4xy-azba45bc0c8da111.elb.us-east 1.amazonaws.com”

左右滑動查看完整代碼

我們將其稱為<cadence_host>。

3、現在可以通過列出當前域來測試連接：

cadence --ad <cadence_host>:7933 admin domain list

4、添加instafood域：

cadence --ad <cadence_host>:7933 --do instafood domain register --global_domain=false

?左右滑動查看完整代碼

5、檢查它是否已經注冊：

cadence --ad <cadence_host>:7933 --do instafood domain describe

第3步：運行Instafood示例項目?

1、從Instafood項目的Git代碼倉庫中克隆Gradle項目。

2、打開位于instafood/src/main/resources/instafood.properties路徑的配置文件，將cadenceHost的值替換為自己的負載均衡器地址：

cadenceHost=<cadence_host>

?3、通過以下方式運行該應用程序：

cadence-cookbooks-instafood/instafood$ ./gradlew run

或從IDE中執行InstafoodApplication的main class：

?4、查看終端輸出以確認其是否已經正常運行：

?了解MegaBurger的API

在了解Instafood如何與MegaBurger集成之前，讓我們先快速了解一下他們的API。

1、運行MegaBurger服務?

讓我們從運行服務開始。通過以下命令來啟動服務：

cadence-cookbooks-instafood/megaburger$ ./gradlew run

或者在IDE中運行MegaburgerRestApplication。這是一個以內存作為持久層的Spring Boot Rest API演示示例。當應用程序關閉時，所有數據都會丟失。

2、MegaBurger的訂單API?

MegaBurger發布其Orders API以便跟蹤和更新每個食品訂單的狀態。

POST /orders?

創建一個訂單并返回其ID。

Request:

curl -X POST localhost:8080/orders -H “Content-Type: application/json” --data ‘{“meal”: “Vegan Burger”, “quantity”: 1}’

左右滑動查看完整代碼

Response:

{
  “id”: 1,
  “meal”: “Vegan Burger”,
  “quantity”: 1,
  “status”: “PENDING”,
  “eta_minutes”: null
}

GET /orders?

返回一個包含所有訂單信息的列表。

Request:

curl -X GET localhost:8080/orders

?Response:

[
{
    “id”: 0,
    “meal”: “Vegan Burger”,
    “quantity”: 1,
    “status”: “PENDING”,
    “eta_minutes”: null
},
{
    “id”: 1,
    “meal”: “Onion Rings”,
    “quantity”: 2,
    “status”: “PENDING”,
    “eta_minutes”: null
}
]

GET /orders / {orderId}?

返回ID與orderId一致的訂單。

Request:

curl -X GET localhost:8080/orders/1

?Response:

{
  “id”: 1,
  “meal”: “Onion Rings”,
  “quantity”: 2,
  “status”: “PENDING”,
  “eta_minutes”: null
}

PATCH /orders/{orderId}

更新ID與orderId一致的訂單

Request:

curl -X PATCH localhost:8080/orders/1 -H “Content-Type: application/ json” --data ‘{“status”:“ACCEPTED”}’

?左右滑動查看完整代碼

Response:

{
    “id”: 1,
    “meal”: “Onion Rings”,
    “quantity”: 2,
    “status”: “ACCEPTED”,
    “eta_minutes”: null
}

MegaBurger輪詢集成項目回顧

現在已經完成了所有配置的初始化，讓我們看看Instafood和MegaBurger之間集成的實際效果如何。

1、輪詢工作流?

首先定義新的工作流MegaBurgerOrderWorkflow：

public interface MegaBurgerOrderWorkflow {
@WorkflowMethod
void orderFood(FoodOrder order);
// ...
}

此工作流有一個orderFood方法，該方法將通過與MegaBurger集成來發送和跟蹤相應的FoodOrder。

現在來看看它的實現方式：

public class MegaBurgerOrderWorkflowImpl implements MegaBurgerOrderWork flow {
// ...
@Override
public void orderFood(FoodOrder order) {
  OrderWorkflow parentOrderWorkflow = getParentOrderWorkflow(); 
Integer orderId = megaBurgerOrderActivities.createOrder(mapMega   BurgerFoodOrder(order));
 updateOrderStatus(parentOrderWorkflow, OrderStatus.PENDING);
// Poll until Order is accepted/rejected
updateOrderStatus(parentOrderWorkflow,              pollOrderStatusTransition(orderId, OrderStatus.
PENDING));
if (OrderStatus.REJECTED.equals(currentStatus)) {
 throw new RuntimeException(“Order with id “ + orderId + “      was rejected”);
}
  // Send ETA to parent workflow
  parentOrderWorkflow.updateEta(getOrderEta(orderId));  // Poll until Order is cooking
 updateOrderStatus(parentOrderWorkflow,            pollOrderStatusTransition(orderId, OrderStatus.ACCEPTED)); //   Poll until Order is ready
updateOrderStatus(parentOrderWorkflow,                pollOrderStatusTransition(orderId, OrderStatus.COOKING)); //   Poll until Order is delivered
updateOrderStatus(parentOrderWorkflow,
pollOrderStatusTransition(orderId, OrderStatus.READY)); }
// ...
}

左右滑動查看完整代碼

該工作流首先獲取其父工作流。MegaBurgerOrderWorkflow只處理與MegaBurger的集成，將訂單交付給由獨立工作流管理的客戶端處理；這意味著我們使用的是子工作流。然后，通過活動來創建訂單，并獲得訂單ID。

活動只是API客戶端的裝飾器，該API客戶端負責發送POST請求到/orders。創建訂單后，父工作流會收到一個訂單現在處于PENDING狀態的信號（這是一個發送給工作流的，來自外部的異步請求）。

現在我們必須等待訂單從PENDING轉變為ACCEPTED或REJECTED。這就是輪詢發揮作用的地方。現在看看我們的函數pollOrderStatusTransition做了什么：

private OrderStatus pollOrderStatusTransition(Integer orderId,  OrderStatus orderStatus) { OrderStatus polledStatus =
megaBurgerOrderActivities.getOrderById(orderId).getStatus();  while (orderStatus.equals(polledStatus)) {
    Workflow.sleep(Duration.ofSeconds(30));
 polledStatus = megaBurgerOrderActivities.
 getOrderById(orderId).getStatus();
}
return polledStatus;
}

左右滑動查看完整代碼

這與本文介紹的其他輪詢循環非常相似。唯一的區別是它用一個輪詢的特定狀態代替等待，直到訂單狀態發生變化。同樣的，用于通過ID獲取訂單的真實API調用隱藏在活動的后面，該活動啟用了重試功能。如果訂單被拒絕，則會引發運行狀態異常，使工作流失敗。如果訂單被接受，則將MegaBurger的預計完成時間返回給父工作流（父工作流使用預計完成時間來完成交付調度）。最后，圖3中所示的狀態將會被轉換，直到訂單被標記為已交付。

2、運行正常的場景?

最后，讓完成一個完整的訂單場景。

這個場景是示例項目中測試套件的一部分。唯一的要求是同時運行Instafood和MegaBurger服務器，然后按照前文中的步驟操作。

測試用例描述了客戶端通過Instafood下單MegaBurger的新素食漢堡，并且來店面取餐：

cadence-cookbooks-instafood/instafood$ ./gradlew test

或在IDE中運行InstafoodApplicationTest：

class InstafoodApplicationTest {
// ...
@Test
public void
givenAnOrderItShouldBeSentToMegaBurgerAndBeDeliveredAccordingly() {   FoodOrder order = new FoodOrder(Restaurant.MEGABURGER,      “Vegan Burger”, 2, “+54 11 2343-2324”, “Díaz velez 433, La        lucila”, true);
  // Client orders food
        WorkflowExecution workflowExecution= WorkflowClient       start(orderWorkflow::orderFood, order);
  // Wait until order is pending Megaburger’s acceptance await().until(() -> OrderStatus.PENDING.equals(orderWorkflow.    getStatus()));
// Megaburger accepts order and sends ETA
megaBurgerOrdersApiClient.updateStatusAndEta(getLastOrderId(),    “ACCEPTED”, 15);
await().until(() -> OrderStatus.ACCEPTED.equals(orderWorkflow.    getStatus()));
// Megaburger starts cooking order
megaBurgerOrdersApiClient.updateStatus(getLastOrderId(),    “COOKING”);
await().until(() -> OrderStatus.COOKING.equals(orderWorkflow.    getStatus()));
// Megaburger signals order is ready
megaBurgerOrdersApiClient.updateStatus(getLastOrderId(),    “READY”);
await().until(() -> OrderStatus.READY.equals(orderWorkflow.      getStatus()));
// Megaburger signals order has been picked-up
megaBurgerOrdersApiClient.updateStatus(getLastOrderId(),    “RESTAURANT_DELIVERED”);
await().until(() -> OrderStatus.RESTAURANT_DELIVERED.
equals(orderWorkflow.getStatus()));
await().until(() -> workflowHistoryHasEvent(workflowClient,         workflowExecution, EventType.WorkflowExecutionCompleted)):   }
 }

左右滑動查看完整代碼

在這個場景中，有3個參與者：Instafood、MegaBurger和客戶端。

1. 客戶端將訂單發送到Instafood。

2. 一旦訂單到達MegaBurger（訂單狀態為PENDING），MegaBurgers將其標記為ACCEPTED并返回預計完成時間。

3. 然后我們看下整個狀態更新序列：

MegaBurger將訂單標記為COOKING。
MegaBurger將訂單標記為READY（這意味著它已準備好外送或自取）。
MegaBurger將訂單標記為RESTAURANT_DELIVERED 。

4. 由于該訂單是以取餐的形式交付，因此一旦客戶端完成交付，整個工作流程就結束了。

總結

在本文中，我們學習了如何使用Cadence實現輪詢。我們展示了如何讓Cadence集群在Instaclustr平臺上運行，以及讓應用程序連接到它是多么容易。參考鏈接：https://dzone.com/articles/how-to-use-open-source-cadence-for-polling

譯者介紹

仇凱，51CTO社區編輯，目前就職于北京宅急送快運股份有限公司，職位為信息安全工程師。主要負責公司信息安全規劃和建設（等保，ISO27001），日常主要工作內容為安全方案制定和落地、內部安全審計和風險評估以及管理。

責任編輯：閆懷德來源： 51CTO

Cadence 輪詢開源