php线程轮询技巧_若安在开源项目Cadence中实现轮询

然而，这会对开拓职员产生负面影响，由于大部分代码都是用于管道的——这粉饰了大量底层细节背后的实际业务逻辑。
Cadence是一个完备开源的编排框架，可以帮助开拓职员编写高容错且能够永劫光运行的运用程序，这常日也被称为事情流。

从实质上讲，它供应了一个与特定进程无关联的虚拟内存，并保留了完全的运用程序状态，包括函数堆栈以及兼容各种主机和软件故障的局部变量。
这使得开拓职员在编写代码时能够充分利用编程措辞的功能特性，Cadence则卖力运用程序的持久性、可用性和可扩展性。
由于繁忙等待常日会花费大量非必要的CPU周期，因而应尽可能避免利用轮询，而是利用由事宜触发的中断来进行实现，除非以下两者情形：

对付打算机而言，这相称于在长途旅行中每5分钟讯问一次间隔目的地还有多远。
只管如此，在很多情形下，这是唯一可用的选择。
Cadence为持久计时器、永劫光运行的活动和无限制重试供应强大的支持，使得其非常适宜此类功能的实现。

实现轮询机制有很多种方法。
本文紧张讲实现对外部做事的轮询，并剖析这样做会从Cadence中得到若何的收益。
首先，我们来大略的阐明一下Cadence的观点。
Cadence的核心理念是一个无端障状态的事情流。
这意味着事情流代码的状态，包括局部变量和它创建的任何线程，不受进程和Cadence做事故障的影响。
这是一个非常强大的理念，由于它封装了状态、线程处理、持久计时器和事宜处理程序。
为了知足确定性的实行哀求，事情流不许可直接调用任何外部API。
相反，它们卖力对活动的实行进行调度。
活动是用来实现业务级功能的运用程序逻辑，例如调用做事或对媒体文件进行转码。
当活动涌现故障时，Cadence并不会规复其运行状态。
因此，活动函数可以包含任何代码，且不会受到任何限定。

代码本身非常大略——我们将逐行阐明代码的浸染：

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

我们首先调用一个活动，在这种情形下，外部做事可能是REST API。
然后我们就须要进行条件判断。
如果未达到所需的状态，会有10秒的等待。

这不是常日意义上的等待，而是一个持久的计时器。
在这种情形下，轮询会实行周期性的等待，但韶光可能会更长；而且，如果实行失落败，我们一定不会希望摧残浪费蹂躏全体韶光周期。
Cadence通过将计时器以事宜的办法进行持久化，并在完成后关照相应的事情做事（即管理事情流和活动履行的做事）来办理此问题。

这些计时器可以对从几秒到几分钟、几小时、几天乃至几个月或几年的韶光间隔进行管理。
末了，通过再次调用外部做事来刷新状态。
在连续进行操作之前，我们先快速理解一下Cadence究竟在后台做了哪些事情来避免潜在的问题。

Cadence是如何实现无端障状态事情流的呢？关键在于Cadence是如何坚持用事情流程实行实现的。
事情流状态规复利用事宜溯源，而事宜溯源对代码的编写办法施加了一些限定。
事宜溯源将一系列不断变革的事宜转为持久化的状态。

每当事情流状态发生变革时，都会有一个新的事宜追加到该事情流的事宜历史记录中。
然后，Cadence通过历史记录来进行操作重放，以重新建立事情流确当前状态。
这便是为什么与外部环境的所有通信都该当通过活动进行，并且必须利用Cadence API来获取当前韶光、等待和创建新线程。

轮询须要根据判断条件不断地循环。
由于每个活动调用和计时器事宜都是持久的，因此纵然是短的轮询间隔也可能会演化成不可接管的韶光花费。
现在我们来研究轮询片段的历史记录会以若何的办法呈现。

Cadence中轮询代码片段的事宜历史记录

如果事情流在中间某个地方失落败，且必须重放其历史操作记录，这可能会导致大量的事宜清单被实行。
有一些方法可以避免这些操作分开掌控：避免利用较短的轮询周期，在事情流中设置合理的超时时间，限定轮询的次数。

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

如果外部做事由于某些缘故原由失落败了怎么办？我们须要考试测验，考试测验，再考试测验！
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());1.2.3.4.

通过这样的操作，我们见告Cadence，在ExternalServiceActivities中的操作最多可以重试3次，且每次重试的间隔为10秒。
这样，每个对外部做事活动的调用都可以轻松的实现重试功能，且无需编写任何重试逻辑。

为了展示这种模式的实际效果，我们将在示例项目中集成一个虚构的轮询。

Instafood是一个基于在线运用的送餐做事。
客户可以通过Instafood的移动运用从他们当地最喜好的餐厅中订购食品。
订单可以是自取或外卖。

如果选择外卖，Instafood将关照其外卖司机从餐厅取餐并将其送到客户手中。
Instafood为每个餐厅供应一个展示屏或平板电脑，用于Instafood和餐厅之间的通信。
客户下单后，Instafood就会关照餐厅，然后餐厅可以接管订单、供应估量完成韶光或将其标记为已完成等。
对付外送订单，Instafood将根据估量完成韶光折衷外卖司机取餐。

MegaBurgers是一家大型跨国快餐汉堡连锁店。
他们有自己的移动运用程序和网站，并利用REST API作为后端为客户供应订单做事。
Instafood和MegaBurgers已达成协议，Instafood客户可以通过Instafood的运用程序在MegaBurger下单，并可选择自取和外卖。
与通用方案不同的是，MegaBurger并未选择在所有店面安装Instafood展示屏，而是赞许将Instafood的订餐系统以集成的办法与自身基于REST的订餐系统进行对接，以完成下单和吸收更新。

MegaBurger的订单状态机

MegaBurger的REST API没有推送机制（WebSockets、WebHooks等），无法吸收订单状态更新。

相反，其须要定期发送GET要求来确定订单状态，这些轮询可能会导致订单事情流在Instafood端反复实行（例如安排外卖司机取餐）。

你须要配置一个Cadence集群来运行示例项目。
在此示例中，我们将利用Instaclustr平台来实行操作。

Cadence集群须要连接Apache Cassandra集群作为持久层。
为了成功配置Cadence和Cassandra集群，我们将遵照“创建Cadence集群”文档的操作辅导。

Cadence的后端由多租户做事供应支持，个中隔离单元被称为域。
为了让Instafood运用程序运行，我们首先须要为它注册一个域。

1、为了与Cadence集群进行交互，我们须要安装其命令行界面客户端。

如果利用macOS，可以通过Homebrew安装Cadence CLI，如下所示：

brew install cadence-workflow# run command line clientcadence <command> <arguments>1.2.3.其他操作系统​

可以通过Docker Hub镜像仓库ubercadence/cli来运行和利用CLI：

# run command line clientdocker run --network=host --rm ubercadence/cli:master <command> <arguments>1.2.

在往后的步骤中，我们将利用cadence来指代客户端。

2、为了连接的稳定性，建议通过负载均衡器地址来连接和访问集群。
可以在“连接信息”选项卡的顶部找到负载均衡器地址，如下所示：

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

我们将其称为<cadence_host>。

3、现在可以通过列出当前域来测试连接：

cadence --ad <cadence_host>:7933 admin domain list1.

4、添加instafood域：

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

5、检讨它是否已经注册：

cadence --ad <cadence_host>:7933 --do instafood domain describe1.第3步：运行Instafood示例项目​

1、从Instafood项目的Git代码仓库中克隆Gradle项目。

2、打开位于instafood/src/main/resources/instafood.properties路径的配置文件，将cadenceHost的值更换为自己的负载均衡器地址：

cadenceHost=<cadence_host>1.

​3、通过以下办法运行该运用程序：

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

或从IDE中实行InstafoodApplication的main class：

​4、查看终端输出以确认其是否已经正常运行：

​

在理解Instafood如何与MegaBurger集成之前，让我们先快速理解一下他们的API。

让我们从运行做事开始。
通过以下命令来启动做事：

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

或者在IDE中运行MegaburgerRestApplication。
这是一个以内存作为持久层的Spring Boot Rest 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}’1.

Response:

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

GET /orders​

返回一个包含所有订单信息的列表。

Request:

curl -X GET localhost:8080/orders1.

​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}]1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.

GET /orders / {orderId}​

返回ID与orderId同等的订单。

Request:

curl -X GET localhost:8080/orders/11.

​Response:

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

PATCH /orders/{orderId}

更新ID与orderId同等的订单

Request:

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

Response:

{ “id”: 1, “meal”: “Onion Rings”, “quantity”: 2, “status”: “ACCEPTED”, “eta_minutes”: null}1.2.3.4.5.6.7.MegaBurger轮询集成项目回顾

现在已经完成了所有配置的初始化，让我们看看Instafood和MegaBurger之间集成的实际效果如何。

首先定义新的事情流MegaBurgerOrderWorkflow：

public interface MegaBurgerOrderWorkflow {@WorkflowMethodvoid orderFood(FoodOrder order);// ...}1.2.3.4.5.

此事情流有一个orderFood方法，该方法将通过与MegaBurger集成来发送和跟踪相应的FoodOrder。

现在来看看它的实现办法：

public class MegaBurgerOrderWorkflowImpl implements MegaBurgerOrderWork flow {// ...@Overridepublic void orderFood(FoodOrder order) { OrderWorkflow parentOrderWorkflow = getParentOrderWorkflow(); Integer orderId = megaBurgerOrderActivities.createOrder(mapMega BurgerFoodOrder(order)); updateOrderStatus(parentOrderWorkflow, OrderStatus.PENDING);// Poll until Order is accepted/rejectedupdateOrderStatus(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 readyupdateOrderStatus(parentOrderWorkflow, pollOrderStatusTransition(orderId, OrderStatus.COOKING)); // Poll until Order is deliveredupdateOrderStatus(parentOrderWorkflow,pollOrderStatusTransition(orderId, OrderStatus.READY)); }// ...}1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.

该事情流首先获取其父事情流。
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;}1.2.3.4.5.6.7.8.

这与本文先容的其他轮询循环非常相似。
唯一的差异是它用一个轮询的特定状态代替等待，直到订单状态发生变革。
同样的，用于通过ID获取订单的真实API调用隐蔽在活动的后面，该活动启用了重试功能。
如果订单被谢绝，则会引发运行状态非常，使事情流失落败。
如果订单被接管，则将MegaBurger的估量完成韶光返回给父事情流（父事情流利用估量完成韶光来完成交付调度）。
末了，图3中所示的状态将会被转换，直到订单被标记为已交付。

末了，让完成一个完全的订单场景。

这个场景是示例项目中测试套件的一部分。
唯一的哀求是同时运行Instafood和MegaBurger做事器，然后按照前文中的步骤操作。

测试用例描述了客户端通过Instafood下单MegaBurger的新素食汉堡，并且来店面取餐：

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

或在IDE中运行InstafoodApplicationTest：

class InstafoodApplicationTest {// ...@Testpublic voidgivenAnOrderItShouldBeSentToMegaBurgerAndBeDeliveredAccordingly() { 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 ETAmegaBurgerOrdersApiClient.updateStatusAndEta(getLastOrderId(), “ACCEPTED”, 15);await().until(() -> OrderStatus.ACCEPTED.equals(orderWorkflow. getStatus()));// Megaburger starts cooking ordermegaBurgerOrdersApiClient.updateStatus(getLastOrderId(), “COOKING”);await().until(() -> OrderStatus.COOKING.equals(orderWorkflow. getStatus()));// Megaburger signals order is readymegaBurgerOrdersApiClient.updateStatus(getLastOrderId(), “READY”);await().until(() -> OrderStatus.READY.equals(orderWorkflow. getStatus()));// Megaburger signals order has been picked-upmegaBurgerOrdersApiClient.updateStatus(getLastOrderId(), “RESTAURANT_DELIVERED”);await().until(() -> OrderStatus.RESTAURANT_DELIVERED.equals(orderWorkflow.getStatus()));await().until(() -> workflowHistoryHasEvent(workflowClient, workflowExecution, EventType.WorkflowExecutionCompleted)): } }1.2.3.4.5.6.7.8.9.10.11.12.13.14.15.16.17.18.19.20.21.22.23.

在这个场景中，有3个参与者：Instafood、MegaBurger和客户端。

1. 客户端将订单发送到Instafood。

2. 一旦订单到达MegaBurger（订单状态为PENDING），MegaBurgers将其标记为ACCEPTED并返回估量完成韶光。

3. 然后我们看下全体状态更新序列：

4. 由于该订单因此取餐的形式交付，因此一旦客户端完成交付，全体事情流程就结束了。

在本文中，我们学习了如何利用Cadence实现轮询。
我们展示了如何让Cadence集群在Instaclustr平台上运行，以及让运用程序连接到它是多么随意马虎。
参考链接：https://dzone.com/articles/how-to-use-open-source-cadence-for-polling

仇凯，51CTO社区编辑，目前就职于北京宅急送快运株式会社，职位为信息安全工程师。
紧张卖力公司信息安全方案和培植（等保，ISO27001），日常紧张事情内容为安全方案制订和落地、内部安全审计和风险评估以及管理。

作者： 仇凯

来源： 51CTO技能栈