如何在開源項目Cadence中實現(xiàn)輪詢？

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

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

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

Cadence的優(yōu)勢

大量用例遍及單個請求-回復(fù)、復(fù)雜的狀態(tài)追蹤和異步事件響應(yīng)，并與外部的不可靠依賴項進(jìn)行通訊。構(gòu)建此類應(yīng)用程序的常用方法是將無狀態(tài)服務(wù)、數(shù)據(jù)庫、定時任務(wù)和隊列系統(tǒng)像大雜燴一樣整合在一起。

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

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

只需要進(jìn)行短時間的輪詢
能夠接受在輪詢過程中出現(xiàn)合理的等待

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

使用Cadence輪詢外部服務(wù)

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

輪詢的實現(xiàn)

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

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

左右滑動查看完整代碼

我們首先調(diào)用一個活動，在這種情況下，外部服務(wù)可能是REST API。然后我們就需要進(jìn)行條件判斷。如果未達(dá)到所需的狀態(tài)，會有10秒的等待。

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

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

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

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

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

1、謹(jǐn)慎使用輪詢?

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

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

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

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

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

2、配置活動重試次數(shù)?

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

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秒。這樣，每個對外部服務(wù)活動的調(diào)用都可以輕松的實現(xiàn)重試功能，且無需編寫任何重試邏輯。

用例示例：Instafood和MegaBurgers

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

1、Instafood簡介?

Instafood是一個基于在線應(yīng)用的送餐服務(wù)。客戶可以通過Instafood的移動應(yīng)用從他們當(dāng)?shù)刈钕矚g的餐廳中訂購食物。訂單可以是自取或外賣。

如果選擇外賣，Instafood將通知其外賣司機(jī)從餐廳取餐并將其送到客戶手中。Instafood為每個餐廳提供一個展示屏或平板電腦，用于Instafood和餐廳之間的通信。客戶下單后，Instafood就會通知餐廳，然后餐廳可以接受訂單、提供預(yù)計完成時間或?qū)⑵錁?biāo)記為已完成等。對于外送訂單，Instafood將根據(jù)預(yù)計完成時間協(xié)調(diào)外賣司機(jī)取餐。

2、輪詢"MegaBurgers"?

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

MegaBurger的訂單狀態(tài)機(jī)

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

相反，其需要定期發(fā)送GET請求來確定訂單狀態(tài)，這些輪詢可能會導(dǎo)致訂單工作流在Instafood端反復(fù)執(zhí)行（例如安排外賣司機(jī)取餐）。

建立Instafood項目

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

第1步：創(chuàng)建Instaclustr托管集群?

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

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

第2步：配置Cadence域?

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

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

macOS?

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

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

其他操作系統(tǒng)?

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

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

左右滑動查看完整代碼

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

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

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

左右滑動查看完整代碼

我們將其稱為<cadence_host>。

3、現(xiàn)在可以通過列出當(dāng)前域來測試連接：

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

4、添加instafood域：

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

?左右滑動查看完整代碼

5、檢查它是否已經(jīng)注冊：

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

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

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

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

cadenceHost=<cadence_host>

?3、通過以下方式運行該應(yīng)用程序：

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

或從IDE中執(zhí)行InstafoodApplication的main class：

?4、查看終端輸出以確認(rèn)其是否已經(jīng)正常運行：

?了解MegaBurger的API

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

1、運行MegaBurger服務(wù)?

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

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

或者在IDE中運行MegaburgerRestApplication。這是一個以內(nèi)存作為持久層的Spring Boot Rest API演示示例。當(dāng)應(yīng)用程序關(guān)閉時，所有數(shù)據(jù)都會丟失。

2、MegaBurger的訂單API?

MegaBurger發(fā)布其Orders API以便跟蹤和更新每個食品訂單的狀態(tài)。

POST /orders?

創(chuàng)建一個訂單并返回其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輪詢集成項目回顧

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

1、輪詢工作流?

首先定義新的工作流MegaBurgerOrderWorkflow：

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

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

現(xiàn)在來看看它的實現(xiàn)方式：

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的集成，將訂單交付給由獨立工作流管理的客戶端處理；這意味著我們使用的是子工作流。然后，通過活動來創(chuàng)建訂單，并獲得訂單ID。

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

現(xiàn)在我們必須等待訂單從PENDING轉(zhuǎn)變?yōu)锳CCEPTED或REJECTED。這就是輪詢發(fā)揮作用的地方。現(xiàn)在看看我們的函數(shù)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;
}

左右滑動查看完整代碼

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

2、運行正常的場景?

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

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

測試用例描述了客戶端通過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. 客戶端將訂單發(fā)送到Instafood。

2. 一旦訂單到達(dá)MegaBurger（訂單狀態(tài)為PENDING），MegaBurgers將其標(biāo)記為ACCEPTED并返回預(yù)計完成時間。

3. 然后我們看下整個狀態(tài)更新序列：

MegaBurger將訂單標(biāo)記為COOKING。
MegaBurger將訂單標(biāo)記為READY（這意味著它已準(zhǔn)備好外送或自取）。
MegaBurger將訂單標(biāo)記為RESTAURANT_DELIVERED 。

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

總結(jié)

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

譯者介紹

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

責(zé)任編輯：閆懷德來源： 51CTO

Cadence 輪詢開源

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看

如何在開源項目Cadence中實現(xiàn)輪詢？

Cadence的優(yōu)勢

使用Cadence輪詢外部服務(wù)

輪詢的實現(xiàn)

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

1、謹(jǐn)慎使用輪詢?

2、配置活動重試次數(shù)?

用例示例：Instafood和MegaBurgers

1、Instafood簡介?

2、輪詢"MegaBurgers"?

建立Instafood項目

第1步：創(chuàng)建Instaclustr托管集群?

第2步：配置Cadence域?

macOS?

其他操作系統(tǒng)?

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

?了解MegaBurger的API

1、運行MegaBurger服務(wù)?

2、MegaBurger的訂單API?

MegaBurger輪詢集成項目回顧

1、輪詢工作流?

2、運行正常的場景?

總結(jié)

譯者介紹