指南：如何编写和管理工作流

注意

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

让我们创建一个 Dapr 工作流，并使用控制台调用它。 通过提供的工作流示例，您将：

这个示例使用dapr init中的默认配置，在自托管模式下。

前期准备

设置环境

克隆JavaScript SDK存储库并进入该存储库。

  1. git clone https://github.com/dapr/js-sdk
  2. cd js-sdk

从 JavaScript SDK 根目录中，导航到 Dapr Workflow 示例。

  1. cd examples/workflow/authoring

运行以下命令使用 Dapr JavaScript SDK 安装运行此工作流示例所需的依赖项。

  1. npm install

运行activity-sequence.ts

activity-sequence 文件在 Dapr 工作流运行时中注册了一个工作流和一个活动。 工作流是按顺序执行的一系列活动。 我们使用DaprWorkflowClient来调度一个新的工作流实例并等待其完成。

  1. const daprHost = "localhost";
  2. const daprPort = "50001";
  3. const workflowClient = new DaprWorkflowClient({
  4.   daprHost,
  5.   daprPort,
  6. });
  7. const workflowRuntime = new WorkflowRuntime({
  8.   daprHost,
  9.   daprPort,
  10. });
  12. const hello = async (_: WorkflowActivityContext, name: string) => {
  13.   return `Hello ${name}!`;
  14. };
  16. const sequence: TWorkflow = async function* (ctx: WorkflowContext): any {
  17.   const cities: string[] = [];
  19.   const result1 = yield ctx.callActivity(hello, "Tokyo");
  20.   cities.push(result1);
  21.   const result2 = yield ctx.callActivity(hello, "Seattle");
  22.   cities.push(result2);
  23.   const result3 = yield ctx.callActivity(hello, "London");
  24.   cities.push(result3);
  26.   return cities;
  27. };
  29. workflowRuntime.registerWorkflow(sequence).registerActivity(hello);
  31. // Wrap the worker startup in a try-catch block to handle any errors during startup
  32. try {
  33.   await workflowRuntime.start();
  34.   console.log("Workflow runtime started successfully");
  35. } catch (error) {
  36.   console.error("Error starting workflow runtime:", error);
  37. }
  39. // Schedule a new orchestration
  40. try {
  41.   const id = await workflowClient.scheduleNewWorkflow(sequence);
  42.   console.log(`Orchestration scheduled with ID: ${id}`);
  44.   // Wait for orchestration completion
  45.   const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);
  47.   console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
  48. } catch (error) {
  49.   console.error("Error scheduling or waiting for orchestration:", error);
  50. }

在上面的代码中:

  • workflowRuntime.registerWorkflow(sequence) 在 Dapr Workflow 运行时中将 sequence 注册为一个工作流。
  • await workflowRuntime.start(); 构建并启动 Dapr Workflow 运行时内的引擎。
  • await workflowClient.scheduleNewWorkflow(sequence) 使用 Dapr Workflow 运行时调度一个新的工作流实例。
  • await workflowClient.waitForWorkflowCompletion(id, undefined, 30) 等待工作流实例完成。

在终端中执行以下命令来启动 activity-sequence.ts：

  1. npm run start:dapr:activity-sequence

预期输出

  1. You're up and running! Both Dapr and your app logs will appear here.
  3. ...
  5. == APP == Orchestration scheduled with ID: dc040bea-6436-4051-9166-c9294f9d2201
  6. == APP == Waiting 30 seconds for instance dc040bea-6436-4051-9166-c9294f9d2201 to complete...
  7. == APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
  8. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 0 history event...
  9. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, EXECUTIONSTARTED=1]
  10. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Waiting for 1 task(s) and 0 event(s) to complete...
  11. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
  12. == APP == Received "Activity Request" work item
  13. == APP == Activity hello completed with output "Hello Tokyo!" (14 chars)
  14. == APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
  15. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 3 history event...
  16. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  17. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Waiting for 1 task(s) and 0 event(s) to complete...
  18. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
  19. == APP == Received "Activity Request" work item
  20. == APP == Activity hello completed with output "Hello Seattle!" (16 chars)
  21. == APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
  22. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 6 history event...
  23. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  24. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Waiting for 1 task(s) and 0 event(s) to complete...
  25. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
  26. == APP == Received "Activity Request" work item
  27. == APP == Activity hello completed with output "Hello London!" (15 chars)
  28. == APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
  29. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 9 history event...
  30. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
  31. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Orchestration completed with status COMPLETED
  32. == APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
  33. INFO[0006] dc040bea-6436-4051-9166-c9294f9d2201: 'sequence' completed with a COMPLETED status.  app_id=activity-sequence-workflow instance=kaibocai-devbox scope=wfengine.backend type=log ver=1.12.3
  34. == APP == Instance dc040bea-6436-4051-9166-c9294f9d2201 completed
  35. == APP == Orchestration completed! Result: ["Hello Tokyo!","Hello Seattle!","Hello London!"]

下一步