指南：如何管理工作流

管理和运行工作流

注意

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

现在，您已经在应用程序中编写了工作流及其活动，可以使用HTTP API调用来启动、终止和获取有关工作流的信息。 更多信息，请阅读 工作流 API 参考。

在代码中管理工作流。 在编写工作流指南中的工作流示例中，工作流是使用以下API在代码中注册的：

  • start_workflow: 启动一个工作流实例
  • get_workflow: 获取工作流的状态信息
  • pause_workflow: 暂停或中止一个工作流实例，稍后可恢复该实例
  • resume_workflow: 恢复暂停的工作流实例
  • raise_workflow_event: 在工作流中引发事件
  • purge_workflow: 删除与特定工作流实例相关的所有元数据
  • terminate_workflow: 终止或停止工作流的特定实例
  1. from dapr.ext.workflow import WorkflowRuntime, DaprWorkflowContext, WorkflowActivityContext
  2. from dapr.clients import DaprClient
  4. # Sane parameters
  5. instanceId = "exampleInstanceID"
  6. workflowComponent = "dapr"
  7. workflowName = "hello_world_wf"
  8. eventName = "event1"
  9. eventData = "eventData"
  11. # Start the workflow
  12. start_resp = d.start_workflow(instance_id=instanceId, workflow_component=workflowComponent,
  13.                         workflow_name=workflowName, input=inputData, workflow_options=workflowOptions)
  15. # Get info on the workflow
  16. getResponse = d.get_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  18. # Pause the workflow
  19. d.pause_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  21. # Resume the workflow
  22. d.resume_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  24. # Raise an event on the workflow. 
  25.  d.raise_workflow_event(instance_id=instanceId, workflow_component=workflowComponent,
  26.                     event_name=eventName, event_data=eventData)
  28. # Purge the workflow
  29. d.purge_workflow(instance_id=instanceId, workflow_component=workflowComponent)
  31. # Terminate the workflow
  32. d.terminate_workflow(instance_id=instanceId, workflow_component=workflowComponent)

在代码中管理工作流。 在编写工作流指南中的工作流示例中，工作流是使用以下API在代码中注册的：

  • client.workflow.start: 启动一个工作流实例
  • client.workflow.get: 获取工作流的状态信息
  • client.workflow.pause: 暂停或中止一个工作流实例，稍后可恢复该实例
  • client.workflow.resume: 恢复暂停的工作流实例
  • client.workflow.purge: 删除与特定工作流实例相关的所有元数据
  • client.workflow.terminate: 终止或停止工作流的特定实例
  1. import { DaprClient } from "@dapr/dapr";
  3. async function printWorkflowStatus(client: DaprClient, instanceId: string) {
  4.   const workflow = await client.workflow.get(instanceId);
  5.   console.log(
  6.     `Workflow ${workflow.workflowName}, created at ${workflow.createdAt.toUTCString()}, has status ${
  7.       workflow.runtimeStatus
  8.     }`,
  9.   );
  10.   console.log(`Additional properties: ${JSON.stringify(workflow.properties)}`);
  11.   console.log("--------------------------------------------------\n\n");
  12. }
  14. async function start() {
  15.   const client = new DaprClient();
  17.   // Start a new workflow instance
  18.   const instanceId = await client.workflow.start("OrderProcessingWorkflow", {
  19.     Name: "Paperclips",
  20.     TotalCost: 99.95,
  21.     Quantity: 4,
  22.   });
  23.   console.log(`Started workflow instance ${instanceId}`);
  24.   await printWorkflowStatus(client, instanceId);
  26.   // Pause a workflow instance
  27.   await client.workflow.pause(instanceId);
  28.   console.log(`Paused workflow instance ${instanceId}`);
  29.   await printWorkflowStatus(client, instanceId);
  31.   // Resume a workflow instance
  32.   await client.workflow.resume(instanceId);
  33.   console.log(`Resumed workflow instance ${instanceId}`);
  34.   await printWorkflowStatus(client, instanceId);
  36.   // Terminate a workflow instance
  37.   await client.workflow.terminate(instanceId);
  38.   console.log(`Terminated workflow instance ${instanceId}`);
  39.   await printWorkflowStatus(client, instanceId);
  41.   // Wait for the workflow to complete, 30 seconds!
  42.   await new Promise((resolve) => setTimeout(resolve, 30000));
  43.   await printWorkflowStatus(client, instanceId);
  45.   // Purge a workflow instance
  46.   await client.workflow.purge(instanceId);
  47.   console.log(`Purged workflow instance ${instanceId}`);
  48.   // This will throw an error because the workflow instance no longer exists.
  49.   await printWorkflowStatus(client, instanceId);
  50. }
  52. start().catch((e) => {
  53.   console.error(e);
  54.   process.exit(1);
  55. });

在代码中管理工作流。 在OrderProcessingWorkflow示例中，来自编写工作流指南，工作流在代码中注册。 现在，您可以启动、终止正在运行的工作流程，并获取相关信息：

  1. string orderId = "exampleOrderId";
  2. string workflowComponent = "dapr";
  3. string workflowName = "OrderProcessingWorkflow";
  4. OrderPayload input = new OrderPayload("Paperclips", 99.95);
  5. Dictionary<string, string> workflowOptions; // This is an optional parameter
  7. // Start the workflow. This returns back a "StartWorkflowResponse" which contains the instance ID for the particular workflow instance.
  8. StartWorkflowResponse startResponse = await daprClient.StartWorkflowAsync(orderId, workflowComponent, workflowName, input, workflowOptions);
  10. // Get information on the workflow. This response contains information such as the status of the workflow, when it started, and more!
  11. GetWorkflowResponse getResponse = await daprClient.GetWorkflowAsync(orderId, workflowComponent, eventName);
  13. // Terminate the workflow
  14. await daprClient.TerminateWorkflowAsync(orderId, workflowComponent);
  16. // Raise an event (an incoming purchase order) that your workflow will wait for. This returns the item waiting to be purchased.
  17. await daprClient.RaiseWorkflowEventAsync(orderId, workflowComponent, workflowName, input);
  19. // Pause
  20. await daprClient.PauseWorkflowAsync(orderId, workflowComponent);
  22. // Resume
  23. await daprClient.ResumeWorkflowAsync(orderId, workflowComponent);
  25. // Purge the workflow, removing all inbox and history information from associated instance
  26. await daprClient.PurgeWorkflowAsync(orderId, workflowComponent);

在代码中管理工作流。 在Java SDK的工作流示例中，工作流是使用以下API在代码中注册的：

  • scheduleNewWorkflow: 启动新的工作流实例
  • getInstanceState: 获取工作流的状态信息
  • waitForInstanceStart: 暂停或中止一个工作流实例，稍后可恢复该实例
  • raiseEvent: 为正在运行的工作流实例触发事件/任务
  • waitForInstanceCompletion: 等待工作流完成其任务
  • purgeInstance: 删除与特定工作流实例相关的所有元数据
  • terminateWorkflow: 终止工作流程
  • purgeInstance: 删除与特定工作流相关的所有元数据
  1. package io.dapr.examples.workflows;
  3. import io.dapr.workflows.client.DaprWorkflowClient;
  4. import io.dapr.workflows.client.WorkflowInstanceStatus;
  6. // ...
  7. public class DemoWorkflowClient {
  9.   // ...
  10.   public static void main(String[] args) throws InterruptedException {
  11.     DaprWorkflowClient client = new DaprWorkflowClient();
  13.     try (client) {
  14.       // Start a workflow
  15.       String instanceId = client.scheduleNewWorkflow(DemoWorkflow.class, "input data");
  17.       // Get status information on the workflow
  18.       WorkflowInstanceStatus workflowMetadata = client.getInstanceState(instanceId, true);
  20.       // Wait or pause for the workflow instance start
  21.       try {
  22.         WorkflowInstanceStatus waitForInstanceStartResult =
  23.             client.waitForInstanceStart(instanceId, Duration.ofSeconds(60), true);
  24.       }
  26.       // Raise an event for the workflow; you can raise several events in parallel
  27.       client.raiseEvent(instanceId, "TestEvent", "TestEventPayload");
  28.       client.raiseEvent(instanceId, "event1", "TestEvent 1 Payload");
  29.       client.raiseEvent(instanceId, "event2", "TestEvent 2 Payload");
  30.       client.raiseEvent(instanceId, "event3", "TestEvent 3 Payload");
  32.       // Wait for workflow to complete running through tasks
  33.       try {
  34.         WorkflowInstanceStatus waitForInstanceCompletionResult =
  35.             client.waitForInstanceCompletion(instanceId, Duration.ofSeconds(60), true);
  36.       } 
  38.       // Purge the workflow instance, removing all metadata associated with it
  39.       boolean purgeResult = client.purgeInstance(instanceId);
  41.       // Terminate the workflow instance
  42.       client.terminateWorkflow(instanceToTerminateId, null);
  44.     System.exit(0);
  45.   }
  46. }

在代码中管理工作流。 在Go SDK的工作流示例中，工作流是使用以下API在代码中注册的：

  • StartWorkflow: 启动新的工作流实例
  • GetWorkflow: 获取工作流的状态信息
  • PauseWorkflow: 暂停或中止一个工作流实例，稍后可恢复该实例
  • RaiseEventWorkflow: 为正在运行的工作流实例触发事件/任务
  • ResumeWorkflow: 等待工作流完成其任务
  • PurgeWorkflow：删除与特定工作流实例相关的所有元数据
  • TerminateWorkflow: 终止工作流程
  1. // Start workflow
  2. type StartWorkflowRequest struct {
  3.     InstanceID        string // Optional instance identifier
  4.     WorkflowComponent string
  5.     WorkflowName      string
  6.     Options           map[string]string // Optional metadata
  7.     Input             any               // Optional input
  8.     SendRawInput      bool              // Set to True in order to disable serialization on the input
  9. }
  11. type StartWorkflowResponse struct {
  12.     InstanceID string
  13. }
  15. // Get the workflow status
  16. type GetWorkflowRequest struct {
  17.     InstanceID        string
  18.     WorkflowComponent string
  19. }
  21. type GetWorkflowResponse struct {
  22.     InstanceID    string
  23.     WorkflowName  string
  24.     CreatedAt     time.Time
  25.     LastUpdatedAt time.Time
  26.     RuntimeStatus string
  27.     Properties    map[string]string
  28. }
  30. // Purge workflow
  31. type PurgeWorkflowRequest struct {
  32.     InstanceID        string
  33.     WorkflowComponent string
  34. }
  36. // Terminate workflow
  37. type TerminateWorkflowRequest struct {
  38.     InstanceID        string
  39.     WorkflowComponent string
  40. }
  42. // Pause workflow
  43. type PauseWorkflowRequest struct {
  44.     InstanceID        string
  45.     WorkflowComponent string
  46. }
  48. // Resume workflow
  49. type ResumeWorkflowRequest struct {
  50.     InstanceID        string
  51.     WorkflowComponent string
  52. }
  54. // Raise an event for the running workflow
  55. type RaiseEventWorkflowRequest struct {
  56.     InstanceID        string
  57.     WorkflowComponent string
  58.     EventName         string
  59.     EventData         any
  60.     SendRawData       bool // Set to True in order to disable serialization on the data
  61. }

使用 HTTP 调用管理工作流。 下面的示例插入了 编写工作流示例 中的属性，并随机设置了实例 ID 号。

启动工作流

要使用 ID 12345678 启动工作流，请运行：

  1. POST http://localhost:3500/v1.0-beta1/workflows/dapr/OrderProcessingWorkflow/start?instanceID=12345678

请注意，工作流实例 ID 只能包含字母数字字符、下划线和破折号。

终止工作流程

要使用 ID 12345678 终止工作流，请运行：

  1. POST http://localhost:3500/v1.0-beta1/workflows/dapr/12345678/terminate

引发事件

对于支持订阅外部事件的工作流组件（如 Dapr 工作流引擎），您可以使用以下 “raise event” API 向特定工作流实例发送已命名的事件。

  1. POST http://localhost:3500/v1.0-beta1/workflows/<workflowComponentName>/<instanceID>/raiseEvent/<eventName>

eventName 可以是任何函数。

暂停或恢复工作流

若要规划停机时间、等待输入等，可以暂停，然后恢复工作流。 要暂停一个带有ID 12345678的工作流，直到触发恢复，请运行：

  1. POST http://localhost:3500/v1.0-beta1/workflows/dapr/12345678/pause

要恢复一个带有 ID 12345678 的工作流，请运行：

  1. POST http://localhost:3500/v1.0-beta1/workflows/dapr/12345678/resume

清除工作流

清除 API 可用于从底层状态存储中永久删除工作流元数据，包括任何已存储的输入、输出和工作流历史记录。 这通常有助于实施数据保留政策和释放资源。

只有处于 “已完成”、“已失败 “或 “已终止 “状态的工作流实例才能被清除。 如果工作流处于任何其他状态，调用清除会返回错误。

  1. POST http://localhost:3500/v1.0-beta1/workflows/dapr/12345678/purge

获取工作流程信息

要获取工作流信息（输出和输入）的 ID 为 12345678，请运行:

  1. GET http://localhost:3500/v1.0-beta1/workflows/dapr/12345678

了解有关这些HTTP调用的更多信息，请参阅工作流API参考指南。

下一步