快速入门：工作流

开始使用 Dapr 工作流构建块

注意

Dapr工作流目前处于beta阶段。 查看已知限制 1.14.1.

让我们来看看 Dapr 的Workflow构建块。 在这个快速入门中，您将创建一个简单的控制台应用程序，以演示Dapr的工作流编程模型和工作流管理API。

在本指南中，您将：

  • 再次运行 order-processor 应用程序。
  • 启动工作流并观察工作流活动/任务的执行。
  • 查看工作流逻辑和工作流活动，以及它们在代码中的表示方式。

Workflow - 图1

在继续快速入门之前，请选择您首选的特定语言 Dapr SDK。

order-processor 控制台应用启动并管理 order_processing_workflow，该工作流模拟从商店购买物品。 工作流由五个独特的工作流活动或任务组成：

  • notify_activity: 利用记录器在整个工作流中打印出消息。 这些消息在以下情况下通知您：
    • 您的库存不足
    • 您的付款无法处理等等。
  • process_payment_activity: 处理并授权支付。
  • verify_inventory_activity: 检查状态存储以确保购买时有足够的库存存在。
  • update_inventory_activity: 从状态存储中删除请求的项目，并使用新的剩余库存值更新存储。
  • request_approval_activity: 如果付款金额超过50,000美元，则向经理请求批准。

第1步：先决条件

对于此示例，您将需要：

第2步：设置环境

克隆在Quickstarts存储库中提供的示例。

  1. git clone https://github.com/dapr/quickstarts.git

在一个新的终端窗口中，导航到 order-processor 目录：

  1. cd workflows/python/sdk/order-processor

安装 Dapr Python SDK 包:

  1. pip3 install -r requirements.txt

第3步：运行订单处理程序应用

在终端中，使用 Multi-App Run 启动订单处理应用程序，并与一个 Dapr sidecar 并行运行：

  1. dapr run -f .

这将使用唯一的工作流ID启动order-processor应用程序，并运行工作流活动。

预期输出：

  1. == APP == Starting order workflow, purchasing 10 of cars
  2. == APP == 2023-06-06 09:35:52.945 durabletask-worker INFO: Successfully connected to 127.0.0.1:65406. Waiting for work items...
  3. == APP == INFO:NotifyActivity:Received order f4e1926e-3721-478d-be8a-f5bebd1995da for 10 cars at $150000 !
  4. == APP == INFO:VerifyInventoryActivity:Verifying inventory for order f4e1926e-3721-478d-be8a-f5bebd1995da of 10 cars
  5. == APP == INFO:VerifyInventoryActivity:There are 100 Cars available for purchase
  6. == APP == INFO:RequestApprovalActivity:Requesting approval for payment of 165000 USD for 10 cars
  7. == APP == 2023-06-06 09:36:05.969 durabletask-worker INFO: f4e1926e-3721-478d-be8a-f5bebd1995da Event raised: manager_approval
  8. == APP == INFO:NotifyActivity:Payment for order f4e1926e-3721-478d-be8a-f5bebd1995da has been approved!
  9. == APP == INFO:ProcessPaymentActivity:Processing payment: f4e1926e-3721-478d-be8a-f5bebd1995da for 10 cars at 150000 USD
  10. == APP == INFO:ProcessPaymentActivity:Payment for request ID f4e1926e-3721-478d-be8a-f5bebd1995da processed successfully
  11. == APP == INFO:UpdateInventoryActivity:Checking inventory for order f4e1926e-3721-478d-be8a-f5bebd1995da for 10 cars
  12. == APP == INFO:UpdateInventoryActivity:There are now 90 cars left in stock
  13. == APP == INFO:NotifyActivity:Order f4e1926e-3721-478d-be8a-f5bebd1995da has completed!
  14. == APP == 2023-06-06 09:36:06.106 durabletask-worker INFO: f4e1926e-3721-478d-be8a-f5bebd1995da: Orchestration completed with status: COMPLETED
  15. == APP == Workflow completed! Result: Completed
  16. == APP == Purchase of item is  Completed

（可选）第4步：在Zipkin中查看

运行 dapr init 启动 openzipkin/zipkin Docker 容器。 如果容器已停止运行，请使用以下命令启动Zipkin Docker容器：

  1. docker run -d -p 9411:9411 openzipkin/zipkin

在 Zipkin web UI 中查看工作流 trace span（通常在 http://localhost:9411/zipkin/）。

Workflow - 图2

发生了什么？

当你运行 dapr run -f .：

  1. 生成一个唯一的工作流订单ID（在上面的示例中，f4e1926e-3721-478d-be8a-f5bebd1995da）并安排工作流。
  2. NotifyActivity 工作流活动发送通知，说明已收到一份包含10辆车的订单。
  3. ReserveInventoryActivity工作流活动检查库存数据，确定是否可以提供订购的物料，并使用库存中的汽车数量进行响应。
  4. 您的工作流程开始并通知您其状态。
  5. ProcessPaymentActivity 工作流活动开始处理订单 f4e1926e-3721-478d-be8a-f5bebd1995da 的付款，并确认是否成功。
  6. UpdateInventoryActivity 工作流活动在订单处理完成后，更新库存中当前可用的汽车。
  7. NotifyActivity 工作流活动发送通知，说明该订单 f4e1926e-3721-478d-be8a-f5bebd1995da 已完成。
  8. 工作流程已完成终止。

order-processor/app.py

在应用程序的程序文件中：

  • 生成了唯一的工作流顺序 ID
  • 工作流程已安排
  • 检索工作流状态
  • 工作流程及其调用的工作流程活动已注册
  1. class WorkflowConsoleApp:    
  2.     def main(self):
  3.         # Register workflow and activities
  4.         workflowRuntime = WorkflowRuntime(settings.DAPR_RUNTIME_HOST, settings.DAPR_GRPC_PORT)
  5.         workflowRuntime.register_workflow(order_processing_workflow)
  6.         workflowRuntime.register_activity(notify_activity)
  7.         workflowRuntime.register_activity(requst_approval_activity)
  8.         workflowRuntime.register_activity(verify_inventory_activity)
  9.         workflowRuntime.register_activity(process_payment_activity)
  10.         workflowRuntime.register_activity(update_inventory_activity)
  11.         workflowRuntime.start()
  13.         print("==========Begin the purchase of item:==========", flush=True)
  14.         item_name = default_item_name
  15.         order_quantity = 10
  17.         total_cost = int(order_quantity) * baseInventory[item_name].per_item_cost
  18.         order = OrderPayload(item_name=item_name, quantity=int(order_quantity), total_cost=total_cost)
  20.         # Start Workflow
  21.         print(f'Starting order workflow, purchasing {order_quantity} of {item_name}', flush=True)
  22.         start_resp = daprClient.start_workflow(workflow_component=workflow_component,
  23.                                                workflow_name=workflow_name,
  24.                                                input=order)
  25.         _id = start_resp.instance_id
  27.         def prompt_for_approval(daprClient: DaprClient):
  28.             daprClient.raise_workflow_event(instance_id=_id, workflow_component=workflow_component, 
  29.                                             event_name="manager_approval", event_data={'approval': True})
  31.         approval_seeked = False
  32.         start_time = datetime.now()
  33.         while True:
  34.             time_delta = datetime.now() - start_time
  35.             state = daprClient.get_workflow(instance_id=_id, workflow_component=workflow_component)
  36.             if not state:
  37.                 print("Workflow not found!")  # not expected
  38.             elif state.runtime_status == "Completed" or\
  39.                     state.runtime_status == "Failed" or\
  40.                     state.runtime_status == "Terminated":
  41.                 print(f'Workflow completed! Result: {state.runtime_status}', flush=True)
  42.                 break
  43.             if time_delta.total_seconds() >= 10:
  44.                 state = daprClient.get_workflow(instance_id=_id, workflow_component=workflow_component)
  45.                 if total_cost > 50000 and (
  46.                     state.runtime_status != "Completed" or 
  47.                     state.runtime_status != "Failed" or
  48.                     state.runtime_status != "Terminated"
  49.                     ) and not approval_seeked:
  50.                     approval_seeked = True
  51.                     threading.Thread(target=prompt_for_approval(daprClient), daemon=True).start()
  53.         print("Purchase of item is ", state.runtime_status, flush=True)
  55.     def restock_inventory(self, daprClient: DaprClient, baseInventory):
  56.         for key, item in baseInventory.items():
  57.             print(f'item: {item}')
  58.             item_str = f'{{"name": "{item.item_name}", "quantity": {item.quantity},\
  59.                           "per_item_cost": {item.per_item_cost}}}'
  60.             daprClient.save_state("statestore-actors", key, item_str)
  62. if __name__ == '__main__':
  63.     app = WorkflowConsoleApp()
  64.     app.main()

order-processor/workflow.py

在workflow.py中，工作流被定义为一个类，其中包含所有相关任务（由工作流活动确定）。

  1.   def order_processing_workflow(ctx: DaprWorkflowContext, order_payload_str: OrderPayload):
  2.     """Defines the order processing workflow.
  3.     When the order is received, the inventory is checked to see if there is enough inventory to
  4.     fulfill the order. If there is enough inventory, the payment is processed and the inventory is
  5.     updated. If there is not enough inventory, the order is rejected.
  6.     If the total order is greater than $50,000, the order is sent to a manager for approval.
  7.     """
  8.     order_id = ctx.instance_id
  9.     order_payload=json.loads(order_payload_str)
  10.     yield ctx.call_activity(notify_activity, 
  11.                             input=Notification(message=('Received order ' +order_id+ ' for '
  12.                                                +f'{order_payload["quantity"]}' +' ' +f'{order_payload["item_name"]}'
  13.                                                +' at $'+f'{order_payload["total_cost"]}' +' !')))
  14.     result = yield ctx.call_activity(verify_inventory_activity,
  15.                                      input=InventoryRequest(request_id=order_id,
  16.                                                             item_name=order_payload["item_name"],
  17.                                                             quantity=order_payload["quantity"]))
  18.     if not result.success:
  19.         yield ctx.call_activity(notify_activity,
  20.                                 input=Notification(message='Insufficient inventory for '
  21.                                                    +f'{order_payload["item_name"]}'+'!'))
  22.         return OrderResult(processed=False)
  24.     if order_payload["total_cost"] > 50000:
  25.         yield ctx.call_activity(requst_approval_activity, input=order_payload)
  26.         approval_task = ctx.wait_for_external_event("manager_approval")
  27.         timeout_event = ctx.create_timer(timedelta(seconds=200))
  28.         winner = yield when_any([approval_task, timeout_event])
  29.         if winner == timeout_event:
  30.             yield ctx.call_activity(notify_activity, 
  31.                                     input=Notification(message='Payment for order '+order_id
  32.                                                        +' has been cancelled due to timeout!'))
  33.             return OrderResult(processed=False)
  34.         approval_result = yield approval_task
  35.         if approval_result["approval"]:
  36.             yield ctx.call_activity(notify_activity, input=Notification(
  37.                 message=f'Payment for order {order_id} has been approved!'))
  38.         else:
  39.             yield ctx.call_activity(notify_activity, input=Notification(
  40.                 message=f'Payment for order {order_id} has been rejected!'))
  41.             return OrderResult(processed=False)    
  43.     yield ctx.call_activity(process_payment_activity, input=PaymentRequest(
  44.         request_id=order_id, item_being_purchased=order_payload["item_name"],
  45.         amount=order_payload["total_cost"], quantity=order_payload["quantity"]))
  47.     try:
  48.         yield ctx.call_activity(update_inventory_activity, 
  49.                                 input=PaymentRequest(request_id=order_id,
  50.                                                      item_being_purchased=order_payload["item_name"],
  51.                                                      amount=order_payload["total_cost"],
  52.                                                      quantity=order_payload["quantity"]))
  53.     except Exception:
  54.         yield ctx.call_activity(notify_activity, 
  55.                                 input=Notification(message=f'Order {order_id} Failed!'))
  56.         return OrderResult(processed=False)
  58.     yield ctx.call_activity(notify_activity, input=Notification(
  59.         message=f'Order {order_id} has completed!'))
  60.     return OrderResult(processed=True)

order-processor 控制台应用启动并管理订单处理工作流的生命周期，该工作流在状态存储中存储和检索数据。 工作流由四个工作流活动或任务组成:

  • notifyActivity: 利用记录器在整个工作流中打印出消息。 这些消息会在库存不足、付款无法处理等情况下通知用户。
  • ReserveInventoryActivity：检查状态存储以确保购买所需的库存足够。
  • RequestApprovalActivity：请求批准超过一定阈值的订单
  • processPaymentActivity: 处理并授权支付。
  • updateInventoryActivity: 使用新的剩余库存值更新状态存储。

第1步：先决条件

对于此示例，您将需要：

第2步：设置环境

克隆在Quickstarts存储库中提供的示例。

  1. git clone https://github.com/dapr/quickstarts.git

在一个新的终端窗口中，导航到 order-processor 目录：

  1. cd workflows/javascript/sdk/order-processor

安装依赖项：

  1. cd ./javascript/sdk
  2. npm install
  3. npm run build
  4. cd ..

第3步：运行订单处理程序应用

在终端中，使用 Multi-App Run 启动订单处理应用程序，并与一个 Dapr sidecar 并行运行：

  1. dapr run -f .

这将使用唯一的工作流ID启动order-processor应用程序，并运行工作流活动。

预期输出：

  1. == APP - workflowApp == == APP == Orchestration scheduled with ID: 0c332155-1e02-453a-a333-28cfc7777642
  2. == APP - workflowApp == == APP == Waiting 30 seconds for instance 0c332155-1e02-453a-a333-28cfc7777642 to complete...
  3. == APP - workflowApp == == APP == Received "Orchestrator Request" work item with instance id '0c332155-1e02-453a-a333-28cfc7777642'
  4. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Rebuilding local state with 0 history event...
  5. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, EXECUTIONSTARTED=1]
  6. == APP - workflowApp == == APP == Processing order 0c332155-1e02-453a-a333-28cfc7777642...
  7. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Waiting for 1 task(s) and 0 event(s) to complete...
  8. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Returning 1 action(s)
  9. == APP - workflowApp == == APP == Received "Activity Request" work item
  10. == APP - workflowApp == == APP == Received order 0c332155-1e02-453a-a333-28cfc7777642 for 10 item1 at a total cost of 100
  11. == APP - workflowApp == == APP == Activity notifyActivity completed with output undefined (0 chars)
  12. == APP - workflowApp == == APP == Received "Orchestrator Request" work item with instance id '0c332155-1e02-453a-a333-28cfc7777642'
  13. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Rebuilding local state with 3 history event...
  14. == APP - workflowApp == == APP == Processing order 0c332155-1e02-453a-a333-28cfc7777642...
  15. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  16. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Waiting for 1 task(s) and 0 event(s) to complete...
  17. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Returning 1 action(s)
  18. == APP - workflowApp == == APP == Received "Activity Request" work item
  19. == APP - workflowApp == == APP == Reserving inventory for 0c332155-1e02-453a-a333-28cfc7777642 of 10 item1
  20. == APP - workflowApp == == APP == 2024-02-16T03:15:59.498Z INFO [HTTPClient, HTTPClient] Sidecar Started
  21. == APP - workflowApp == == APP == There are 100 item1 in stock
  22. == APP - workflowApp == == APP == Activity reserveInventoryActivity completed with output {"success":true,"inventoryItem":{"perItemCost":100,"quantity":100,"itemName":"item1"}} (86 chars)
  23. == APP - workflowApp == == APP == Received "Orchestrator Request" work item with instance id '0c332155-1e02-453a-a333-28cfc7777642'
  24. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Rebuilding local state with 6 history event...
  25. == APP - workflowApp == == APP == Processing order 0c332155-1e02-453a-a333-28cfc7777642...
  26. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  27. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Waiting for 1 task(s) and 0 event(s) to complete...
  28. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Returning 1 action(s)
  29. == APP - workflowApp == == APP == Received "Activity Request" work item
  30. == APP - workflowApp == == APP == Processing payment for order item1
  31. == APP - workflowApp == == APP == Payment of 100 for 10 item1 processed successfully
  32. == APP - workflowApp == == APP == Activity processPaymentActivity completed with output true (4 chars)
  33. == APP - workflowApp == == APP == Received "Orchestrator Request" work item with instance id '0c332155-1e02-453a-a333-28cfc7777642'
  34. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Rebuilding local state with 9 history event...
  35. == APP - workflowApp == == APP == Processing order 0c332155-1e02-453a-a333-28cfc7777642...
  36. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  37. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Waiting for 1 task(s) and 0 event(s) to complete...
  38. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Returning 1 action(s)
  39. == APP - workflowApp == == APP == Received "Activity Request" work item
  40. == APP - workflowApp == == APP == Updating inventory for 0c332155-1e02-453a-a333-28cfc7777642 of 10 item1
  41. == APP - workflowApp == == APP == Inventory updated for 0c332155-1e02-453a-a333-28cfc7777642, there are now 90 item1 in stock
  42. == APP - workflowApp == == APP == Activity updateInventoryActivity completed with output {"success":true,"inventoryItem":{"perItemCost":100,"quantity":90,"itemName":"item1"}} (85 chars)
  43. == APP - workflowApp == == APP == Received "Orchestrator Request" work item with instance id '0c332155-1e02-453a-a333-28cfc7777642'
  44. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Rebuilding local state with 12 history event...
  45. == APP - workflowApp == == APP == Processing order 0c332155-1e02-453a-a333-28cfc7777642...
  46. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  47. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Waiting for 1 task(s) and 0 event(s) to complete...
  48. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Returning 1 action(s)
  49. == APP - workflowApp == == APP == Received "Activity Request" work item
  50. == APP - workflowApp == == APP == order 0c332155-1e02-453a-a333-28cfc7777642 processed successfully!
  51. == APP - workflowApp == == APP == Activity notifyActivity completed with output undefined (0 chars)
  52. == APP - workflowApp == == APP == Received "Orchestrator Request" work item with instance id '0c332155-1e02-453a-a333-28cfc7777642'
  53. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Rebuilding local state with 15 history event...
  54. == APP - workflowApp == == APP == Processing order 0c332155-1e02-453a-a333-28cfc7777642...
  55. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  56. == APP - workflowApp == == APP == Order 0c332155-1e02-453a-a333-28cfc7777642 processed successfully!
  57. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Orchestration completed with status COMPLETED
  58. == APP - workflowApp == == APP == 0c332155-1e02-453a-a333-28cfc7777642: Returning 1 action(s)
  59. == APP - workflowApp == time="2024-02-15T21:15:59.5589687-06:00" level=info msg="0c332155-1e02-453a-a333-28cfc7777642: 'orderProcessingWorkflow' completed with a COMPLETED status." app_id=activity-sequence-workflow instance=kaibocai-devbox scope=wfengine.backend type=log ver=1.12.4
  60. == APP - workflowApp == == APP == Instance 0c332155-1e02-453a-a333-28cfc7777642 completed

（可选）第4步：在Zipkin中查看

运行 dapr init 启动 openzipkin/zipkin Docker 容器。 如果容器已停止运行，请使用以下命令启动Zipkin Docker容器：

  1. docker run -d -p 9411:9411 openzipkin/zipkin

在 Zipkin web UI 中查看工作流 trace span（通常在 http://localhost:9411/zipkin/）。

Workflow - 图3

发生了什么？

当你运行 dapr run -f .：

  1. 生成一个唯一的工作流订单ID（在上面的示例中，0c332155-1e02-453a-a333-28cfc7777642）并安排工作流。
  2. notifyActivity 工作流活动发送通知，说明已收到一份包含10辆车的订单。
  3. reserveInventoryActivity工作流活动检查库存数据，确定是否可以提供订购的物料，并使用库存中的汽车数量进行响应。
  4. 您的工作流程开始并通知您其状态。
  5. processPaymentActivity 工作流活动开始处理订单 0c332155-1e02-453a-a333-28cfc7777642 的付款，并确认是否成功。
  6. updateInventoryActivity 工作流活动在订单处理完成后，更新库存中当前可用的汽车。
  7. notifyActivity 工作流活动发送通知，说明该订单 0c332155-1e02-453a-a333-28cfc7777642 已完成。
  8. 工作流程已完成终止。

order-processor/workflowApp.ts

在应用程序的文件中：

  • 生成了唯一的工作流顺序 ID
  • 工作流程已安排
  • 检索工作流状态
  • 工作流程及其调用的工作流程活动已注册
  1. import { DaprWorkflowClient, WorkflowRuntime, DaprClient } from "@dapr/dapr-dev";
  2. import { InventoryItem, OrderPayload } from "./model";
  3. import { notifyActivity, orderProcessingWorkflow, processPaymentActivity, requestApprovalActivity, reserveInventoryActivity, updateInventoryActivity } from "./orderProcessingWorkflow";
  5. async function start() {
  6.   // Update the gRPC client and worker to use a local address and port
  7.   const workflowClient = new DaprWorkflowClient();
  8.   const workflowWorker = new WorkflowRuntime();
  10.   const daprClient = new DaprClient();
  11.   const storeName = "statestore";
  13.   const inventory = new InventoryItem("item1", 100, 100);
  14.   const key = inventory.itemName;
  16.   await daprClient.state.save(storeName, [
  17.     {
  18.       key: key,
  19.       value: inventory,
  20.     }
  21.   ]);
  23.   const order = new OrderPayload("item1", 100, 10);
  25.   workflowWorker
  26.   .registerWorkflow(orderProcessingWorkflow)
  27.   .registerActivity(notifyActivity)
  28.   .registerActivity(reserveInventoryActivity)
  29.   .registerActivity(requestApprovalActivity)
  30.   .registerActivity(processPaymentActivity)
  31.   .registerActivity(updateInventoryActivity);
  33.   // Wrap the worker startup in a try-catch block to handle any errors during startup
  34.   try {
  35.     await workflowWorker.start();
  36.     console.log("Workflow runtime started successfully");
  37.   } catch (error) {
  38.     console.error("Error starting workflow runtime:", error);
  39.   }
  41.   // Schedule a new orchestration
  42.   try {
  43.     const id = await workflowClient.scheduleNewWorkflow(orderProcessingWorkflow, order);
  44.     console.log(`Orchestration scheduled with ID: ${id}`);
  46.     // Wait for orchestration completion
  47.     const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);
  49.     console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
  50.   } catch (error) {
  51.     console.error("Error scheduling or waiting for orchestration:", error);
  52.     throw error;
  53.   }
  55.   await workflowWorker.stop();
  56.   await workflowClient.stop();
  57. }
  59. start().catch((e) => {
  60.   console.error(e);
  61.   process.exit(1);
  62. });

order-processor 控制台应用启动并管理订单处理工作流的生命周期，该工作流在状态存储中存储和检索数据。 工作流由四个工作流活动或任务组成:

  • NotifyActivity: 利用记录器在整个工作流中打印出消息
  • ReserveInventoryActivity: 检查状态存储以确保购买时有足够的库存
  • ProcessPaymentActivity: 处理并授权付款
  • UpdateInventoryActivity: 从状态存储中删除请求的项目，并使用新的剩余库存值更新存储

第1步：先决条件

对于此示例，您将需要：

第2步：设置环境

克隆在Quickstarts存储库中提供的示例。

  1. git clone https://github.com/dapr/quickstarts.git

在一个新的终端窗口中，导航到 order-processor 目录：

  1. cd workflows/csharp/sdk/order-processor

第3步：运行订单处理程序应用

在终端中，使用 Multi-App Run 启动订单处理应用程序，并与一个 Dapr sidecar 并行运行：

  1. dapr run -f .

这将使用唯一的工作流ID启动order-processor应用程序，并运行工作流活动。

预期输出：

  1. == APP == Starting workflow 6d2abcc9 purchasing 10 Cars
  3. == APP == info: Microsoft.DurableTask.Client.Grpc.GrpcDurableTaskClient[40]
  4. == APP ==       Scheduling new OrderProcessingWorkflow orchestration with instance ID '6d2abcc9' and 47 bytes of input data.
  5. == APP == info: WorkflowConsoleApp.Activities.NotifyActivity[0]
  6. == APP ==       Received order 6d2abcc9 for 10 Cars at $15000
  7. == APP == info: WorkflowConsoleApp.Activities.ReserveInventoryActivity[0]
  8. == APP ==       Reserving inventory for order 6d2abcc9 of 10 Cars
  9. == APP == info: WorkflowConsoleApp.Activities.ReserveInventoryActivity[0]
  10. == APP ==       There are: 100, Cars available for purchase
  12. == APP == Your workflow has started. Here is the status of the workflow: Dapr.Workflow.WorkflowState
  14. == APP == info: WorkflowConsoleApp.Activities.ProcessPaymentActivity[0]
  15. == APP ==       Processing payment: 6d2abcc9 for 10 Cars at $15000
  16. == APP == info: WorkflowConsoleApp.Activities.ProcessPaymentActivity[0]
  17. == APP ==       Payment for request ID '6d2abcc9' processed successfully
  18. == APP == info: WorkflowConsoleApp.Activities.UpdateInventoryActivity[0]
  19. == APP ==       Checking Inventory for: Order# 6d2abcc9 for 10 Cars
  20. == APP == info: WorkflowConsoleApp.Activities.UpdateInventoryActivity[0]
  21. == APP ==       There are now: 90 Cars left in stock
  22. == APP == info: WorkflowConsoleApp.Activities.NotifyActivity[0]
  23. == APP ==       Order 6d2abcc9 has completed!
  25. == APP == Workflow Status: Completed

（可选）第4步：在Zipkin中查看

运行 dapr init 启动 openzipkin/zipkin Docker 容器。 如果容器已停止运行，请使用以下命令启动Zipkin Docker容器：

  1. docker run -d -p 9411:9411 openzipkin/zipkin

在 Zipkin web UI 中查看工作流 trace span（通常在 http://localhost:9411/zipkin/）。

Workflow - 图4

发生了什么？

当你运行 dapr run -f .：

  1. 生成一个唯一的工作流订单ID（在上面的示例中，6d2abcc9）并安排工作流。
  2. NotifyActivity 工作流活动发送通知，说明已收到一份包含10辆车的订单。
  3. ReserveInventoryActivity工作流活动检查库存数据，确定是否可以提供订购的物料，并使用库存中的汽车数量进行响应。
  4. 您的工作流程开始并通知您其状态。
  5. ProcessPaymentActivity 工作流活动开始处理订单 6d2abcc9 的付款，并确认是否成功。
  6. UpdateInventoryActivity 工作流活动在订单处理完成后，更新库存中当前可用的汽车。
  7. NotifyActivity 工作流活动发送通知，表示订单 6d2abcc9 已完成。
  8. 工作流程已完成终止。

order-processor/Program.cs

在应用程序的程序文件中：

  • 生成了唯一的工作流顺序 ID
  • 工作流程已安排
  • 检索工作流状态
  • 工作流程及其调用的工作流程活动已注册
  1. using Dapr.Client;
  2. using Dapr.Workflow;
  3. //...
  5. {
  6.     services.AddDaprWorkflow(options =>
  7.     {
  8.         // Note that it's also possible to register a lambda function as the workflow
  9.         // or activity implementation instead of a class.
  10.         options.RegisterWorkflow<OrderProcessingWorkflow>();
  12.         // These are the activities that get invoked by the workflow(s).
  13.         options.RegisterActivity<NotifyActivity>();
  14.         options.RegisterActivity<ReserveInventoryActivity>();
  15.         options.RegisterActivity<ProcessPaymentActivity>();
  16.         options.RegisterActivity<UpdateInventoryActivity>();
  17.     });
  18. };
  20. //...
  22. // Generate a unique ID for the workflow
  23. string orderId = Guid.NewGuid().ToString()[..8];
  24. string itemToPurchase = "Cars";
  25. int ammountToPurchase = 10;
  27. // Construct the order
  28. OrderPayload orderInfo = new OrderPayload(itemToPurchase, 15000, ammountToPurchase);
  30. // Start the workflow
  31. Console.WriteLine("Starting workflow {0} purchasing {1} {2}", orderId, ammountToPurchase, itemToPurchase);
  33. await daprClient.StartWorkflowAsync(
  34.     workflowComponent: DaprWorkflowComponent,
  35.     workflowName: nameof(OrderProcessingWorkflow),
  36.     input: orderInfo,
  37.     instanceId: orderId);
  39. // Wait for the workflow to start and confirm the input
  40. GetWorkflowResponse state = await daprClient.WaitForWorkflowStartAsync(
  41.     instanceId: orderId,
  42.     workflowComponent: DaprWorkflowComponent);
  44. Console.WriteLine("Your workflow has started. Here is the status of the workflow: {0}", state.RuntimeStatus);
  46. // Wait for the workflow to complete
  47. state = await daprClient.WaitForWorkflowCompletionAsync(
  48.     instanceId: orderId,
  49.     workflowComponent: DaprWorkflowComponent);
  51. Console.WriteLine("Workflow Status: {0}", state.RuntimeStatus);

order-processor/Workflows/OrderProcessingWorkflow.cs

在OrderProcessingWorkflow.cs中，工作流被定义为一个类，其中包含所有相关任务（由工作流活动确定）。

  1. using Dapr.Workflow;
  2. //...
  4. class OrderProcessingWorkflow : Workflow<OrderPayload, OrderResult>
  5.     {
  6.         public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderPayload order)
  7.         {
  8.             string orderId = context.InstanceId;
  10.             // Notify the user that an order has come through
  11.             await context.CallActivityAsync(
  12.                 nameof(NotifyActivity),
  13.                 new Notification($"Received order {orderId} for {order.Quantity} {order.Name} at ${order.TotalCost}"));
  15.             string requestId = context.InstanceId;
  17.             // Determine if there is enough of the item available for purchase by checking the inventory
  18.             InventoryResult result = await context.CallActivityAsync<InventoryResult>(
  19.                 nameof(ReserveInventoryActivity),
  20.                 new InventoryRequest(RequestId: orderId, order.Name, order.Quantity));
  22.             // If there is insufficient inventory, fail and let the user know 
  23.             if (!result.Success)
  24.             {
  25.                 // End the workflow here since we don't have sufficient inventory
  26.                 await context.CallActivityAsync(
  27.                     nameof(NotifyActivity),
  28.                     new Notification($"Insufficient inventory for {order.Name}"));
  29.                 return new OrderResult(Processed: false);
  30.             }
  32.             // There is enough inventory available so the user can purchase the item(s). Process their payment
  33.             await context.CallActivityAsync(
  34.                 nameof(ProcessPaymentActivity),
  35.                 new PaymentRequest(RequestId: orderId, order.Name, order.Quantity, order.TotalCost));
  37.             try
  38.             {
  39.                 // There is enough inventory available so the user can purchase the item(s). Process their payment
  40.                 await context.CallActivityAsync(
  41.                     nameof(UpdateInventoryActivity),
  42.                     new PaymentRequest(RequestId: orderId, order.Name, order.Quantity, order.TotalCost));                
  43.             }
  44.             catch (TaskFailedException)
  45.             {
  46.                 // Let them know their payment was processed
  47.                 await context.CallActivityAsync(
  48.                     nameof(NotifyActivity),
  49.                     new Notification($"Order {orderId} Failed! You are now getting a refund"));
  50.                 return new OrderResult(Processed: false);
  51.             }
  53.             // Let them know their payment was processed
  54.             await context.CallActivityAsync(
  55.                 nameof(NotifyActivity),
  56.                 new Notification($"Order {orderId} has completed!"));
  58.             // End the workflow with a success result
  59.             return new OrderResult(Processed: true);
  60.         }
  61.     }

order-processor/Activities 文件夹

Activities目录包含工作流使用的四个工作流活动，这些活动在以下文件中定义：

  • NotifyActivity.cs
  • ReserveInventoryActivity.cs
  • ProcessPaymentActivity.cs
  • UpdateInventoryActivity.cs

观看演示

观看此视频演练Dapr Workflow .NET演示:

order-processor 控制台应用启动并管理订单处理工作流的生命周期，该工作流在状态存储中存储和检索数据。 工作流由四个工作流活动或任务组成:

  • NotifyActivity: 利用记录器在整个工作流中打印出消息
  • RequestApprovalActivity：请求批准处理付款
  • ReserveInventoryActivity: 检查状态存储以确保购买时有足够的库存
  • ProcessPaymentActivity: 处理并授权付款
  • UpdateInventoryActivity: 从状态存储中删除请求的项目，并使用新的剩余库存值更新存储

第1步：先决条件

对于此示例，您将需要：

第2步：设置环境

克隆在Quickstarts存储库中提供的示例。

  1. git clone https://github.com/dapr/quickstarts.git

导航到 order-processor 目录：

  1. cd workflows/java/sdk/order-processor

安装依赖项：

  1. mvn clean install

第3步：运行订单处理程序应用

在终端中，使用 Multi-App Run 启动订单处理应用程序，并与一个 Dapr sidecar 并行运行：

  1. dapr run -f .

这将使用唯一的工作流ID启动order-processor应用程序，并运行工作流活动。

预期输出：

  1. == APP == *** Welcome to the Dapr Workflow console app sample!
  2. == APP == *** Using this app, you can place orders that start workflows.
  3. == APP == Start workflow runtime
  4. == APP == Sep 20, 2023 3:23:05 PM com.microsoft.durabletask.DurableTaskGrpcWorker startAndBlock
  5. == APP == INFO: Durable Task worker is connecting to sidecar at 127.0.0.1:50001.
  7. == APP == ==========Begin the purchase of item:==========
  8. == APP == Starting order workflow, purchasing 10 of cars
  10. == APP == scheduled new workflow instance of OrderProcessingWorkflow with instance ID: edceba90-9c45-4be8-ad40-60d16e060797
  11. == APP == [Thread-0] INFO io.dapr.workflows.WorkflowContext - Starting Workflow: io.dapr.quickstarts.workflows.OrderProcessingWorkflow
  12. == APP == [Thread-0] INFO io.dapr.workflows.WorkflowContext - Instance ID(order ID): edceba90-9c45-4be8-ad40-60d16e060797
  13. == APP == [Thread-0] INFO io.dapr.workflows.WorkflowContext - Current Orchestration Time: 2023-09-20T19:23:09.755Z
  14. == APP == [Thread-0] INFO io.dapr.workflows.WorkflowContext - Received Order: OrderPayload [itemName=cars, totalCost=150000, quantity=10]
  15. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.NotifyActivity - Received Order: OrderPayload [itemName=cars, totalCost=150000, quantity=10]
  16. == APP == workflow instance edceba90-9c45-4be8-ad40-60d16e060797 started
  17. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.ReserveInventoryActivity - Reserving inventory for order 'edceba90-9c45-4be8-ad40-60d16e060797' of 10 cars
  18. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.ReserveInventoryActivity - There are 100 cars available for purchase
  19. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.ReserveInventoryActivity - Reserved inventory for order 'edceba90-9c45-4be8-ad40-60d16e060797' of 10 cars
  20. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.RequestApprovalActivity - Requesting approval for order: OrderPayload [itemName=cars, totalCost=150000, quantity=10]
  21. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.RequestApprovalActivity - Approved requesting approval for order: OrderPayload [itemName=cars, totalCost=150000, quantity=10]
  22. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.ProcessPaymentActivity - Processing payment: edceba90-9c45-4be8-ad40-60d16e060797 for 10 cars at $150000
  23. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.ProcessPaymentActivity - Payment for request ID 'edceba90-9c45-4be8-ad40-60d16e060797' processed successfully
  24. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.UpdateInventoryActivity - Updating inventory for order 'edceba90-9c45-4be8-ad40-60d16e060797' of 10 cars
  25. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.UpdateInventoryActivity - Updated inventory for order 'edceba90-9c45-4be8-ad40-60d16e060797': there are now 90 cars left in stock
  26. == APP == [Thread-0] INFO io.dapr.quickstarts.workflows.activities.NotifyActivity - Order completed! : edceba90-9c45-4be8-ad40-60d16e060797
  28. == APP == workflow instance edceba90-9c45-4be8-ad40-60d16e060797 completed, out is: {"processed":true}

（可选）第4步：在Zipkin中查看

运行 dapr init 启动 openzipkin/zipkin Docker 容器。 如果容器已停止运行，请使用以下命令启动Zipkin Docker容器：

  1. docker run -d -p 9411:9411 openzipkin/zipkin

在 Zipkin web UI 中查看工作流 trace span（通常在 http://localhost:9411/zipkin/）。

Workflow - 图5

发生了什么？

当你运行 dapr run -f .：

  1. 生成一个唯一的工作流订单ID（在上面的示例中，edceba90-9c45-4be8-ad40-60d16e060797）并安排工作流。
  2. NotifyActivity 工作流活动发送通知，说明已收到一份包含10辆车的订单。
  3. ReserveInventoryActivity工作流活动检查库存数据，确定是否可以提供订购的物料，并使用库存中的汽车数量进行响应。
  4. 一旦批准，您的工作流程开始并通知您其状态。
  5. ProcessPaymentActivity 工作流活动开始处理订单 edceba90-9c45-4be8-ad40-60d16e060797 的付款，并确认是否成功。
  6. UpdateInventoryActivity 工作流活动在订单处理完成后，更新库存中当前可用的汽车。
  7. NotifyActivity 工作流活动发送通知，说明该订单 edceba90-9c45-4be8-ad40-60d16e060797 已完成。
  8. 工作流程已完成终止。

order-processor/WorkflowConsoleApp.java

在应用程序的程序文件中：

  • 生成了唯一的工作流顺序 ID
  • 工作流程已安排
  • 检索工作流状态
  • 工作流程及其调用的工作流程活动已注册
  1. package io.dapr.quickstarts.workflows;
  2. import io.dapr.client.DaprClient;
  3. import io.dapr.client.DaprClientBuilder;
  4. import io.dapr.workflows.client.DaprWorkflowClient;
  6. public class WorkflowConsoleApp {
  8.   private static final String STATE_STORE_NAME = "statestore-actors";
  10.   // ...
  11.   public static void main(String[] args) throws Exception {
  12.     System.out.println("*** Welcome to the Dapr Workflow console app sample!");
  13.     System.out.println("*** Using this app, you can place orders that start workflows.");
  14.     // Wait for the sidecar to become available
  15.     Thread.sleep(5 * 1000);
  17.     // Register the OrderProcessingWorkflow and its activities with the builder.
  18.     WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().registerWorkflow(OrderProcessingWorkflow.class);
  19.     builder.registerActivity(NotifyActivity.class);
  20.     builder.registerActivity(ProcessPaymentActivity.class);
  21.     builder.registerActivity(RequestApprovalActivity.class);
  22.     builder.registerActivity(ReserveInventoryActivity.class);
  23.     builder.registerActivity(UpdateInventoryActivity.class);
  25.     // Build the workflow runtime
  26.     try (WorkflowRuntime runtime = builder.build()) {
  27.       System.out.println("Start workflow runtime");
  28.       runtime.start(false);
  29.     }
  31.     InventoryItem inventory = prepareInventoryAndOrder();
  33.     DaprWorkflowClient workflowClient = new DaprWorkflowClient();
  34.     try (workflowClient) {
  35.       executeWorkflow(workflowClient, inventory);
  36.     }
  38.   }
  40.   // Start the workflow runtime, pulling and executing tasks
  41.   private static void executeWorkflow(DaprWorkflowClient workflowClient, InventoryItem inventory) {
  42.     System.out.println("==========Begin the purchase of item:==========");
  43.     String itemName = inventory.getName();
  44.     int orderQuantity = inventory.getQuantity();
  45.     int totalcost = orderQuantity * inventory.getPerItemCost();
  46.     OrderPayload order = new OrderPayload();
  47.     order.setItemName(itemName);
  48.     order.setQuantity(orderQuantity);
  49.     order.setTotalCost(totalcost);
  50.     System.out.println("Starting order workflow, purchasing " + orderQuantity + " of " + itemName);
  52.     String instanceId = workflowClient.scheduleNewWorkflow(OrderProcessingWorkflow.class, order);
  53.     System.out.printf("scheduled new workflow instance of OrderProcessingWorkflow with instance ID: %s%n",
  54.         instanceId);
  56.     // Check workflow instance start status
  57.     try {
  58.       workflowClient.waitForInstanceStart(instanceId, Duration.ofSeconds(10), false);
  59.       System.out.printf("workflow instance %s started%n", instanceId);
  60.     } catch (TimeoutException e) {
  61.       System.out.printf("workflow instance %s did not start within 10 seconds%n", instanceId);
  62.       return;
  63.     }
  65.     // Check workflow instance complete status
  66.     try {
  67.       WorkflowInstanceStatus workflowStatus = workflowClient.waitForInstanceCompletion(instanceId,
  68.           Duration.ofSeconds(30),
  69.           true);
  70.       if (workflowStatus != null) {
  71.         System.out.printf("workflow instance %s completed, out is: %s %n", instanceId,
  72.             workflowStatus.getSerializedOutput());
  73.       } else {
  74.         System.out.printf("workflow instance %s not found%n", instanceId);
  75.       }
  76.     } catch (TimeoutException e) {
  77.       System.out.printf("workflow instance %s did not complete within 30 seconds%n", instanceId);
  78.     }
  80.   }
  82.   private static InventoryItem prepareInventoryAndOrder() {
  83.     // prepare 100 cars in inventory
  84.     InventoryItem inventory = new InventoryItem();
  85.     inventory.setName("cars");
  86.     inventory.setPerItemCost(15000);
  87.     inventory.setQuantity(100);
  88.     DaprClient daprClient = new DaprClientBuilder().build();
  89.     restockInventory(daprClient, inventory);
  91.     // prepare order for 10 cars
  92.     InventoryItem order = new InventoryItem();
  93.     order.setName("cars");
  94.     order.setPerItemCost(15000);
  95.     order.setQuantity(10);
  96.     return order;
  97.   }
  99.   private static void restockInventory(DaprClient daprClient, InventoryItem inventory) {
  100.     String key = inventory.getName();
  101.     daprClient.saveState(STATE_STORE_NAME, key, inventory).block();
  102.   }
  103. }

OrderProcessingWorkflow.java

在OrderProcessingWorkflow.java中，工作流被定义为一个类，其中包含所有相关任务（由工作流活动确定）。

  1. package io.dapr.quickstarts.workflows;
  2. import io.dapr.workflows.Workflow;
  4. public class OrderProcessingWorkflow extends Workflow {
  6.   @Override
  7.   public WorkflowStub create() {
  8.     return ctx -> {
  9.       Logger logger = ctx.getLogger();
  10.       String orderId = ctx.getInstanceId();
  11.       logger.info("Starting Workflow: " + ctx.getName());
  12.       logger.info("Instance ID(order ID): " + orderId);
  13.       logger.info("Current Orchestration Time: " + ctx.getCurrentInstant());
  15.       OrderPayload order = ctx.getInput(OrderPayload.class);
  16.       logger.info("Received Order: " + order.toString());
  17.       OrderResult orderResult = new OrderResult();
  18.       orderResult.setProcessed(false);
  20.       // Notify the user that an order has come through
  21.       Notification notification = new Notification();
  22.       notification.setMessage("Received Order: " + order.toString());
  23.       ctx.callActivity(NotifyActivity.class.getName(), notification).await();
  25.       // Determine if there is enough of the item available for purchase by checking
  26.       // the inventory
  27.       InventoryRequest inventoryRequest = new InventoryRequest();
  28.       inventoryRequest.setRequestId(orderId);
  29.       inventoryRequest.setItemName(order.getItemName());
  30.       inventoryRequest.setQuantity(order.getQuantity());
  31.       InventoryResult inventoryResult = ctx.callActivity(ReserveInventoryActivity.class.getName(),
  32.           inventoryRequest, InventoryResult.class).await();
  34.       // If there is insufficient inventory, fail and let the user know
  35.       if (!inventoryResult.isSuccess()) {
  36.         notification.setMessage("Insufficient inventory for order : " + order.getItemName());
  37.         ctx.callActivity(NotifyActivity.class.getName(), notification).await();
  38.         ctx.complete(orderResult);
  39.         return;
  40.       }
  42.       // Require orders over a certain threshold to be approved
  43.       if (order.getTotalCost() > 5000) {
  44.         ApprovalResult approvalResult = ctx.callActivity(RequestApprovalActivity.class.getName(),
  45.             order, ApprovalResult.class).await();
  46.         if (approvalResult != ApprovalResult.Approved) {
  47.           notification.setMessage("Order " + order.getItemName() + " was not approved.");
  48.           ctx.callActivity(NotifyActivity.class.getName(), notification).await();
  49.           ctx.complete(orderResult);
  50.           return;
  51.         }
  52.       }
  54.       // There is enough inventory available so the user can purchase the item(s).
  55.       // Process their payment
  56.       PaymentRequest paymentRequest = new PaymentRequest();
  57.       paymentRequest.setRequestId(orderId);
  58.       paymentRequest.setItemBeingPurchased(order.getItemName());
  59.       paymentRequest.setQuantity(order.getQuantity());
  60.       paymentRequest.setAmount(order.getTotalCost());
  61.       boolean isOK = ctx.callActivity(ProcessPaymentActivity.class.getName(),
  62.           paymentRequest, boolean.class).await();
  63.       if (!isOK) {
  64.         notification.setMessage("Payment failed for order : " + orderId);
  65.         ctx.callActivity(NotifyActivity.class.getName(), notification).await();
  66.         ctx.complete(orderResult);
  67.         return;
  68.       }
  70.       inventoryResult = ctx.callActivity(UpdateInventoryActivity.class.getName(),
  71.           inventoryRequest, InventoryResult.class).await();
  72.       if (!inventoryResult.isSuccess()) {
  73.         // If there is an error updating the inventory, refund the user
  74.         // paymentRequest.setAmount(-1 * paymentRequest.getAmount());
  75.         // ctx.callActivity(ProcessPaymentActivity.class.getName(),
  76.         // paymentRequest).await();
  78.         // Let users know their payment processing failed
  79.         notification.setMessage("Order failed to update inventory! : " + orderId);
  80.         ctx.callActivity(NotifyActivity.class.getName(), notification).await();
  81.         ctx.complete(orderResult);
  82.         return;
  83.       }
  85.       // Let user know their order was processed
  86.       notification.setMessage("Order completed! : " + orderId);
  87.       ctx.callActivity(NotifyActivity.class.getName(), notification).await();
  89.       // Complete the workflow with order result is processed
  90.       orderResult.setProcessed(true);
  91.       ctx.complete(orderResult);
  92.     };
  93.   }
  95. }

activities 目录

Activities目录包含工作流使用的四个工作流活动，这些活动在以下文件中定义：

order-processor 控制台应用程序启动和管理 OrderProcessingWorkflow 工作流，模拟从商店购买物品。 工作流由五个独特的工作流活动或任务组成：

  • NotifyActivity: 利用记录器在整个工作流中打印出消息。 这些消息在以下情况下通知您：
    • 您的库存不足
    • 您的付款无法处理等等。
  • ProcessPaymentActivity: 处理并授权付款.
  • VerifyInventoryActivity: 检查状态存储以确保有足够的库存可供购买。
  • UpdateInventoryActivity: 从状态存储中删除请求的项目，并使用新的剩余库存值更新存储。
  • RequestApprovalActivity: 如果付款金额超过50,000美元，则向经理请求批准。

第1步：先决条件

对于此示例，您将需要：

第2步：设置环境

克隆在Quickstarts存储库中提供的示例。

  1. git clone https://github.com/dapr/quickstarts.git

在一个新的终端窗口中，导航到 order-processor 目录：

  1. cd workflows/go/sdk/order-processor

第3步：运行订单处理程序应用

在终端中，使用 Multi-App Run 启动订单处理应用程序，并与一个 Dapr sidecar 并行运行：

  1. dapr run -f .

这将使用唯一的工作流ID启动order-processor应用程序，并运行工作流活动。

预期输出：

  1. == APP - order-processor == *** Welcome to the Dapr Workflow console app sample!
  2. == APP - order-processor == *** Using this app, you can place orders that start workflows.
  3. == APP - order-processor == dapr client initializing for: 127.0.0.1:50056
  4. == APP - order-processor == adding base stock item: paperclip
  5. == APP - order-processor == 2024/02/01 12:59:52 work item listener started
  6. == APP - order-processor == INFO: 2024/02/01 12:59:52 starting background processor
  7. == APP - order-processor == adding base stock item: cars
  8. == APP - order-processor == adding base stock item: computers
  9. == APP - order-processor == ==========Begin the purchase of item:==========
  10. == APP - order-processor == NotifyActivity: Received order 48ee83b7-5d80-48d5-97f9-6b372f5480a5 for 10 cars - $150000
  11. == APP - order-processor == VerifyInventoryActivity: Verifying inventory for order 48ee83b7-5d80-48d5-97f9-6b372f5480a5 of 10 cars
  12. == APP - order-processor == VerifyInventoryActivity: There are 100 cars available for purchase
  13. == APP - order-processor == RequestApprovalActivity: Requesting approval for payment of 150000USD for 10 cars
  14. == APP - order-processor == NotifyActivity: Payment for order 48ee83b7-5d80-48d5-97f9-6b372f5480a5 has been approved!
  15. == APP - order-processor == ProcessPaymentActivity: 48ee83b7-5d80-48d5-97f9-6b372f5480a5 for 10 - cars (150000USD)
  16. == APP - order-processor == UpdateInventoryActivity: Checking Inventory for order 48ee83b7-5d80-48d5-97f9-6b372f5480a5 for 10 * cars
  17. == APP - order-processor == UpdateInventoryActivity: There are now 90 cars left in stock
  18. == APP - order-processor == NotifyActivity: Order 48ee83b7-5d80-48d5-97f9-6b372f5480a5 has completed!
  19. == APP - order-processor == Workflow completed - result: COMPLETED
  20. == APP - order-processor == Purchase of item is complete

使用 CTRL+C 或以下命令停止 Dapr 工作流:

  1. dapr stop -f .

（可选）第4步：在Zipkin中查看

运行 dapr init 启动 openzipkin/zipkin Docker 容器。 如果容器已停止运行，请使用以下命令启动Zipkin Docker容器：

  1. docker run -d -p 9411:9411 openzipkin/zipkin

在 Zipkin web UI 中查看工作流 trace span（通常在 http://localhost:9411/zipkin/）。

Workflow - 图6

发生了什么？

当你运行 dapr run:

  1. 生成一个唯一的工作流订单ID（在上面的示例中，48ee83b7-5d80-48d5-97f9-6b372f5480a5）并安排工作流。
  2. NotifyActivity 工作流活动发送通知，说明已收到一份包含10辆车的订单。
  3. ReserveInventoryActivity工作流活动检查库存数据，确定是否可以提供订购的物料，并使用库存中的汽车数量进行响应。
  4. 您的工作流程开始并通知您其状态。
  5. ProcessPaymentActivity 工作流活动开始处理订单 48ee83b7-5d80-48d5-97f9-6b372f5480a5 的付款，并确认是否成功。
  6. UpdateInventoryActivity 工作流活动在订单处理完成后，更新库存中当前可用的汽车。
  7. NotifyActivity 工作流活动发送通知，说明该订单 48ee83b7-5d80-48d5-97f9-6b372f5480a5 已完成。
  8. 工作流程已完成终止。

order-processor/main.go

在应用程序的程序文件中：

  • 生成了唯一的工作流顺序 ID
  • 工作流程已安排
  • 检索工作流状态
  • 工作流程及其调用的工作流程活动已注册
  1. func main() {
  2.     fmt.Println("*** Welcome to the Dapr Workflow console app sample!")
  3.     fmt.Println("*** Using this app, you can place orders that start workflows.")
  5.   // ...
  7.   // Register workflow and activities
  8.     if err := w.RegisterWorkflow(OrderProcessingWorkflow); err != nil {
  9.         log.Fatal(err)
  10.     }
  11.     if err := w.RegisterActivity(NotifyActivity); err != nil {
  12.         log.Fatal(err)
  13.     }
  14.     if err := w.RegisterActivity(RequestApprovalActivity); err != nil {
  15.         log.Fatal(err)
  16.     }
  17.     if err := w.RegisterActivity(VerifyInventoryActivity); err != nil {
  18.         log.Fatal(err)
  19.     }
  20.     if err := w.RegisterActivity(ProcessPaymentActivity); err != nil {
  21.         log.Fatal(err)
  22.     }
  23.     if err := w.RegisterActivity(UpdateInventoryActivity); err != nil {
  24.         log.Fatal(err)
  25.     }
  27.   // Build and start workflow runtime, pulling and executing tasks
  28.     if err := w.Start(); err != nil {
  29.         log.Fatal(err)
  30.     }
  32.     daprClient, err := client.NewClient()
  33.     if err != nil {
  34.         log.Fatalf("failed to initialise dapr client: %v", err)
  35.     }
  36.     wfClient, err := workflow.NewClient(workflow.WithDaprClient(daprClient))
  37.     if err != nil {
  38.         log.Fatalf("failed to initialise workflow client: %v", err)
  39.     }
  41.   // Check inventory
  42.     inventory := []InventoryItem{
  43.         {ItemName: "paperclip", PerItemCost: 5, Quantity: 100},
  44.         {ItemName: "cars", PerItemCost: 15000, Quantity: 100},
  45.         {ItemName: "computers", PerItemCost: 500, Quantity: 100},
  46.     }
  47.     if err := restockInventory(daprClient, inventory); err != nil {
  48.         log.Fatalf("failed to restock: %v", err)
  49.     }
  51.     fmt.Println("==========Begin the purchase of item:==========")
  53.     itemName := defaultItemName
  54.     orderQuantity := 10
  56.     totalCost := inventory[1].PerItemCost * orderQuantity
  58.     orderPayload := OrderPayload{
  59.         ItemName:  itemName,
  60.         Quantity:  orderQuantity,
  61.         TotalCost: totalCost,
  62.     }
  64.   // Start workflow events, like receiving order, verifying inventory, and processing payment 
  65.     id, err := wfClient.ScheduleNewWorkflow(context.Background(), workflowName, workflow.WithInput(orderPayload))
  66.     if err != nil {
  67.         log.Fatalf("failed to start workflow: %v", err)
  68.     }
  70.     // ...
  72.   // Notification that workflow has completed or failed
  73.     for {
  74.         timeDelta := time.Since(startTime)
  75.         metadata, err := wfClient.FetchWorkflowMetadata(context.Background(), id)
  76.         if err != nil {
  77.             log.Fatalf("failed to fetch workflow: %v", err)
  78.         }
  79.         if (metadata.RuntimeStatus == workflow.StatusCompleted) || (metadata.RuntimeStatus == workflow.StatusFailed) || (metadata.RuntimeStatus == workflow.StatusTerminated) {
  80.             fmt.Printf("Workflow completed - result: %v\n", metadata.RuntimeStatus.String())
  81.             break
  82.         }
  83.         if timeDelta.Seconds() >= 10 {
  84.             metadata, err := wfClient.FetchWorkflowMetadata(context.Background(), id)
  85.             if err != nil {
  86.                 log.Fatalf("failed to fetch workflow: %v", err)
  87.             }
  88.             if totalCost > 50000 && !approvalSought && ((metadata.RuntimeStatus != workflow.StatusCompleted) || (metadata.RuntimeStatus != workflow.StatusFailed) || (metadata.RuntimeStatus != workflow.StatusTerminated)) {
  89.                 approvalSought = true
  90.                 promptForApproval(id)
  91.             }
  92.         }
  93.         // Sleep to not DoS the dapr dev instance
  94.         time.Sleep(time.Second)
  95.     }
  97.     fmt.Println("Purchase of item is complete")
  98. }
  100. // Request approval (RequestApprovalActivity)
  101. func promptForApproval(id string) {
  102.     wfClient, err := workflow.NewClient()
  103.     if err != nil {
  104.         log.Fatalf("failed to initialise wfClient: %v", err)
  105.     }
  106.     if err := wfClient.RaiseEvent(context.Background(), id, "manager_approval"); err != nil {
  107.         log.Fatal(err)
  108.     }
  109. }
  111. // Update inventory for remaining stock (UpdateInventoryActivity)
  112. func restockInventory(daprClient client.Client, inventory []InventoryItem) error {
  113.     for _, item := range inventory {
  114.         itemSerialized, err := json.Marshal(item)
  115.         if err != nil {
  116.             return err
  117.         }
  118.         fmt.Printf("adding base stock item: %s\n", item.ItemName)
  119.         if err := daprClient.SaveState(context.Background(), stateStoreName, item.ItemName, itemSerialized, nil); err != nil {
  120.             return err
  121.         }
  122.     }
  123.     return nil
  124. }

同时，OrderProcessingWorkflow及其活动在workflow.go中定义为方法。

告诉我们您的想法

我们一直在努力改进我们的快速入门示例，并重视您的反馈。 您觉得此快速入门有帮助吗？ 您有改进的建议吗？

加入我们的discord频道参与讨论。

下一步

探索 Dapr 教程 >>