This is the multi-page printable view of this section. Click here to print.

Return to the regular view of this page.

工作流

跨多个微服务编排逻辑

1 - 工作流概述

Dapr Workflow 概述

Dapr workflow 使开发者能够以可靠的方式编写业务逻辑和集成。 由于 Dapr workflows 是有状态的,因此支持长时间运行和容错的应用程序,非常适合编排微服务。 Dapr workflow 可与其他 Dapr 构建块无缝协作,例如服务调用、发布订阅、状态管理和绑定。

持久化、有弹性的 Dapr Workflow 能力:

  • 提供内置的工作流运行时来驱动 Dapr Workflow 执行。
  • 提供用于以任何语言编写工作流的 SDK。
  • 提供 HTTP 和 gRPC API 用于管理工作流(启动、查询、暂停/恢复、触发事件、终止、清除)。
显示 Dapr Workflow 基础的图表

Dapr Workflow 可以执行的一些示例场景包括:

  • 涉及库存管理、支付系统和物流服务之间编排的订单处理。
  • 跨多个部门和参与者协调任务的 HR 入职工作流。
  • 在全国连锁餐厅中编排数字菜单更新的推出。
  • 涉及基于 API 的分类和存储的图像处理工作流。

功能

工作流和活动

使用 Dapr Workflow,你可以编写活动,然后在工作流中编排这些活动。 工作流活动具有以下特点:

  • 工作流中的基本工作单元
  • 用于调用其他(Dapr)服务、与状态存储和发布订阅代理交互。
  • 用于调用外部第三方服务。

详细了解工作流活动。

子工作流

除了活动之外,你还可以编写工作流来调度其他工作流作为子工作流。 子工作流有其自己的实例 ID、历史记录和状态,独立于启动它的父工作流,但终止父工作流会终止其创建的所有子工作流除外。 子工作流还支持自动重试策略。

详细了解子工作流。

多应用程序工作流

多应用程序工作流使你能够编排跨多个应用程序的复杂业务流程。 这允许工作流在不同应用程序中调用活动或启动子工作流,在保持 Dapr 工作流引擎的安全、可靠性和持久性保证的同时分配工作流执行。

详细了解多应用程序工作流。

定时器和提醒

与 Dapr actors 一样,你可以为任意时间范围安排类似提醒的持久延迟。

详细了解工作流定时器提醒

用于管理工作流的 Workflow HTTP 调用

当你使用工作流代码创建应用程序并使用 Dapr 运行它时,你可以调用驻留在该应用程序中的特定工作流。 每个独立的工作流可以:

  • 通过 POST 请求启动或终止
  • 通过 POST 请求触发传递命名事件
  • 通过 POST 请求暂停然后恢复
  • 通过 POST 请求从状态存储中清除
  • 通过 GET 请求查询工作流状态

详细了解如何使用 HTTP 调用管理工作流。

工作流模式

Dapr Workflow 简化了微服务架构中复杂的有状态协调需求。 以下部分描述了可以从 Dapr Workflow 中受益的几种应用程序模式。

详细了解不同类型的工作流模式

工作流 SDK

Dapr Workflow authoring SDK 是特定语言的 SDK,包含用于实现工作流逻辑的类型和函数。 工作流逻辑存在于你的应用程序中,并通过 gRPC 流由 Dapr 边车中运行的 Dapr Workflow 引擎编排。

支持的 SDK

你可以使用以下 SDK 编写工作流。

语言堆栈
Pythondapr-ext-workflow
JavaScriptDaprWorkflowClient
.NETDapr.Workflow
Javaio.dapr.workflows
Goworkflow

试用工作流

快速入门和教程

想测试工作流吗?请完成以下快速入门和教程来查看工作流的实际应用:

快速入门/教程描述
Workflow 快速入门运行一个包含四个工作流活动的工作流应用程序,了解 Dapr Workflow 的实际应用
Workflow Python SDK 示例了解如何使用 Python dapr-ext-workflow 包创建 Dapr Workflow 并调用它。
Workflow JavaScript SDK 示例了解如何使用 JavaScript SDK 创建 Dapr Workflow 并调用它。
Workflow .NET SDK 示例了解如何使用 ASP.NET Core Web API 创建 Dapr Workflow 并调用它。
Workflow Java SDK 示例了解如何使用 Java io.dapr.workflows 包创建 Dapr Workflow 并调用它。
Workflow Go SDK 示例了解如何使用 Go workflow 包创建 Dapr Workflow 并调用它。

直接在你的应用中开始使用工作流

想跳过快速入门吗?没问题。你可以直接在你的应用程序中试用工作流构建块。 安装 Dapr后,你可以开始使用工作流,从如何编写工作流开始。

管理工作流

Dapr 通过 HTTP API 和 CLI 提供全面的工作流管理能力。

工作流生命周期操作

启动工作流

dapr workflow run MyWorkflow --app-id myapp --input '{"key": "value"}'

监控工作流

# 列出给定应用程序的活跃工作流
dapr workflow list --app-id myapp --filter-status RUNNING

# 查看执行历史
dapr workflow history <instance-id> --app-id myapp

控制工作流

# 暂停、恢复或终止
dapr workflow suspend <instance-id> --app-id myapp
dapr workflow resume <instance-id> --app-id myapp
dapr workflow terminate <instance-id> --app-id myapp

维护操作

# 清除已完成的工作流
dapr workflow purge --app-id myapp --all-older-than 720h

详细说明请参阅如何:管理工作流

限制

观看演示

观看此视频了解 Dapr Workflow 概述

下一步

工作流功能和概念 >>

相关链接

2 - 功能与概念

了解更多关于 Dapr 工作流的功能与概念

现在您已经对工作流构建块有了整体了解,让我们深入探讨 Dapr 工作流引擎和 SDK 提供的功能和概念。 Dapr 工作流公开了几个核心功能和概念,这些在所有支持的语言中都是通用的。

工作流

Dapr 工作流是您编写的函数,用于定义要按特定顺序执行的一系列任务。 Dapr 工作流引擎负责任务的调度和执行,包括管理失败和重试。 如果托管工作流的应用程序扩展到多台机器上,工作流引擎会在多台机器之间对工作流及其任务的执行进行负载均衡。

工作流可以调度多种不同类型的任务,包括:

  • 活动:用于执行自定义逻辑
  • 持久化定时器:用于将工作流暂停任意时长
  • 子工作流:用于将较大的工作流拆分为较小的部分
  • 外部事件等待器:用于阻塞工作流直到收到外部事件信号。这些任务将在其对应章节中详细描述。

工作流实例管理

查询工作流状态

您可以使用 CLI 查询工作流实例:

# 查找所有运行中的工作流
dapr workflow list --app-id myapp --filter-status RUNNING

# 按名称查找工作流
dapr workflow list --app-id myapp --filter-name OrderProcessing

# 查找最近的工作流(过去 2 小时)
dapr workflow list --app-id myapp --filter-max-age 2h

# 获取详细的 JSON 输出
dapr workflow list --app-id myapp --output json

工作流历史

查看完整的执行历史:

dapr workflow history wf-12345 --app-id myapp --output json

这将显示所有事件、活动和状态转换。

外部事件

通过 CLI 触发事件

dapr workflow raise-event wf-12345/ApprovalReceived \
  --app-id myapp \
  --input '{"approved": true, "comments": "Approved by manager"}'

工作流暂停与恢复

使用 CLI

# 暂停以进行手动干预
dapr workflow suspend wf-12345 \
  --app-id myapp \
  --reason "Awaiting customer response"

# 准备就绪后恢复
dapr workflow resume wf-12345 \
  --app-id myapp \
  --reason "Customer responded"

工作流标识

每个您定义的工作流都有一个类型名称,而工作流的每次单独执行都需要一个唯一的_实例 ID_。工作流实例 ID 可以由您的应用程序代码生成,这在工作流对应于文档或作业等业务实体时非常有用;也可以是自动生成的 UUID。工作流的实例 ID 可用于调试,也可用于使用 Workflow API 管理工作流。

任何给定时刻只能存在一个具有给定 ID 的工作流实例。但是,如果某个工作流实例完成或失败,其 ID 可以被新的工作流实例重用。但请注意,新的工作流实例将有效地替换配置的状态存储中的旧实例。

工作流重放

Dapr 工作流使用一种称为事件溯源的技术来维护其执行状态。工作流引擎不存储工作流的当前状态快照,而是管理一个仅追加的历史事件日志,记录描述工作流所执行的各种步骤。当使用工作流 SDK 时,这些历史事件会在工作流"等待"调度任务的结果时自动存储。

当工作流"等待"调度任务时,它会从内存中卸载,直到任务完成。一旦任务完成,工作流引擎会调度工作流函数再次运行。第二次工作流函数执行称为_重放_。

当工作流函数被重放时,它会从头开始运行。但是,当它遇到已经完成的任务时,工作流引擎不会再次调度该任务,而是:

  1. 将已完成任务的存储结果返回给工作流。
  2. 继续执行直到下一个"等待"点。

这种"重放"行为会持续进行,直到工作流函数完成或以错误失败。

使用这种重放技术,工作流能够从任何"等待"点恢复执行,就好像它从未从内存中卸载一样。甚至可以恢复前次运行中的局部变量值,而工作流引擎无需知道它们存储了什么数据。这种恢复状态的能力使 Dapr 工作流具有_持久性_和_容错性_。

无限循环与永生工作流

工作流重放章节所述,工作流维护其所有操作的只写事件溯源历史日志。为了避免资源失控使用,工作流必须限制其调度的操作数量。例如,确保您的工作流不会:

  • 在其实现中使用无限循环
  • 调度数千个任务。

如果工作流可能需要调度大量任务,您可以使用以下两种技术来编写工作流:

  1. 使用 continue-as-new API: 每个工作流 SDK 都暴露了一个 continue-as-new API,工作流可以调用它来使用新的输入和历史重新启动自身。continue-as-new API 特别适合实现"永生工作流",如监控代理,否则将使用类似 while (true) 的构造来实现。使用 continue-as-new 是保持工作流历史记录规模较小的好方法。

    continue-as-new API 会截断现有历史记录,用新的历史记录替换它。

  2. 使用子工作流: 每个工作流 SDK 都暴露了用于创建子工作流的 API。子工作流的行为与任何其他工作流一样,只是它由父工作流调度。子工作流具有:

    • 自己的历史记录
    • 跨多台机器分发工作流函数执行的好处。

    如果工作流需要调度数千个或更多任务,建议将这些任务分布在子工作流中,以避免单个工作流的历史记录规模过大。

更新工作流代码

由于工作流是长期运行且持久的,更新工作流代码必须非常小心。如工作流确定性限制章节所述,工作流代码必须具有确定性。如果系统中存在任何未完成的工作流实例,则对工作流代码的更新必须保持这种确定性。否则,对工作流代码的更新可能导致这些工作流下次执行时出现运行时故障。

查看已知限制

工作流活动

工作流活动是工作流中的基本工作单元,也是业务流程中被编排的任务。例如,您可能创建一个工作流来处理订单。任务可能涉及检查库存、向客户收费和创建发货。每个任务都是一个单独的活动。这些活动可以串行执行、并行执行或两者结合执行。

与工作流不同,活动对您可以在其中执行的工作类型没有限制。活动经常用于发出网络调用或运行 CPU 密集型操作。活动也可以将数据返回给工作流。

Dapr 工作流引擎保证每个被调用的活动作为工作流执行的一部分至少执行一次。由于活动只保证至少一次执行,建议尽可能将活动逻辑实现为幂等的。

子工作流

除了活动,工作流还可以将其他工作流调度为_子工作流_。子工作流有自己的实例 ID、历史记录和状态,独立于启动它的父工作流。

子工作流有许多好处:

  • 您可以将大型工作流拆分为一系列较小的子工作流,使代码更易于维护。
  • 您可以跨多个计算节点并发分发工作流逻辑,如果您的工作流逻辑需要协调大量任务,这将非常有用。
  • 您可以通过保持父工作流的历史记录较小来减少内存使用和 CPU 开销。

子工作流的返回值就是其输出。如果子工作流因异常失败,则该异常会像活动任务因异常失败一样被暴露到父工作流。子工作流也支持自动重试策略。

终止父工作流会终止由该工作流实例创建的所有子工作流。详见终止工作流 API

持久化定时器

Dapr 工作流允许您为任意时间范围安排类似提醒的持久化延迟,包括分钟、天甚至数年。这些_持久化定时器_可以由工作流调度以实现简单延迟或为其他异步任务设置临时超时。更具体地说,持久化定时器可以设置为在特定日期触发或在指定持续时间后触发。持久化定时器的最大持续时间没有限制,它们在内部由内部 actor 提醒器支持。例如,跟踪某项服务 30 天免费订阅的工作流可以使用在创建工作流 30 天后触发的持久化定时器来实现。工作流在等待持久化定时器触发时可以安全地从内存中卸载。

重试策略

工作流支持针对活动和子工作流的持久化重试策略。工作流重试策略与 Dapr 弹性策略 在以下方面有所不同:

  • 工作流重试策略由工作流作者在代码中配置,而 Dapr 弹性策略由应用程序操作员在 YAML 中配置。
  • 工作流重试策略是持久的,在应用程序重启后保持其状态,而 Dapr 弹性策略不是持久的,必须在应用程序重启后重新应用。
  • 工作流重试策略由活动和子工作流中未处理的错误/异常触发,而 Dapr 弹性策略由操作超时和连接故障触发。

重试在内部使用持久化定时器实现。这意味着工作流在等待重试触发时可以安全地从内存中卸载,从而节省系统资源。这也意味着重试之间的延迟可以任意长,包括分钟、小时甚至数天。

可以同时使用工作流重试策略和 Dapr 弹性策略。例如,如果工作流活动使用 Dapr 客户端调用服务,则 Dapr 客户端使用配置好的弹性策略。详见快速入门:服务间弹性以获取更多信息和示例。但是,如果活动本身因任何原因失败,包括耗尽弹性策略的重试次数,则工作流的弹性策略会介入。

由于工作流重试策略是在代码中配置的,确切的开发者体验可能因工作流 SDK 版本而异。一般来说,工作流重试策略可以使用以下参数进行配置:

参数描述
最大尝试次数执行活动或子工作流的最大次数。如果设置为 0,则不会进行任何尝试。
首次重试间隔等待第一次重试的时间量。
退避系数用于确定退避增加速率的系数。例如,系数为 2 会使每次后续重试的等待时间翻倍。
最大重试间隔每次后续重试之前等待的最大时间量。如果设置为 0,则不会发生重试。
重试超时重试的全局超时时间,无论配置的最大尝试次数如何。此超时到期后,将不再尝试执行活动。

外部事件

有时工作流需要等待由外部系统触发的事件。例如,审批工作流可能要求人工在工作流处理订单时明确批准订单请求(如果总成本超过某个阈值)。另一个例子是琐事游戏编排工作流,在等待所有参与者提交他们对琐事问题的答案时暂停。这些中途执行的输入称为_外部事件_。

外部事件具有_名称_和_有效载荷_,并被传递到单个工作流实例。工作流可以创建"等待外部事件"任务来订阅外部事件,并_等待_这些任务以阻塞执行直到收到事件。然后工作流可以读取这些事件的有效载荷,并决定下一步采取什么行动。外部事件可以串行或并行处理。外部事件可以由其他工作流或工作流代码触发。

工作流也可以等待多个相同名称的外部事件信号,在这种情况下,它们会以先进先出(FIFO)的方式被分派到相应的工作流任务。如果工作流收到外部事件信号但尚未创建"等待外部事件"任务,该事件将被保存到工作流的历史记录中,并在工作流请求该事件后立即被消费。

了解更多关于与外部系统交互的信息。

清除

工作流状态可以从状态存储中清除,清除其所有历史记录并移除与特定工作流实例相关的所有元数据。清除功能用于已运行到 COMPLETEDFAILEDTERMINATED 状态的工作流。

workflow API 参考指南 中了解更多。

版本控制

工作流代码是长期运行的,必须在更新期间保持确定性。有关修补和命名工作流版本控制的详细信息,请参阅工作流版本控制

限制

工作流确定性与代码约束

为了利用工作流重放技术,您的工作流代码需要具有确定性。为了使您的工作流代码具有确定性,您可能需要解决一些限制。

工作流函数必须调用确定性 API

生成随机数、随机 UUID 或获取当前日期的 API 是_非确定性的_。要解决此限制,您可以:

  • 在活动函数中使用这些 API,或
  • (推荐)使用 SDK 提供的内置等效 API。例如,每个创作 SDK 都提供了一种以确定性方式获取当前时间的 API。

例如,不要这样做:

// 不要这样做!
DateTime currentTime = DateTime.UtcNow;
Guid newIdentifier = Guid.NewGuid();
string randomString = GetRandomString();
// 不要这样做!
Instant currentTime = Instant.now();
UUID newIdentifier = UUID.randomUUID();
String randomString = getRandomString();
// 不要这样做!
const currentTime = new Date();
const newIdentifier = uuidv4();
const randomString = getRandomString();
// 不要这样做!
const currentTime = time.Now()

这样做:

// 这样做!!
DateTime currentTime = context.CurrentUtcDateTime;
Guid newIdentifier = context.NewGuid();
string randomString = await context.CallActivityAsync<string>(nameof("GetRandomString")); //使用 "nameof" 以防止指定应用程序中不存在的活动名称
// 这样做!!
Instant currentTime = context.getCurrentInstant();
Guid newIdentifier = context.newGuid();
String randomString = context.callActivity(GetRandomString.class.getName(), String.class).await();
// 这样做!!
const currentTime = context.getCurrentUtcDateTime();
const randomString = yield context.callActivity(getRandomString);
const currentTime = ctx.CurrentUTCDateTime()

工作流函数必须仅_间接_与外部状态交互。

外部数据包括任何不存储在工作流状态中的数据。工作流不得与全局变量、环境变量、文件系统交互,或进行网络调用。

相反,工作流应该使用工作流输入、活动任务和外部事件处理来_间接_地与外部状态交互。

例如,不要这样做:

// 不要这样做!
string configuration = Environment.GetEnvironmentVariable("MY_CONFIGURATION")!;
string data = await new HttpClient().GetStringAsync("https://example.com/api/data");
// 不要这样做!
String configuration = System.getenv("MY_CONFIGURATION");

HttpRequest request = HttpRequest.newBuilder().uri(new URI("https://postman-echo.com/post")).GET().build();
HttpResponse<String> response = HttpClient.newBuilder().build().send(request, HttpResponse.BodyHandlers.ofString());
// 不要这样做!
// 访问环境变量 (Node.js)
const configuration = process.env.MY_CONFIGURATION;

fetch('https://postman-echo.com/get')
  .then(response => response.text())
  .then(data => {
    console.log(data);
  })
  .catch(error => {
    console.error('Error:', error);
  });
// 不要这样做!
resp, err := http.Get("http://example.com/api/data")

这样做:

// 这样做!!
string configuration = workflowInput.Configuration; // 假设的工作流输入参数
string data = await context.CallActivityAsync<string>(nameof("MakeHttpCall"), "https://example.com/api/data");
// 这样做!!
String configuration = ctx.getInput(InputType.class).getConfiguration(); // 假设的工作流输入参数
String data = ctx.callActivity(MakeHttpCall.class, "https://example.com/api/data", String.class).await();
// 这样做!!
const configuration = workflowInput.getConfiguration(); // 假设的工作流输入参数
const data = yield ctx.callActivity(makeHttpCall, "https://example.com/api/data");
// 这样做!!
err := ctx.CallActivity(MakeHttpCallActivity, workflow.ActivityInput("https://example.com/api/data")).Await(&output)

工作流函数必须仅在工作流调度线程上执行。

每种语言 SDK 的实现都要求所有工作流函数操作在与调度函数的同一线程(goroutine 等)上操作。工作流函数必须永远不要:

  • 调度后台线程,或
  • 使用调度回调函数在另一个线程上运行的 API。

违反此规则可能导致未定义的行为。任何后台处理都应委托给活动任务,活动任务可以串行或并发调度。

例如,不要这样做:

// 不要这样做!
Task t = Task.Run(() => context.CallActivityAsync("DoSomething"));
await context.CreateTimer(5000).ConfigureAwait(false);
// 不要这样做!
new Thread(() -> {
    ctx.callActivity(DoSomethingActivity.class.getName()).await();
}).start();
ctx.createTimer(Duration.ofSeconds(5)).await();

不要将 JavaScript 工作流声明为 async。Node.js 运行时不能保证异步函数是确定性的。

// 不要这样做!
go func() {
  err := ctx.CallActivity(DoSomething).Await(nil)
}()
err := ctx.CreateTimer(time.Second).Await(nil)

这样做:

// 这样做!!
Task t = context.CallActivityAsync(nameof("DoSomething"));
await context.CreateTimer(5000).ConfigureAwait(true);
// 这样做!!
ctx.callActivity(DoSomethingActivity.class.getName()).await();
ctx.createTimer(Duration.ofSeconds(5)).await();

由于 Node.js 运行时不能保证异步函数是确定性的,请始终将 JavaScript 工作流声明为同步生成器函数。

// 这样做!
task := ctx.CallActivity(DoSomething)
task.Await(nil)

更新工作流代码

确保您对工作流代码的更新保持其确定性。以下是一些可能破坏工作流确定性的代码更新示例:

  • 更改工作流函数签名:更改工作流或活动的名称、输入或输出被视为破坏性更改,必须避免。
  • 更改工作流任务的数量或顺序:更改工作流任务的数量或顺序会导致工作流的历史记录与工作流代码不再匹配,并可能导致运行时错误或其他意外行为。

要解决这些约束,请使用版本控制指南中描述的工作流版本控制概念来修补和引入新的命名工作流版本,以确定性地将更改合并到您的工作流中。

下一步

工作流模式 >>

相关链接

3 - 工作流版本控制

在代码演进时安全地为工作流进行版本控制

版本控制

在许多场景中,需要在工作流活跃运行时引入对工作流代码的更改。在这些更改具有非确定性的情况下,需要采用版本控制策略,以便现有工作流可以继续使用原始代码执行,而新工作流则使用更新后的版本。

工作流可以使用两种互补的方法进行版本控制,这两种方法旨在结合使用。它们是:

版本控制解决了什么问题?

为了最好地理解版本控制如何工作以及如何有效地使用它,首先了解它解决了什么以及为什么需要版本控制会很有帮助。工作流必须是确定性的,这意味着每次运行时,它们都会产生与上次完全相同的输出。

这是一个关键概念,因为 Dapr 工作流实现使用事件溯源方法来持久化工作流状态。当在工作流中执行活动和子工作流时,Dapr 会维护一个历史记录,记录这些边界执行的输入和输出。当每个活动完成时,我们会持久化更新后的事件历史,然后从顶部重新调用工作流(这称为"重放")。这一次,当它遇到这些活动或子工作流之一时,它会替换已实现的输出并跳过重新执行,然后重复。

因此,至关重要的是,在工作流运行期间不得随意引入代码,因为当引擎重放工作流时,您的更改可能会导致工作流不再与其历史记录匹配并失败。

修补

修补是我们的两种版本控制方法中的第一种,它允许您在保持重放之间确定性的同时引入对工作流的更改。它的工作原理是允许您通过 if 语句插入与命名标识符关联的分支逻辑。工作流历史记录会在重放工作流时跟踪这些命名标识符的观察位置,从而确保在重放之间遵循一致的逻辑路径。

这使您能够仅对需要调整的特定位置对工作流代码进行有针对性的更改。

要添加补丁,请选择一个稳定的唯一名称来标识该补丁。这可以是描述性的内容,如 use-sms,它描述了补丁更改的内容,也可以是本质不同的内容,如当前时间戳或日期。您使用的具体值并不重要,只要不使用已部署到生产环境中的标识符即可。您将通过添加一个 if 语句来插入补丁,该语句评估工作流是否已应用此补丁——如果是,则使用补丁的代码;如果不是,则采用原始代码路径。

补丁标识符的唯一性要求仅扩展到此工作流;不会评估工作流之间的补丁,因此可以在多个工作流中使用重复的标识符而不会产生冲突。

以下示例演示了这一点,通过检查此工作流实例是否已应用 use-sms 补丁。如果是,则执行 SendSMS 活动;否则,它回退到原始路径并使用 SendEmail 活动。

public sealed class MyWorkflow : Workflow<string, object?>
{
    public overrride async Task<object?> RunAsync(WorkflowContext context, string input)
    {
        // ...
        if (context.IsPatched("use-sms"))
        {
            // 修补后的代码
            await context.CallActivityAsync(nameof(SendSMS), input);
        }
        else {
            // 原始代码
            await context.CallActivityAsync(nameof(SendEmail), input);
        }
        return null;
    }
}
import "github.com/dapr/durabletask-go/workflow"

func Workflow(ctx *workflow.WorkflowContext) error {
  // ...
  if ctx.IsPatched("use-sms") {
    if err := ctx.CallActivity("SendSMS", ctx.GetInput()).Await(); err != nil {
      return err
    }
  } else {
    if err := ctx.CallActivity("SendEmail", ctx.GetInput()).Await(); err != nil {
      return err
    }
  }
  // ...
}
from dapr.ext.workflow import DaprWorkflowContext, WorkflowRuntime

wfr = WorkflowRuntime()

@wfr.workflow
def my_workflow(ctx: DaprWorkflowContext, wf_input: str):
  # ...
  if ctx.is_patched("use-sms"):
    yield ctx.call_activity(send_sms)
  else:
    yield ctx.call_activity(send_email)
  # ...

使用这种方法,新的工作流实例将采用修补后的代码路径,但在引入补丁时正在运行的现有工作流将继续使用原始代码路径。补丁检查在首次评估时会记录在工作流实例历史记录中,这意味着可以在工作流的不同位置安全地使用相同的标识符。

同样,至关重要的是,一旦修补后的工作流在生产环境中运行,您就不应再重新使用该标识符。部署后,应假设引擎将以确定性方式处理工作流,但使用以前使用的标识符添加新补丁会破坏该契约:如果您的新代码插入位置早于正在运行的工作流实例评估到的位置,当它重放时,它将采用您的补丁分支,但您的更改可能不再与现有工作流历史记录匹配,并导致工作流无限期停滞(至少直到您删除新补丁或更新它以使用不同的唯一标识符)。

应用于工作流的补丁列表存储在工作流的历史记录中,因此重要的是要注意,您使用的补丁越多,工作流状态历史记录就越大。这将越来越对工作流性能产生负面影响,因为每次重放都需要检索它,并且随着状态的增长,这会增加检索和解析开销。

验证停滞

除了上述重新使用标识符外,使用补丁时还有其他几个原因导致工作流停滞:

  • 删除(或重命名)补丁标识符
  • 更改补丁的顺序

您可以使用 Dapr CLI 中的工作流命令 来检查停滞的工作流。例如,以下是当工作流停滞时 dapr workflow list 命令的显示结果:

> dapr workflow list
NAME           ID          STATUS     AGE
workflow       <ID>        STALLED    9m39s

以下是当工作流停滞时 dapr workflow history 命令的显示结果:

> dapr workflow history <ID> -k -o wide
PLAY  TYPE                 NAME      TIMESTAMP    ELAPSED    STATUS   DETAILS        ROUTER     EXECUTION ID  ATTRS
0     ExecutionStarted     workflow  <TIMESTAMP>  Age:15.8h  RUNNING  workflowStart  workflows  <EXEC_ID>     input=2026-01-22T14:44:02.728101
1     OrchestratorStarted            <TIMESTAMP>  25.3ms     RUNNING  replay                                  versionName=workflow_v1
1     ExecutionStalled               <TIMESTAMP>  8.9m       STALLED                                          reason=VERSION_NAME_MISMATCH;description=Version not available: workflow_v1

修补最佳实践

以下代表一些额外的建议,应考虑这些建议以消除因使用补丁而导致工作流停滞或失败的可能性:

  • 应用补丁时,原始代码必须保持可用且不变,以便正在运行的工作流能够以相同的方式评估它。
  • 补丁应以增量方式应用,这意味着一旦添加了补丁并部署了应用程序,就不应在工作流中移动或删除它们。它们必须存在以保持正在运行的工作流的确定性。
  • 如上所述,您不得重新使用以前部署中使用的补丁标识符,因为这将破坏确定性保证。但是,您可以在同一部署中的多个位置使用相同的标识符,而无需在不同工作流之间担心冲突。
  • 如果您可以在应用补丁时避免使用 else 分支,将使应用未来的补丁更容易。虽然替换现有代码时通常无法避免这种情况,但如果您只是向工作流添加新逻辑,这当然是可以管理的。
  • 如果必须嵌套补丁,则必须在现有补丁内进行。例如,您的新补丁不能在 else 分支中包装现有的补丁。否则,正在运行的工作流将无法访问原始代码,从而破坏确定性保证。

命名工作流版本控制

命名工作流版本控制代表我们的第二种方法,有助于以确定性安全的方式对工作流进行版本控制。在此方法中,您通过复制现有工作流并为其分配一个新的"版本化"名称来引入新的工作流版本。因为您可以从头开始重构工作流逻辑以删除任何补丁并引入您想要的任何其他更改,所以这种方法提供了与以前工作流版本的干净断开。

虽然实现此目的的具体细节取决于您使用的语言 SDK,但通常,您将复制最新的工作流,为其分配一个反映较新版本的唯一名称,并向您的 SDK 注册它。将为每个工作流构建一个注册表,以便在运行时,Dapr 将使用工作流名称进行调用,SDK 将路由到旧工作流(如果正在运行)或较新版本。

请注意,在所有 SDK 中,运行时不会在版本之间顺序迁移工作流。相反,当 SDK 收到运行新工作流的请求时,它选择运行最新版本,而不仅仅是"下一个"版本,因此无需处理版本之间的任何补偿逻辑。

语言 SDK 可能会公开一种在使用相同工作流名称时注册版本的方法,但这因 SDK 而异,因此请参阅特定 SDK 的文档以获取更多信息。例如,某些可能使用一种机制,您可以在其中显式提供 is_latest 标志来指示哪个版本是最新版本。

当 SDK 收到运行未注册版本的工作流的请求时,工作流将停滞。这可能在应用程序滚动部署期间自然发生,但也可能在某个版本被删除而某些工作流实例仍在使用它时发生。建议是,除非您已独立确认没有未完成的(正在运行或休眠的)工作流实例正在针对该版本运行,否则不应删除较旧的命名工作流版本。

.NET 使用源生成器自动识别和注册您的工作流版本,因此无需手动注册工作流(活动需要注册)。默认情况下,.NET 使用内置的可配置数字版本控制策略,其中版本作为名称的后缀提供。有关如何更改版本控制策略或配置以及如何在应用程序中设置版本控制,请参阅 .NET SDK 文档

由于 .NET 应用程序不允许存在多个具有相同名称的类型,因此这不是此 SDK 中的选项。

要创建新的命名工作流版本,请复制现有工作流并修改类的名称以使用相同的前缀,但更改后缀中的版本标识符以反映较新的版本。重新构建应用程序,SDK 将自动处理其余部分。

给定以下工作流定义:

import "github.com/dapr/durabletask-go/workflow"

func WorkflowV1(ctx *workflow.WorkflowContext) error {
  // ...
  return nil
}

func WorkflowV2(ctx *workflow.WorkflowContext) error {
  // ...
  return nil
}

这是您注册这两个工作流的方式:

import "github.com/dapr/durabletask-go/workflow"

registry := workflow.NewRegistry()
// 这是以前的工作流版本,因此 `isLatest` 为 false
registry.AddVersionedWorkflow("Workflow", false, WorkflowV1)
// 这是最新的工作流版本,因此 `isLatest` 为 true
registry.AddVersionedWorkflow("Workflow", true, WorkflowV2)

这是相同的示例,但显式设置版本名称:

import "github.com/dapr/durabletask-go/workflow"

registry := workflow.NewRegistry()
// 这是以前的工作流版本,因此 `isLatest` 为 false
registry.AddVersionedWorkflow("Workflow", "WorkflowV1", false, WorkflowV1)
// 这是最新的工作流版本,因此 `isLatest` 为 true
registry.AddVersionedWorkflow("Workflow", "WorkflowV2", true, WorkflowV2)

注意AddVersionedOrchestratorAddVersionedOrchestratorN 的布尔参数指示工作流是否是最新版本。

这是使用不同版本的工作流:

from dapr.ext.workflow import DaprWorkflowContext, WorkflowRuntime

wfr = WorkflowRuntime()

@wfr.versioned_workflow(name="workflow")
def workflow_v1(ctx: DaprWorkflowContext, wf_input: str):
  # ...

@wfr.versioned_workflow(name="workflow", is_latest=True)
def workflow_v2(ctx: DaprWorkflowContext, wf_input: str):
  # ...

这是相同的示例,但显式设置版本名称:

from dapr.ext.workflow import DaprWorkflowContext, WorkflowRuntime

wfr = WorkflowRuntime()

@wfr.versioned_workflow(name="workflow", version_name="workflow_v1")
def workflow_v1(ctx: DaprWorkflowContext, wf_input: str):
  # ...

@wfr.versioned_workflow(name="workflow", version_name="workflow_v2", is_latest=True)
def workflow_v2(ctx: DaprWorkflowContext, wf_input: str):
  # ...

版本控制流程指导

由于命名工作流与补丁完全兼容,因此该方法是对工作流的更改最初通过向现有工作流逻辑应用补丁来进行的。最终,在应用了几个补丁后,您将遇到以下问题之一:

  • 由于应用的补丁数量,您担心工作流状态历史记录的开销;
  • 很难遵循工作流逻辑的黄金路径;
  • 您需要进行另一个工作流调整,但无法弄清楚如何应用不破坏确定性保证的补丁。

此时,建议您引入工作流的命名版本。复制现有工作流,更改其名称以反映您在 SDK 中使用的任何版本控制策略,并重构以删除所有补丁,仅保留预期的"最新"逻辑。应用您的新更改(此处无需补丁——这是一个新的工作流)并部署它。

在这里,根据需要再次应用补丁以解决未来的更改,并在必要时引入另一个命名工作流版本并继续。无限重复此过程。

建议您保留旧的工作流版本,直到您完全确信没有任何正在运行的(包括使用长期运行计时器延迟的)使用任何旧工作流版本的内容。

工作流版本控制不解决更改工作流本身的输入和输出类型的问题。作为一般指导,建议要么返回序列化值(如字符串),使输出的使用者能够随着时间的推移反序列化不同的输出,要么采用包含可选属性的复杂类型以包含新的预期输出值。

4 - 工作流模式

编写不同类型的工作流模式

Dapr 工作流简化了微服务架构中复杂的有状态协调需求。以下各节介绍了几种可以从 Dapr 工作流中受益的应用程序模式。

任务链

在任务链模式中,工作流中的多个步骤按顺序运行,一个步骤的输出可以作为下一步的输入传递。任务链工作流通常涉及创建需要对某些数据执行的操作序列,例如过滤、转换和规约。

显示任务链工作流模式如何工作的示意图

在某些情况下,工作流的步骤可能需要跨多个微服务进行编排。为了提高可靠性和可扩展性,您可能还会使用队列来触发各个步骤。

虽然该模式很简单,但在实现中隐藏了许多复杂性。例如:

  • 如果某个微服务在较长时间内不可用,会发生什么?
  • 失败的步骤是否可以自动重试?
  • 如果不能,如何方便之前已完成的步骤的回滚(如果适用)?
  • 抛开实现细节,是否有方法可视化工作流,以便其他工程师可以理解它的作用和工作原理?

Dapr 工作流通过允许您使用您选择的编程语言将任务链模式简洁地实现为一个简单的函数来解决这些复杂性,如以下示例所示。

import dapr.ext.workflow as wf


def task_chain_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
    try:
        result1 = yield ctx.call_activity(step1, input=wf_input)
        result2 = yield ctx.call_activity(step2, input=result1)
        result3 = yield ctx.call_activity(step3, input=result2)
    except Exception as e:
        yield ctx.call_activity(error_handler, input=str(e))
        raise
    return [result1, result2, result3]


def step1(ctx, activity_input):
    print(f'Step 1: Received input: {activity_input}.')
    # Do some work
    return activity_input + 1


def step2(ctx, activity_input):
    print(f'Step 2: Received input: {activity_input}.')
    # Do some work
    return activity_input * 2


def step3(ctx, activity_input):
    print(f'Step 3: Received input: {activity_input}.')
    # Do some work
    return activity_input ^ 2


def error_handler(ctx, error):
    print(f'Executing error handler: {error}.')
    # Apply some compensating work

Note 工作流重试策略将在 Python SDK 的未来版本中提供。

import { DaprWorkflowClient, WorkflowActivityContext, WorkflowContext, WorkflowRuntime, TWorkflow } from "@dapr/dapr";

async function start() {
  // Update the gRPC client and worker to use a local address and port
  const daprHost = "localhost";
  const daprPort = "50001";
  const workflowClient = new DaprWorkflowClient({
    daprHost,
    daprPort,
  });
  const workflowRuntime = new WorkflowRuntime({
    daprHost,
    daprPort,
  });

  const hello = async (_: WorkflowActivityContext, name: string) => {
    return `Hello ${name}!`;
  };

  const sequence: TWorkflow = async function* (ctx: WorkflowContext): any {
    const cities: string[] = [];

    const result1 = yield ctx.callActivity(hello, "Tokyo");
    cities.push(result1);
    const result2 = yield ctx.callActivity(hello, "Seattle");
    cities.push(result2);
    const result3 = yield ctx.callActivity(hello, "London");
    cities.push(result3);

    return cities;
  };

  workflowRuntime.registerWorkflow(sequence).registerActivity(hello);

  // Wrap the worker startup in a try-catch block to handle any errors during startup
  try {
    await workflowRuntime.start();
    console.log("Workflow runtime started successfully");
  } catch (error) {
    console.error("Error starting workflow runtime:", error);
  }

  // Schedule a new orchestration
  try {
    const id = await workflowClient.scheduleNewWorkflow(sequence);
    console.log(`Orchestration scheduled with ID: ${id}`);

    // Wait for orchestration completion
    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);

    console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
  } catch (error) {
    console.error("Error scheduling or waiting for orchestration:", error);
  }

  await workflowRuntime.stop();
  await workflowClient.stop();

  // stop the dapr side car
  process.exit(0);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
 # Apply custom compensation logic
});
// Expotential backoff retry policy that survives long outages
var retryOptions = new WorkflowTaskOptions
{
    RetryPolicy = new WorkflowRetryPolicy(
        firstRetryInterval: TimeSpan.FromMinutes(1),
        backoffCoefficient: 2.0,
        maxRetryInterval: TimeSpan.FromHours(1),
        maxNumberOfAttempts: 10),
};

try
{
    var result1 = await context.CallActivityAsync<string>("Step1", wfInput, retryOptions);
    var result2 = await context.CallActivityAsync<byte[]>("Step2", result1, retryOptions);
    var result3 = await context.CallActivityAsync<long[]>("Step3", result2, retryOptions);
    return string.Join(", ", result4);
}
catch (TaskFailedException) // Task failures are surfaced as TaskFailedException
{
    // Retries expired - apply custom compensation logic
    await context.CallActivityAsync<long[]>("MyCompensation", options: retryOptions);
    throw;
}

Note 在上面的示例中,"Step1""Step2""Step3""MyCompensation" 表示工作流活动,这些是实际实现工作流步骤的代码中的函数。为简洁起见,这些活动的实现未包含在此示例中。

public class ChainWorkflow extends Workflow {
    @Override
    public WorkflowStub create() {
        return ctx -> {
            StringBuilder sb = new StringBuilder();
            String wfInput = ctx.getInput(String.class);
            String result1 = ctx.callActivity("Step1", wfInput, String.class).await();
            String result2 = ctx.callActivity("Step2", result1, String.class).await();
            String result3 = ctx.callActivity("Step3", result2, String.class).await();
            String result = sb.append(result1).append(',').append(result2).append(',').append(result3).toString();
            ctx.complete(result);
        };
    }
}

    class Step1 implements WorkflowActivity {

        @Override
        public Object run(WorkflowActivityContext ctx) {
            Logger logger = LoggerFactory.getLogger(Step1.class);
            logger.info("Starting Activity: " + ctx.getName());
            // Do some work
            return null;
        }
    }

    class Step2 implements WorkflowActivity {

        @Override
        public Object run(WorkflowActivityContext ctx) {
            Logger logger = LoggerFactory.getLogger(Step2.class);
            logger.info("Starting Activity: " + ctx.getName());
            // Do some work
            return null;
        }
    }

    class Step3 implements WorkflowActivity {

        @Override
        public Object run(WorkflowActivityContext ctx) {
            Logger logger = LoggerFactory.getLogger(Step3.class);
            logger.info("Starting Activity: " + ctx.getName());
            // Do some work
            return null;
        }
    }
func TaskChainWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	var result1 int
	if err := ctx.CallActivity(Step1, workflow.ActivityInput(input)).Await(&result1); err != nil {
		return nil, err
	}
	var result2 int
	if err := ctx.CallActivity(Step2, workflow.ActivityInput(input)).Await(&result2); err != nil {
		return nil, err
	}
	var result3 int
	if err := ctx.CallActivity(Step3, workflow.ActivityInput(input)).Await(&result3); err != nil {
		return nil, err
	}
	return []int{result1, result2, result3}, nil
}
func Step1(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	fmt.Printf("Step 1: Received input: %s", input)
	return input + 1, nil
}
func Step2(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	fmt.Printf("Step 2: Received input: %s", input)
	return input * 2, nil
}
func Step3(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	fmt.Printf("Step 3: Received input: %s", input)
	return int(math.Pow(float64(input), 2)), nil
}

如您所见,工作流被表示为您选择的编程语言中的一系列简单语句。这使得组织中的任何工程师都可以快速理解端到端流程,而无需一定理解端到端系统架构。

在幕后,Dapr 工作流运行时:

  • 负责执行工作流并确保其运行完成。
  • 自动保存进度。
  • 如果工作流进程因任何原因失败,会自动从最后完成的步骤恢复工作流。
  • 允许用目标编程语言自然地表达错误处理,使您可以轻松实现补偿逻辑。
  • 提供内置的重试配置原语,以简化为工作流中的各个步骤配置复杂重试策略的过程。

扇出/扇入

在扇出/扇入设计模式中,您跨多个工作器同时执行多个任务,等待它们完成,然后对结果执行某种聚合。

显示扇出/扇入工作流模式如何工作的示意图

除了前面模式中提到的挑战之外,手动实现扇出/扇入模式时还有几个重要问题需要考虑:

  • 您如何控制并行度?
  • 您如何知道何时触发后续聚合步骤?
  • 如果并行步骤的数量是动态的怎么办?

Dapr 工作流提供了一种将扇出/扇入模式表示为简单函数的方法,如以下示例所示:

# Start the workflow
dapr workflow run DataProcessingWorkflow \
  --app-id processor \
  --input '{"items": ["item1", "item2", "item3"]}'

# Monitor parallel execution
dapr workflow history <instance-id> --app-id processor --output json
import time
from typing import List
import dapr.ext.workflow as wf


def batch_processing_workflow(ctx: wf.DaprWorkflowContext, wf_input: int):
    # get a batch of N work items to process in parallel
    work_batch = yield ctx.call_activity(get_work_batch, input=wf_input)

    # schedule N parallel tasks to process the work items and wait for all to complete
    parallel_tasks = [ctx.call_activity(process_work_item, input=work_item) for work_item in work_batch]
    outputs = yield wf.when_all(parallel_tasks)

    # aggregate the results and send them to another activity
    total = sum(outputs)
    yield ctx.call_activity(process_results, input=total)


def get_work_batch(ctx, batch_size: int) -> List[int]:
    return [i + 1 for i in range(batch_size)]


def process_work_item(ctx, work_item: int) -> int:
    print(f'Processing work item: {work_item}.')
    time.sleep(5)
    result = work_item * 2
    print(f'Work item {work_item} processed. Result: {result}.')
    return result


def process_results(ctx, final_result: int):
    print(f'Final result: {final_result}.')
import {
  Task,
  DaprWorkflowClient,
  WorkflowActivityContext,
  WorkflowContext,
  WorkflowRuntime,
  TWorkflow,
} from "@dapr/dapr";

// Wrap the entire code in an immediately-invoked async function
async function start() {
  // Update the gRPC client and worker to use a local address and port
  const daprHost = "localhost";
  const daprPort = "50001";
  const workflowClient = new DaprWorkflowClient({
    daprHost,
    daprPort,
  });
  const workflowRuntime = new WorkflowRuntime({
    daprHost,
    daprPort,
  });

  function getRandomInt(min: number, max: number): number {
    return Math.floor(Math.random() * (max - min + 1)) + min;
  }

  async function getWorkItemsActivity(_: WorkflowActivityContext): Promise<string[]> {
    const count: number = getRandomInt(2, 10);
    console.log(`generating ${count} work items...`);

    const workItems: string[] = Array.from({ length: count }, (_, i) => `work item ${i}`);
    return workItems;
  }

  function sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  async function processWorkItemActivity(context: WorkflowActivityContext, item: string): Promise<number> {
    console.log(`processing work item: ${item}`);

    // Simulate some work that takes a variable amount of time
    const sleepTime = Math.random() * 5000;
    await sleep(sleepTime);

    // Return a result for the given work item, which is also a random number in this case
    // For more information about random numbers in workflow please check
    // https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-code-constraints?tabpane=csharp#random-numbers
    return Math.floor(Math.random() * 11);
  }

  const workflow: TWorkflow = async function* (ctx: WorkflowContext): any {
    const tasks: Task<any>[] = [];
    const workItems = yield ctx.callActivity(getWorkItemsActivity);
    for (const workItem of workItems) {
      tasks.push(ctx.callActivity(processWorkItemActivity, workItem));
    }
    const results: number[] = yield ctx.whenAll(tasks);
    const sum: number = results.reduce((accumulator, currentValue) => accumulator + currentValue, 0);
    return sum;
  };

  workflowRuntime.registerWorkflow(workflow);
  workflowRuntime.registerActivity(getWorkItemsActivity);
  workflowRuntime.registerActivity(processWorkItemActivity);

  // Wrap the worker startup in a try-catch block to handle any errors during startup
  try {
    await workflowRuntime.start();
    console.log("Worker started successfully");
  } catch (error) {
    console.error("Error starting worker:", error);
  }

  // Schedule a new orchestration
  try {
    const id = await workflowClient.scheduleNewWorkflow(workflow);
    console.log(`Orchestration scheduled with ID: ${id}`);

    // Wait for orchestration completion
    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);

    console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
  } catch (error) {
    console.error("Error scheduling or waiting for orchestration:", error);
  }

  // stop worker and client
  await workflowRuntime.stop();
  await workflowClient.stop();

  // stop the dapr side car
  process.exit(0);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});
// Get a list of N work items to process in parallel.
object[] workBatch = await context.CallActivityAsync<object[]>("GetWorkBatch", null);

// Schedule the parallel tasks, but don't wait for them to complete yet.
var parallelTasks = new List<Task<int>>(workBatch.Length);
for (int i = 0; i < workBatch.Length; i++)
{
    Task<int> task = context.CallActivityAsync<int>("ProcessWorkItem", workBatch[i]);
    parallelTasks.Add(task);
}

// Everything is scheduled. Wait here until all parallel tasks have completed.
await Task.WhenAll(parallelTasks);

// Aggregate all N outputs and publish the result.
int sum = parallelTasks.Sum(t => t.Result);
await context.CallActivityAsync("PostResults", sum);
public class FaninoutWorkflow extends Workflow {
    @Override
    public WorkflowStub create() {
        return ctx -> {
            // Get a list of N work items to process in parallel.
            Object[] workBatch = ctx.callActivity("GetWorkBatch", Object[].class).await();
            // Schedule the parallel tasks, but don't wait for them to complete yet.
            List<Task<Integer>> tasks = Arrays.stream(workBatch)
                    .map(workItem -> ctx.callActivity("ProcessWorkItem", workItem, int.class))
                    .collect(Collectors.toList());
            // Everything is scheduled. Wait here until all parallel tasks have completed.
            List<Integer> results = ctx.allOf(tasks).await();
            // Aggregate all N outputs and publish the result.
            int sum = results.stream().mapToInt(Integer::intValue).sum();
            ctx.complete(sum);
        };
    }
}
func BatchProcessingWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return 0, err
	}
	var workBatch []int
	if err := ctx.CallActivity(GetWorkBatch, workflow.ActivityInput(input)).Await(&workBatch); err != nil {
		return 0, err
	}
	parallelTasks := workflow.NewTaskSlice(len(workBatch))
	for i, workItem := range workBatch {
		parallelTasks[i] = ctx.CallActivity(ProcessWorkItem, workflow.ActivityInput(workItem))
	}
	var outputs int
	for _, task := range parallelTasks {
		var output int
		err := task.Await(&output)
		if err == nil {
			outputs += output
		} else {
			return 0, err
		}
	}
	if err := ctx.CallActivity(ProcessResults, workflow.ActivityInput(outputs)).Await(nil); err != nil {
		return 0, err
	}
	return 0, nil
}
func GetWorkBatch(ctx workflow.ActivityContext) (any, error) {
	var batchSize int
	if err := ctx.GetInput(&batchSize); err != nil {
		return 0, err
	}
	batch := make([]int, batchSize)
	for i := 0; i < batchSize; i++ {
		batch[i] = i
	}
	return batch, nil
}
func ProcessWorkItem(ctx workflow.ActivityContext) (any, error) {
	var workItem int
	if err := ctx.GetInput(&workItem); err != nil {
		return 0, err
	}
	fmt.Printf("Processing work item: %d\n", workItem)
	time.Sleep(time.Second * 5)
	result := workItem * 2
	fmt.Printf("Work item %d processed. Result: %d\n", workItem, result)
	return result, nil
}
func ProcessResults(ctx workflow.ActivityContext) (any, error) {
	var finalResult int
	if err := ctx.GetInput(&finalResult); err != nil {
		return 0, err
	}
	fmt.Printf("Final result: %d\n", finalResult)
	return finalResult, nil
}

此示例的关键要点是:

  • 扇出/扇入模式可以使用普通编程构造表示为简单的函数
  • 并行任务的数量可以是静态的或动态的
  • 工作流本身能够聚合并行执行的结果

此外,工作流的执行是持久的。如果一个工作流启动 100 个并行任务执行,并且在进程崩溃之前只完成了 40 个,工作流会自动重新启动,并且只安排剩余的 60 个任务。

您可以进一步使用简单的特定语言构造来限制并发度。下面的示例代码说明了如何将扇出度限制为仅 5 个并发活动执行:


//Revisiting the earlier example...
// Get a list of N work items to process in parallel.
object[] workBatch = await context.CallActivityAsync<object[]>("GetWorkBatch", null);

const int MaxParallelism = 5;
var results = new List<int>();
var inFlightTasks = new HashSet<Task<int>>();
foreach(var workItem in workBatch)
{
  if (inFlightTasks.Count >= MaxParallelism)
  {
    var finishedTask = await Task.WhenAny(inFlightTasks);
    results.Add(finishedTask.Result);
    inFlightTasks.Remove(finishedTask);
  }

  inFlightTasks.Add(context.CallActivityAsync<int>("ProcessWorkItem", workItem));
}
results.AddRange(await Task.WhenAll(inFlightTasks));

var sum = results.Sum(t => t);
await context.CallActivityAsync("PostResults", sum);

您可以通过使用 WorkflowContext 上的以下扩展方法来并行处理工作流活动,同时对并发性设置上限:

//Revisiting the earlier example...
// Get a list of work items to process
var workBatch = await context.CallActivityAsync<object[]>("GetWorkBatch", null);

// Process deterministically in parallel with an upper cap of 5 activities at a time
var results = await context.ProcessInParallelAsync(workBatch, workItem => context.CallActivityAsync<int>("ProcessWorkItem", workItem), maxConcurrency: 5);

var sum = results.Sum(t => t);
await context.CallActivityAsync("PostResults", sum);

以这种方式限制并发度对于限制对共享资源的争用非常有用。例如,如果活动需要调用具有自己并发限制的外部资源(如数据库或外部 API),确保不超过指定数量的活动同时调用该资源会很有用。

异步 HTTP API

异步 HTTP API 通常使用异步请求-回复模式来实现。传统实现此模式涉及以下内容:

  1. 客户端向 HTTP API 端点发送请求(启动 API
  2. 启动 API 将消息写入后端队列,触发长时间运行操作的开始
  3. 在安排后端操作后,启动 API 立即向客户端返回 HTTP 202 响应,其中包含可用于轮询状态的标识符
  4. 状态 API 查询包含长时间运行操作状态的数据库
  5. 客户端重复轮询 状态 API,直到某个超时过期或收到"完成"响应

端到端流程如下图所示。

显示异步请求-回复模式如何工作的示意图

实现异步请求-回复模式的挑战在于它涉及使用多个 API 和状态存储。它还涉及正确实现协议,以便客户端知道如何自动轮询状态并了解操作何时完成。

Dapr 工作流 HTTP API 开箱即用地支持异步请求-回复模式,无需您编写任何代码或进行任何状态管理。

以下 curl 命令说明了工作流 API 如何支持此模式。

curl -X POST http://localhost:3500/v1.0/workflows/dapr/OrderProcessingWorkflow/start?instanceID=12345678 -d '{"Name":"Paperclips","Quantity":1,"TotalCost":9.95}'

前面的命令将产生以下响应 JSON:

{"instanceID":"12345678"}

然后,HTTP 客户端可以使用工作流实例 ID 构造状态查询 URL,并重复轮询它,直到在有效负载中看到"COMPLETE"、“FAILURE"或"TERMINATED"状态。

curl http://localhost:3500/v1.0/workflows/dapr/12345678

以下是正在进行中的工作流状态可能看起来像的示例。

{
  "instanceID": "12345678",
  "workflowName": "OrderProcessingWorkflow",
  "createdAt": "2023-05-03T23:22:11.143069826Z",
  "lastUpdatedAt": "2023-05-03T23:22:22.460025267Z",
  "runtimeStatus": "RUNNING",
  "properties": {
    "dapr.workflow.custom_status": "",
    "dapr.workflow.input": "{\"Name\":\"Paperclips\",\"Quantity\":1,\"TotalCost\":9.95}"
  }
}

从上一个示例中可以看出,工作流的运行时状态是 RUNNING,这让客户端知道应该继续轮询。

如果工作流已完成,状态可能如下所示。

{
  "instanceID": "12345678",
  "workflowName": "OrderProcessingWorkflow",
  "createdAt": "2023-05-03T23:30:11.381146313Z",
  "lastUpdatedAt": "2023-05-03T23:30:52.923870615Z",
  "runtimeStatus": "COMPLETED",
  "properties": {
    "dapr.workflow.custom_status": "",
    "dapr.workflow.input": "{\"Name\":\"Paperclips\",\"Quantity\":1,\"TotalCost\":9.95}",
    "dapr.workflow.output": "{\"Processed\":true}"
  }
}

从上一个示例中可以看出,工作流的运行时状态现在是 COMPLETED,这意味着客户端可以停止轮询更新。

监视器

监视器模式是一个循环过程,通常:

  1. 检查系统状态
  2. 根据该状态采取某些操作 - 例如,发送通知
  3. 休眠一段时间
  4. 重复

下图大致说明了此模式。

显示监视器模式如何工作的示意图

根据业务需求,可能只有一个监视器,也可能有多个监视器,每个业务实体一个(例如,股票)。此外,根据情况,休眠的时间可能需要更改。这些要求使得使用基于 cron 的调度系统变得不切实际。

Dapr 工作流通过允许您实现_永久工作流_来原生支持此模式。无需编写无限 while 循环(这是一种反模式),Dapr 工作流公开了一个 continue-as-new API,工作流作者可以使用它从头开始使用新输入重新启动工作流函数。

from dataclasses import dataclass
from datetime import timedelta
import random
import dapr.ext.workflow as wf


@dataclass
class JobStatus:
    job_id: str
    is_healthy: bool


def status_monitor_workflow(ctx: wf.DaprWorkflowContext, job: JobStatus):
    # poll a status endpoint associated with this job
    status = yield ctx.call_activity(check_status, input=job)
    if not ctx.is_replaying:
        print(f"Job '{job.job_id}' is {status}.")

    if status == "healthy":
        job.is_healthy = True
        next_sleep_interval = 60  # check less frequently when healthy
    else:
        if job.is_healthy:
            job.is_healthy = False
            ctx.call_activity(send_alert, input=f"Job '{job.job_id}' is unhealthy!")
        next_sleep_interval = 5  # check more frequently when unhealthy

    yield ctx.create_timer(fire_at=ctx.current_utc_datetime + timedelta(minutes=next_sleep_interval))

    # restart from the beginning with a new JobStatus input
    ctx.continue_as_new(job)


def check_status(ctx, _) -> str:
    return random.choice(["healthy", "unhealthy"])


def send_alert(ctx, message: str):
    print(f'*** Alert: {message}')
const statusMonitorWorkflow: TWorkflow = async function* (ctx: WorkflowContext): any {
    let duration;
    const status = yield ctx.callActivity(checkStatusActivity);
    if (status === "healthy") {
      // Check less frequently when in a healthy state
      // set duration to 1 hour
      duration = 60 * 60;
    } else {
      yield ctx.callActivity(alertActivity, "job unhealthy");
      // Check more frequently when in an unhealthy state
      // set duration to 5 minutes
      duration = 5 * 60;
    }

    // Put the workflow to sleep until the determined time
    ctx.createTimer(duration);

    // Restart from the beginning with the updated state
    ctx.continueAsNew();
  };
public override async Task<object> RunAsync(WorkflowContext context, MyEntityState myEntityState)
{
    TimeSpan nextSleepInterval;

    var status = await context.CallActivityAsync<string>("GetStatus");
    if (status == "healthy")
    {
        myEntityState.IsHealthy = true;

        // Check less frequently when in a healthy state
        nextSleepInterval = TimeSpan.FromMinutes(60);
    }
    else
    {
        if (myEntityState.IsHealthy)
        {
            myEntityState.IsHealthy = false;
            await context.CallActivityAsync("SendAlert", myEntityState);
        }

        // Check more frequently when in an unhealthy state
        nextSleepInterval = TimeSpan.FromMinutes(5);
    }

    // Put the workflow to sleep until the determined time
    await context.CreateTimer(nextSleepInterval);

    // Restart from the beginning with the updated state
    context.ContinueAsNew(myEntityState);
    return null;
}

此示例假设您有一个预定义的 MyEntityState 类,其中包含布尔 IsHealthy 属性。

public class MonitorWorkflow extends Workflow {

  @Override
  public WorkflowStub create() {
    return ctx -> {

      Duration nextSleepInterval;

      var status = ctx.callActivity(DemoWorkflowStatusActivity.class.getName(), DemoStatusActivityOutput.class).await();
      var isHealthy = status.getIsHealthy();

      if (isHealthy) {
        // Check less frequently when in a healthy state
        nextSleepInterval = Duration.ofMinutes(60);
      } else {

        ctx.callActivity(DemoWorkflowAlertActivity.class.getName()).await();

        // Check more frequently when in an unhealthy state
        nextSleepInterval = Duration.ofMinutes(5);
      }

      // Put the workflow to sleep until the determined time
      try {
        ctx.createTimer(nextSleepInterval);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }

      // Restart from the beginning with the updated state
      ctx.continueAsNew();
    }
  }
}
type JobStatus struct {
	JobID     string `json:"job_id"`
	IsHealthy bool   `json:"is_healthy"`
}
func StatusMonitorWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var sleepInterval time.Duration
	var job JobStatus
	if err := ctx.GetInput(&job); err != nil {
		return "", err
	}
	var status string
	if err := ctx.CallActivity(CheckStatus, workflow.ActivityInput(job)).Await(&status); err != nil {
		return "", err
	}
	if status == "healthy" {
		job.IsHealthy = true
		sleepInterval = time.Minutes * 60
	} else {
		if job.IsHealthy {
			job.IsHealthy = false
			err := ctx.CallActivity(SendAlert, workflow.ActivityInput(fmt.Sprintf("Job '%s' is unhealthy!", job.JobID))).Await(nil)
			if err != nil {
				return "", err
			}
		}
		sleepInterval = time.Minutes * 5
	}
	if err := ctx.CreateTimer(sleepInterval).Await(nil); err != nil {
		return "", err
	}
	ctx.ContinueAsNew(job, false)
	return "", nil
}
func CheckStatus(ctx workflow.ActivityContext) (any, error) {
	statuses := []string{"healthy", "unhealthy"}
	return statuses[rand.Intn(1)], nil
}
func SendAlert(ctx workflow.ActivityContext) (any, error) {
	var message string
	if err := ctx.GetInput(&message); err != nil {
		return "", err
	}
	fmt.Printf("*** Alert: %s", message)
	return "", nil
}

实现监视器模式的工作流可以永远循环,也可以通过不调用 continue-as-new 来优雅地终止自己。

外部系统交互

在某些情况下,工作流可能需要暂停并等待外部系统执行某些操作。例如,工作流可能需要暂停并等待收到付款。在这种情况下,支付系统可能会在收到付款时向发布/订阅主题发布事件,并且该主题上的侦听器可以使用引发事件工作流 API向工作流引发事件。

另一个非常常见的场景是工作流需要暂停并等待人员,例如在批准采购订单时。Dapr 工作流通过外部事件功能支持此事件模式。

以下是涉及人员的采购订单的工作流示例:

  1. 收到采购订单时触发工作流。
  2. 工作流中的规则确定需要人员执行某些操作。例如,采购订单成本超过某个自动批准阈值。
  3. 工作流发送请求人员操作的通知。例如,它向指定的批准者发送包含批准链接的电子邮件。
  4. 工作流暂停并等待人员通过单击链接来批准或拒绝订单。
  5. 如果在指定时间内未收到批准,工作流将恢复并执行某些补偿逻辑,例如取消订单。

下图说明了此流程。

显示涉及人员的外部系统交互模式如何工作的示意图

以下示例代码显示了如何使用 Dapr 工作流实现此模式。

from dataclasses import dataclass
from datetime import timedelta
import dapr.ext.workflow as wf


@dataclass
class Order:
    cost: float
    product: str
    quantity: int

    def __str__(self):
        return f'{self.product} ({self.quantity})'


@dataclass
class Approval:
    approver: str

    @staticmethod
    def from_dict(dict):
        return Approval(**dict)


def purchase_order_workflow(ctx: wf.DaprWorkflowContext, order: Order):
    # Orders under $1000 are auto-approved
    if order.cost < 1000:
        return "Auto-approved"

    # Orders of $1000 or more require manager approval
    yield ctx.call_activity(send_approval_request, input=order)

    # Approvals must be received within 24 hours or they will be canceled.
    approval_event = ctx.wait_for_external_event("approval_received")
    timeout_event = ctx.create_timer(timedelta(hours=24))
    winner = yield wf.when_any([approval_event, timeout_event])
    if winner == timeout_event:
        return "Cancelled"

    # The order was approved
    yield ctx.call_activity(place_order, input=order)
    approval_details = Approval.from_dict(approval_event.get_result())
    return f"Approved by '{approval_details.approver}'"


def send_approval_request(_, order: Order) -> None:
    print(f'*** Sending approval request for order: {order}')


def place_order(_, order: Order) -> None:
    print(f'*** Placing order: {order}')
import {
  Task,
  DaprWorkflowClient,
  WorkflowActivityContext,
  WorkflowContext,
  WorkflowRuntime,
  TWorkflow,
} from "@dapr/dapr";
import * as readlineSync from "readline-sync";

// Wrap the entire code in an immediately-invoked async function
async function start() {
  class Order {
    cost: number;
    product: string;
    quantity: number;
    constructor(cost: number, product: string, quantity: number) {
      this.cost = cost;
      this.product = product;
      this.quantity = quantity;
    }
  }

  function sleep(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  // Update the gRPC client and worker to use a local address and port
  const daprHost = "localhost";
  const daprPort = "50001";
  const workflowClient = new DaprWorkflowClient({
    daprHost,
    daprPort,
  });
  const workflowRuntime = new WorkflowRuntime({
    daprHost,
    daprPort,
  });

  // Activity function that sends an approval request to the manager
  const sendApprovalRequest = async (_: WorkflowActivityContext, order: Order) => {
    // Simulate some work that takes an amount of time
    await sleep(3000);
    console.log(`Sending approval request for order: ${order.product}`);
  };

  // Activity function that places an order
  const placeOrder = async (_: WorkflowActivityContext, order: Order) => {
    console.log(`Placing order: ${order.product}`);
  };

  // Orchestrator function that represents a purchase order workflow
  const purchaseOrderWorkflow: TWorkflow = async function* (ctx: WorkflowContext, order: Order): any {
    // Orders under $1000 are auto-approved
    if (order.cost < 1000) {
      return "Auto-approved";
    }

    // Orders of $1000 or more require manager approval
    yield ctx.callActivity(sendApprovalRequest, order);

    // Approvals must be received within 24 hours or they will be cancled.
    const tasks: Task<any>[] = [];
    const approvalEvent = ctx.waitForExternalEvent("approval_received");
    const timeoutEvent = ctx.createTimer(24 * 60 * 60);
    tasks.push(approvalEvent);
    tasks.push(timeoutEvent);
    const winner = ctx.whenAny(tasks);

    if (winner == timeoutEvent) {
      return "Cancelled";
    }

    yield ctx.callActivity(placeOrder, order);
    const approvalDetails = approvalEvent.getResult();
    return `Approved by ${approvalDetails.approver}`;
  };

  workflowRuntime
    .registerWorkflow(purchaseOrderWorkflow)
    .registerActivity(sendApprovalRequest)
    .registerActivity(placeOrder);

  // Wrap the worker startup in a try-catch block to handle any errors during startup
  try {
    await workflowRuntime.start();
    console.log("Worker started successfully");
  } catch (error) {
    console.error("Error starting worker:", error);
  }

  // Schedule a new orchestration
  try {
    const cost = readlineSync.questionInt("Cost of your order:");
    const approver = readlineSync.question("Approver of your order:");
    const timeout = readlineSync.questionInt("Timeout for your order in seconds:");
    const order = new Order(cost, "MyProduct", 1);
    const id = await workflowClient.scheduleNewWorkflow(purchaseOrderWorkflow, order);
    console.log(`Orchestration scheduled with ID: ${id}`);

    // prompt for approval asynchronously
    promptForApproval(approver, workflowClient, id);

    // Wait for orchestration completion
    const state = await workflowClient.waitForWorkflowCompletion(id, undefined, timeout + 2);

    console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
  } catch (error) {
    console.error("Error scheduling or waiting for orchestration:", error);
  }

  // stop worker and client
  await workflowRuntime.stop();
  await workflowClient.stop();

  // stop the dapr side car
  process.exit(0);
}

async function promptForApproval(approver: string, workflowClient: DaprWorkflowClient, id: string) {
  if (readlineSync.keyInYN("Press [Y] to approve the order... Y/yes, N/no")) {
    const approvalEvent = { approver: approver };
    await workflowClient.raiseEvent(id, "approval_received", approvalEvent);
  } else {
    return "Order rejected";
  }
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});
public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderPayload order)
{
    // ...(other steps)...

    // Require orders over a certain threshold to be approved
    if (order.TotalCost > OrderApprovalThreshold)
    {
        try
        {
            // Request human approval for this order
            await context.CallActivityAsync(nameof(RequestApprovalActivity), order);

            // Pause and wait for a human to approve the order
            ApprovalResult approvalResult = await context.WaitForExternalEventAsync<ApprovalResult>(
                eventName: "ManagerApproval",
                timeout: TimeSpan.FromDays(3));
            if (approvalResult == ApprovalResult.Rejected)
            {
                // The order was rejected, end the workflow here
                return new OrderResult(Processed: false);
            }
        }
        catch (TaskCanceledException)
        {
            // An approval timeout results in automatic order cancellation
            return new OrderResult(Processed: false);
        }
    }

    // ...(other steps)...

    // End the workflow with a success result
    return new OrderResult(Processed: true);
}

Note 在上面的示例中,RequestApprovalActivity 是要调用的工作流活动的名称,而 ApprovalResult 是由工作流应用定义的枚举。为简洁起见,这些定义未包含在示例代码中。

public class ExternalSystemInteractionWorkflow extends Workflow {
    @Override
    public WorkflowStub create() {
        return ctx -> {
            // ...other steps...
            Integer orderCost = ctx.getInput(int.class);
            // Require orders over a certain threshold to be approved
            if (orderCost > ORDER_APPROVAL_THRESHOLD) {
                try {
                    // Request human approval for this order
                    ctx.callActivity("RequestApprovalActivity", orderCost, Void.class).await();
                    // Pause and wait for a human to approve the order
                    boolean approved = ctx.waitForExternalEvent("ManagerApproval", Duration.ofDays(3), boolean.class).await();
                    if (!approved) {
                        // The order was rejected, end the workflow here
                        ctx.complete("Process reject");
                    }
                } catch (TaskCanceledException e) {
                    // An approval timeout results in automatic order cancellation
                    ctx.complete("Process cancel");
                }
            }
            // ...other steps...

            // End the workflow with a success result
            ctx.complete("Process approved");
        };
    }
}
type Order struct {
	Cost     float64 `json:"cost"`
	Product  string  `json:"product"`
	Quantity int     `json:"quantity"`
}
type Approval struct {
	Approver string `json:"approver"`
}
func PurchaseOrderWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var order Order
	if err := ctx.GetInput(&order); err != nil {
		return "", err
	}
	// Orders under $1000 are auto-approved
	if order.Cost < 1000 {
		return "Auto-approved", nil
	}
	// Orders of $1000 or more require manager approval
	if err := ctx.CallActivity(SendApprovalRequest, workflow.ActivityInput(order)).Await(nil); err != nil {
		return "", err
	}
	// Approvals must be received within 24 hours or they will be cancelled
	var approval Approval
	if err := ctx.WaitForExternalEvent("approval_received", time.Hour*24).Await(&approval); err != nil {
		// Assuming that a timeout has taken place - in any case; an error.
		return "error/cancelled", err
	}
	// The order was approved
	if err := ctx.CallActivity(PlaceOrder, workflow.ActivityInput(order)).Await(nil); err != nil {
		return "", err
	}
	return fmt.Sprintf("Approved by %s", approval.Approver), nil
}
func SendApprovalRequest(ctx workflow.ActivityContext) (any, error) {
	var order Order
	if err := ctx.GetInput(&order); err != nil {
		return "", err
	}
	fmt.Printf("*** Sending approval request for order: %v\n", order)
	return "", nil
}
func PlaceOrder(ctx workflow.ActivityContext) (any, error) {
	var order Order
	if err := ctx.GetInput(&order); err != nil {
		return "", err
	}
	fmt.Printf("*** Placing order: %v", order)
	return "", nil
}

向等待的工作流实例传递事件以恢复工作流执行的代码在工作流外部。可以使用引发事件工作流管理 API 将工作流事件传递给等待的工作流实例,如以下示例所示:

from dapr.clients import DaprClient
from dataclasses import asdict

with DaprClient() as d:
    d.raise_workflow_event(
        instance_id=instance_id,
        workflow_component="dapr",
        event_name="approval_received",
        event_data=asdict(Approval("Jane Doe")))
import { DaprClient } from "@dapr/dapr";

  public async raiseEvent(workflowInstanceId: string, eventName: string, eventPayload?: any) {
    this._innerClient.raiseOrchestrationEvent(workflowInstanceId, eventName, eventPayload);
  }
// Raise the workflow event to the waiting workflow
await daprClient.RaiseWorkflowEventAsync(
    instanceId: orderId,
    workflowComponent: "dapr",
    eventName: "ManagerApproval",
    eventData: ApprovalResult.Approved);
System.out.println("**SendExternalMessage: RestartEvent**");
client.raiseEvent(restartingInstanceId, "RestartEvent", "RestartEventPayload");
func raiseEvent() {
  daprClient, err := client.NewClient()
  if err != nil {
    log.Fatalf("failed to initialize the client")
  }
  err = daprClient.RaiseEventWorkflow(context.Background(), &client.RaiseEventWorkflowRequest{
    InstanceID: "instance_id",
    WorkflowComponent: "dapr",
    EventName: "approval_received",
    EventData: Approval{
      Approver: "Jane Doe",
    },
  })
  if err != nil {
    log.Fatalf("failed to raise event on workflow")
  }
  log.Println("raised an event on specified workflow")
}

外部事件不必直接由人员触发。它们也可以由其他系统触发。例如,工作流可能需要暂停并等待收到付款。在这种情况下,支付系统可能会在收到付款时向发布/订阅主题发布事件,并且该主题上的侦听器可以使用引发事件工作流 API 向工作流引发事件。

补偿

补偿模式(也称为 Saga 模式)提供了一种在工作流中途失败时回滚或撤销已执行操作的机制。此模式对于跨越多个微服务且传统数据库事务不可行的长时间运行的工作流特别重要。

在分布式微服务架构中,您通常需要跨多个服务协调操作。当这些操作无法包含在单个事务中时,补偿模式通过为工作流中的每个步骤定义补偿操作来提供一种保持一致性的方法。

补偿模式解决了几个关键挑战:

  • 分布式事务管理:当工作流跨越多个微服务时,每个微服务都有自己的数据存储,传统的 ACID 事务是不可能的。补偿模式通过确保操作要么全部成功完成,要么通过补偿全部撤销来提供事务一致性。
  • 部分故障恢复:如果工作流在某些步骤成功完成后失败,补偿模式允许您优雅地撤销这些已完成的步骤。
  • 业务流程完整性:确保业务流程可以在故障时正确回滚,从而保持业务操作的完整性。
  • 长时间运行的流程:对于可能运行数小时、数天或更长时间的工作流,传统的锁定机制是不切实际的。补偿提供了一种在这些场景中处理故障的方法。

补偿模式的常见用例包括:

  • 电子商务订单处理:预留库存、处理付款和发货订单。如果发货失败,您需要释放库存并退还付款。
  • 金融交易:在转账中,如果贷记目标账户失败,您需要回滚对源账户的借记。
  • 资源预配:在跨多个提供程序预配云资源时,如果某个步骤失败,您需要清理所有以前预配的资源。
  • 多步骤业务流程:任何涉及多个不可逆步骤的业务流程,这些步骤可能需要在后续失败时撤销。

Dapr 工作流提供对补偿模式的支持,允许您为每个步骤注册补偿活动,并在需要时以相反的顺序执行它们。

以下是电子商务流程的工作流示例:

  1. 收到订单时触发工作流。
  2. 在库存中为订单进行预留。
  3. 处理付款。
  4. 发货订单。
  5. 如果上述任何操作导致错误,则使用另一个操作进行补偿:
    • 取消发货。
    • 退还付款。
    • 释放库存预留。

下图说明了此流程。

显示补偿模式的示意图。
public class PaymentProcessingWorkflow implements Workflow {

    @Override
    public WorkflowStub create() {
        return ctx -> {
            ctx.getLogger().info("Starting Workflow: " + ctx.getName());
            var orderId = ctx.getInput(String.class);
            List<String> compensations = new ArrayList<>();

            try {
                // Step 1: Reserve inventory
                String reservationId = ctx.callActivity(ReserveInventoryActivity.class.getName(), orderId, String.class).await();
                ctx.getLogger().info("Inventory reserved: {}", reservationId);
                compensations.add("ReleaseInventory");

                // Step 2: Process payment
                String paymentId = ctx.callActivity(ProcessPaymentActivity.class.getName(), orderId, String.class).await();
                ctx.getLogger().info("Payment processed: {}", paymentId);
                compensations.add("RefundPayment");

                // Step 3: Ship order
                String shipmentId = ctx.callActivity(ShipOrderActivity.class.getName(), orderId, String.class).await();
                ctx.getLogger().info("Order shipped: {}", shipmentId);
                compensations.add("CancelShipment");

            } catch (TaskFailedException e) {
                ctx.getLogger().error("Activity failed: {}", e.getMessage());

                // Execute compensations in reverse order
                Collections.reverse(compensations);
                for (String compensation : compensations) {
                    try {
                        switch (compensation) {
                            case "CancelShipment":
                                String shipmentCancelResult = ctx.callActivity(
                                    CancelShipmentActivity.class.getName(),
                                    orderId,
                                    String.class).await();
                                ctx.getLogger().info("Shipment cancellation completed: {}", shipmentCancelResult);
                                break;

                            case "RefundPayment":
                                String refundResult = ctx.callActivity(
                                    RefundPaymentActivity.class.getName(),
                                    orderId,
                                    String.class).await();
                                ctx.getLogger().info("Payment refund completed: {}", refundResult);
                                break;

                            case "ReleaseInventory":
                                String releaseResult = ctx.callActivity(
                                    ReleaseInventoryActivity.class.getName(),
                                    orderId,
                                    String.class).await();
                                ctx.getLogger().info("Inventory release completed: {}", releaseResult);
                                break;
                        }
                    } catch (TaskFailedException ex) {
                        ctx.getLogger().error("Compensation activity failed: {}", ex.getMessage());
                    }
                }
                ctx.complete("Order processing failed, compensation applied");
            }

			// Step 4: Send confirmation
			ctx.callActivity(SendConfirmationActivity.class.getName(), orderId, Void.class).await();
            ctx.getLogger().info("Confirmation sent for order: {}", orderId);

            ctx.complete("Order processed successfully: " + orderId);
        };
    }
}

// Example activities
class ReserveInventoryActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String orderId = ctx.getInput(String.class);
        // Logic to reserve inventory
        String reservationId = "reservation_" + orderId;
        System.out.println("Reserved inventory for order: " + orderId);
        return reservationId;
    }
}

class ReleaseInventoryActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String reservationId = ctx.getInput(String.class);
        // Logic to release inventory reservation
        System.out.println("Released inventory reservation: " + reservationId);
        return "Released: " + reservationId;
    }
}

class ProcessPaymentActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String orderId = ctx.getInput(String.class);
        // Logic to process payment
        String paymentId = "payment_" + orderId;
        System.out.println("Processed payment for order: " + orderId);
        return paymentId;
    }
}

class RefundPaymentActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String paymentId = ctx.getInput(String.class);
        // Logic to refund payment
        System.out.println("Refunded payment: " + paymentId);
        return "Refunded: " + paymentId;
    }
}

class ShipOrderActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String orderId = ctx.getInput(String.class);
        // Logic to ship order
        String shipmentId = "shipment_" + orderId;
        System.out.println("Shipped order: " + orderId);
        return shipmentId;
    }
}

class CancelShipmentActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String shipmentId = ctx.getInput(String.class);
        // Logic to cancel shipment
        System.out.println("Canceled shipment: " + shipmentId);
        return "Canceled: " + shipmentId;
    }
}

class SendConfirmationActivity implements WorkflowActivity {
    @Override
    public Object run(WorkflowActivityContext ctx) {
        String orderId = ctx.getInput(String.class);
        // Logic to send confirmation
        System.out.println("Sent confirmation for order: " + orderId);
        return null;
    }
}

使用 Dapr 工作流补偿模式的主要好处包括:

  • 补偿控制:您可以完全控制何时以及如何执行补偿活动。
  • 灵活配置:您可以实现自定义逻辑来确定要运行哪些补偿。
  • 错误处理:根据您的特定业务需求处理补偿失败。
  • 简单实现:无需额外的框架依赖 - 只需标准的工作流活动和异常处理。

补偿模式确保您的分布式工作流可以保持一致性并从故障中优雅地恢复,使其成为构建可靠微服务架构的重要工具。

后续步骤

工作流架构 >>

相关链接

5 - 工作流架构

Dapr 工作流引擎架构

Dapr 工作流允许开发者使用多种编程语言的普通代码来定义工作流。工作流引擎运行在 Dapr 边车内部,协调作为应用程序一部分部署的工作流代码。Dapr 工作流构建在 Dapr Actor 之上,为工作流执行提供持久性和可扩展性。

本文介绍:

  • Dapr 工作流引擎的架构
  • 工作流引擎如何与应用程序代码交互
  • 工作流引擎如何融入整体 Dapr 架构
  • 不同的工作流后端如何与工作流引擎配合使用

有关如何在应用程序中编写 Dapr 工作流的更多信息,请参阅如何:编写工作流

Dapr 工作流引擎内部由 Dapr 的 actor 运行时驱动。下图展示了 Kubernetes 模式下的 Dapr 工作流架构:

Diagram showing how the workflow architecture works in Kubernetes mode

要使用 Dapr 工作流构建块,您需要使用 Dapr Workflow SDK 在应用程序中编写工作流代码,SDK 内部使用 gRPC 流连接到边车。这会注册工作流以及任何工作流活动,或工作流可以调度的任务。

引擎直接嵌入到边车中,并使用 durabletask-go 框架库实现。该框架允许您交换不同的存储提供程序,包括为 Dapr 创建的存储提供程序,它在幕后利用内部 actor。由于 Dapr 工作流使用 actor,您可以将工作流状态存储在状态存储中。

边车交互

当工作流应用程序启动时,它使用工作流编写 SDK 向 Dapr 边车发送 gRPC 请求,并获取工作流工作项的流,遵循服务器流式 RPC 模式。这些工作项可以是任何内容,从"启动新的 X 工作流"(其中 X 是工作流的类型)到"代表工作流 X 调度具有输入 Z 的活动 Y"。

工作流应用程序执行适当的工作流代码,然后向边车发送 gRPC 请求,其中包含执行结果。

Dapr 工作流引擎协议

所有交互都通过单个 gRPC 通道进行,并由应用程序发起,这意味着应用程序不需要打开任何入站端口。 这些交互的细节由特定语言的 Dapr 工作流编写 SDK 内部处理。

工作流与应用程序 actor 交互的区别

如果您熟悉 Dapr actor,您可能会注意到,与应用程序定义的 actor 相比,工作流的边车交互方式有一些差异。

Actor工作流
应用程序创建的 actor 可以使用 HTTP 或 gRPC 与边车交互。工作流仅使用 gRPC。由于工作流 gRPC 协议的复杂性,实现工作流时_必须_使用 SDK。
Actor 操作从边车推送到应用程序代码。这要求应用程序监听特定的_应用端口_。对于工作流,操作由应用程序使用流式协议从边车_拉取_。应用程序不需要监听任何端口即可运行工作流。
Actor 向边车显式注册自己。工作流不向边车注册自己。嵌入式引擎不跟踪工作流类型。此职责委托给工作流应用程序及其 SDK。

工作流分布式跟踪

工作流引擎使用的 durabletask-go 核心使用 Open Telemetry SDK 编写分布式跟踪。 这些跟踪由 Dapr 边车自动捕获,并导出到配置的 Open Telemetry 提供程序,例如 Zipkin。

引擎管理的每个工作流实例表示为一个或多个 span。 有一个表示完整工作流执行的父 span,以及各种任务的子 span,包括活动任务执行和持久化计时器的 span。

工作流活动代码当前_不能_访问跟踪上下文。

工作流 actor

当工作流客户端连接到边车时,会注册两种类型的 actor 以支持工作流引擎:

  • dapr.internal.{namespace}.{appID}.workflow
  • dapr.internal.{namespace}.{appID}.activity

{namespace} 值是 Dapr 命名空间,如果未配置命名空间,则默认为 default{appID} 值是应用程序的 ID。 例如,如果您有一个名为"wfapp"的工作流应用程序,则工作流 actor 的类型为 dapr.internal.default.wfapp.workflow,活动 actor 的类型为 dapr.internal.default.wfapp.activity

下图演示了工作流 actor 在 Kubernetes 场景中如何运行:

Diagram demonstrating internally registered actors across a cluster

与用户定义的 actor 一样,工作流 actor 通过 actor 放置服务提供的哈希查找表分布在集群中。 它们还维护自己的状态并使用提醒。 但是,与应用程序代码中的 actor 不同,这些工作流 actor 嵌入到 Dapr 边车中。 应用程序代码完全不知道这些 actor 的存在。

任何隐式支持 actor 的状态存储都隐式支持 Dapr 工作流。

工作流 actor部分所述,工作流通过附加到历史日志来增量保存其状态。 工作流的历史日志分布在多个状态存储键中,以便每个"检查点"仅需要附加最新的条目。

每个检查点的大小由工作流在进入空闲状态之前调度的并发操作数确定。 顺序工作流因此会对状态存储进行较小的批量更新,而扇出/扇入工作流将需要更大的批量。 批量大小也受工作流调用活动子工作流时输入和输出大小的影响。

Diagram of workflow actor state store interactions

不同的状态存储实现可能会隐式地限制您可以编写的工作流类型。 例如,Azure Cosmos DB 状态存储将项目大小限制为 2 MB 的 UTF-8 编码 JSON(来源)。 活动或子工作流的输入或输出负载作为单个记录存储在状态存储中,因此 2 MB 的项目限制意味着工作流和活动的输入和输出不能超过 2 MB 的 JSON 序列化数据。

同样,如果状态存储对批量事务的大小施加限制,这可能会限制工作流可以调度的并行操作数量。

可以从状态存储中清除工作流状态,包括其所有历史记录。 每个 Dapr SDK 都公开了用于清除特定工作流实例的所有元数据的 API。

状态存储记录数

每次工作流运行在状态存储中保存为历史记录的记录数由其复杂性或"形状"决定。换句话说,即活动、计时器、子工作流等的数量。 下表显示了不同工作流任务保存的记录数的一般指南。 根据重试或并发性,此数字可能更大或更小。

任务类型保存的记录数
启动工作流5 条记录
调用活动3 条记录
计时器3 条记录
触发事件3 条记录
启动子工作流8 条记录

查询工作流历史

dapr workflow --app-id myapp list
dapr workflow --app-id myapp history <instance-id>

支持的状态存储

工作流引擎支持以下状态存储:

  • PostgreSQL
  • MySQL
  • SQL Server
  • SQLite
  • Oracle Database
  • CockroachDB
  • MongoDB
  • Redis

工作流可扩展性

由于 Dapr 工作流内部使用 actor 实现,Dapr 工作流具有与 actor 相同的可扩展性特征。 放置服务:

  • 不区分工作流 actor 和您在应用程序中定义的 actor
  • 将使用与 actor 相同的算法对工作流进行负载平衡

工作流的预期可扩展性由以下因素决定:

  • 用于托管工作流应用程序的计算机数量
  • 运行工作流的计算机上可用的 CPU 和内存资源
  • 为 actor 配置的状态存储的可扩展性
  • actor 放置服务和提醒子系统的可扩展性

目标应用程序中工作流代码的实现细节也在单个工作流实例的可扩展性中发挥作用。 每个工作流实例一次在单个节点上执行,但工作流可以调度在其他节点上运行的活动和子工作流。

工作流还可以调度这些活动和子工作流并行运行,允许单个工作流可能在整个集群的所有可用节点上分布计算任务。

您可以使用 Dapr 配置配置工作流和活动的最大并发性,如下一节所述。

工作流不控制负载如何在集群中分布的细节。 例如,如果工作流调度 10 个活动任务并行运行,所有 10 个任务可能运行在多达 10 个不同的计算节点上,也可能运行在单个计算节点上。 实际扩展行为由 actor 放置服务决定,该服务管理表示工作流每个任务的 actor 的分布。

Diagram of workflow and activity actors scaled out across multiple Dapr instances

工作流延迟

为了提供持久性和弹性保证,Dapr 工作流频繁写入状态存储并依赖提醒来驱动执行。 因此,Dapr 工作流可能不适合延迟敏感的工作负载。 高延迟的预期来源包括:

  • 持久化工作流状态时状态存储的延迟。
  • 使用大型历史记录重新水合工作流时状态存储的延迟。
  • 集群中太多活动提醒引起的延迟。
  • 集群中高 CPU 使用率引起的延迟。

有关工作流 actor 的设计如何影响执行延迟的更多详细信息,请参阅提醒使用和执行保证部分

提高调度吞吐量

默认情况下,当客户端调度工作流时,工作流引擎会等待工作流完全启动后再向客户端返回响应。 在返回之前等待工作流启动会降低工作流的调度吞吐量。 当调度具有开始时间的工作流时,工作流引擎不会等待工作流启动就向客户端返回响应。 要提高调度吞吐量,请考虑在调度工作流时添加"现在"的开始时间。 以下显示了在 Go SDK 中调度开始时间为"现在"的工作流的示例:

client.ScheduleNewWorkflow(ctx, "MyCoolWorkflow", workflow.WithStartTime(time.Now()))

使用 Dapr Shared 与工作流时的工作流集群部署

当使用 Dapr Shared时,可能会有多个 daprd 边车在单个负载均衡器或服务后面运行。 因此,接收工作的辅助实例可能不是接收工作结果的实例。 Dapr 创建第三种 actor 类型来处理此场景:dapr.internal.{namespace}.{appID}.executor,用于将辅助结果路由回正确的工作流 actor,以确保正确操作。

后续步骤

编写工作流 >>

相关链接

6 - 方法指南:编写工作流

了解如何开发和编写工作流

本文提供有关如何编写由 Dapr 工作流引擎执行的工作流的高级概述。

以代码形式编写工作流

Dapr 工作流逻辑使用通用编程语言实现,使你能够:

  • 使用你偏好的编程语言(无需学习新的 DSL 或 YAML 模式)。
  • 访问语言的标准库。
  • 构建你自己的库和抽象。
  • 使用调试器并检查局部变量。
  • 像应用程序的其他部分一样编写工作流的单元测试。

Dapr 边车不会加载任何工作流定义。相反,边车只是驱动工作流的执行,所有工作流活动都作为应用程序的一部分。

编写工作流活动

工作流活动 是工作流中的基本工作单元,是在业务流程中被编排的任务。

定义你希望工作流执行的工作流活动。活动是一个函数定义,可以接受输入和输出。以下示例创建了一个名为 hello_act 的计数器(活动),用于通知用户当前的计数器值。hello_act 是一个从名为 WorkflowActivityContext 的类派生的函数。

@wfr.activity(name='hello_act')
def hello_act(ctx: WorkflowActivityContext, wf_input):
    global counter
    counter += wf_input
    print(f'New counter value is: {counter}!', flush=True)

查看任务链工作流活动的上下文。

定义你希望工作流执行的工作流活动。活动被包装在 WorkflowActivityContext 类中,该类实现了工作流活动。

export default class WorkflowActivityContext {
  private readonly _innerContext: ActivityContext;
  constructor(innerContext: ActivityContext) {
    if (!innerContext) {
      throw new Error("ActivityContext cannot be undefined");
    }
    this._innerContext = innerContext;
  }

  public getWorkflowInstanceId(): string {
    return this._innerContext.orchestrationId;
  }

  public getWorkflowActivityId(): number {
    return this._innerContext.taskId;
  }
}

查看工作流活动的上下文。

定义你希望工作流执行的工作流活动。活动是一个类定义,可以接受输入和输出。活动还参与依赖注入,例如绑定到 Dapr 客户端。

以下示例中调用的活动包括:

  • NotifyActivity:接收新订单通知。
  • ReserveInventoryActivity:检查是否有足够的库存来满足新订单。
  • ProcessPaymentActivity:处理订单付款。包括 NotifyActivity 用于发送成功订单通知。

NotifyActivity

public class NotifyActivity : WorkflowActivity<Notification, object>
{
    //...

    public NotifyActivity(ILoggerFactory loggerFactory)
    {
        this.logger = loggerFactory.CreateLogger<NotifyActivity>();
    }

    //...
}

查看完整的 NotifyActivity.cs 工作流活动示例。

ReserveInventoryActivity

public class ReserveInventoryActivity : WorkflowActivity<InventoryRequest, InventoryResult>
{
    //...

    public ReserveInventoryActivity(ILoggerFactory loggerFactory, DaprClient client)
    {
        this.logger = loggerFactory.CreateLogger<ReserveInventoryActivity>();
        this.client = client;
    }

    //...

}

查看完整的 ReserveInventoryActivity.cs 工作流活动示例。

ProcessPaymentActivity

public class ProcessPaymentActivity : WorkflowActivity<PaymentRequest, object>
{
    //...
    public ProcessPaymentActivity(ILoggerFactory loggerFactory)
    {
        this.logger = loggerFactory.CreateLogger<ProcessPaymentActivity>();
    }

    //...

}

查看完整的 ProcessPaymentActivity.cs 工作流活动示例。

定义你希望工作流执行的工作流活动。活动被包装在公共 DemoWorkflowActivity 类中,该类实现了工作流活动。

@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
public class DemoWorkflowActivity implements WorkflowActivity {

  @Override
  public DemoActivityOutput run(WorkflowActivityContext ctx) {
    Logger logger = LoggerFactory.getLogger(DemoWorkflowActivity.class);
    logger.info("Starting Activity: " + ctx.getName());

    var message = ctx.getInput(DemoActivityInput.class).getMessage();
    var newMessage = message + " World!, from Activity";
    logger.info("Message Received from input: " + message);
    logger.info("Sending message to output: " + newMessage);

    logger.info("Sleeping for 5 seconds to simulate long running operation...");

    try {
      TimeUnit.SECONDS.sleep(5);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }


    logger.info("Activity finished");

    var output = new DemoActivityOutput(message, newMessage);
    logger.info("Activity returned: " + output);

    return output;
  }
}

查看 Java SDK 工作流活动示例的上下文。

定义工作流活动

定义你希望工作流执行的每个工作流活动。活动输入可以使用 ctx.GetInput 从上下文中解组。活动应定义为接受 ctx workflow.ActivityContext 参数并返回接口和错误的函数。

func BusinessActivity(ctx workflow.ActivityContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}
	
	// Do something here
	return "result", nil
}

定义工作流

使用参数 ctx *workflow.WorkflowContext 定义工作流函数,并返回 any 和 error。从工作流内部调用你定义的活动。

func BusinessWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return nil, err
	}
	var output string
	if err := ctx.CallActivity(BusinessActivity, workflow.ActivityInput(input)).Await(&output); err != nil {
		return nil, err
	}
	if err := ctx.WaitForExternalEvent("businessEvent", time.Minute*60).Await(&output); err != nil {
		return nil, err
	}
	
	if err := ctx.CreateTimer(time.Second).Await(nil); err != nil {
		return nil, nil
	}
	return output, nil
}

注册工作流和活动

在你的应用程序可以执行工作流之前,你必须将工作流编排器和其活动注册到工作流注册表中。这确保 Dapr 知道在执行工作流时调用哪些函数。

func main() {
	// Create a workflow registry
	r := workflow.NewRegistry()

	// Register the workflow orchestrator
	if err := r.AddWorkflow(BusinessWorkflow); err != nil {
		log.Fatal(err)
	}
	fmt.Println("BusinessWorkflow registered")

	// Register the workflow activities
	if err := r.AddActivity(BusinessActivity); err != nil {
		log.Fatal(err)
	}
	fmt.Println("BusinessActivity registered")

	// Create workflow client and start worker
	wclient, err := client.NewWorkflowClient()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Worker initialized")

	ctx, cancel := context.WithCancel(context.Background())
	if err = wclient.StartWorker(ctx, r); err != nil {
		log.Fatal(err)
	}
	fmt.Println("runner started")

	// Your application logic continues here...
	// Example: Start a workflow
	instanceID, err := wclient.ScheduleWorkflow(ctx, "BusinessWorkflow", workflow.WithInput(1))
	if err != nil {
		log.Fatalf("failed to start workflow: %v", err)
	}
	fmt.Printf("workflow started with id: %v\n", instanceID)

	// Stop workflow worker when done
	cancel()
	fmt.Println("workflow worker successfully shutdown")
}

关于注册的关键要点:

  • 使用 workflow.NewRegistry() 创建工作流注册表
  • 使用 r.AddWorkflow() 注册工作流函数
  • 使用 r.AddActivity() 注册活动函数
  • 使用 client.NewWorkflowClient() 创建工作流客户端
  • 调用 wclient.StartWorker() 开始处理工作流
  • 使用 wclient.ScheduleWorkflow 调度命名的工作流实例

查看 Go SDK 工作流活动示例的上下文。

编写工作流

接下来,在工作流中注册和调用活动。

hello_world_wf 函数是一个从名为 DaprWorkflowContext 的类派生的函数,具有输入和输出参数类型。它还包括一个 yield 语句,该语句完成工作流的核心逻辑并调用工作流活动。

@wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
    print(f'{wf_input}')
    yield ctx.call_activity(hello_act, input=1)
    yield ctx.call_activity(hello_act, input=10)
    yield ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
    yield ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)

    # Change in event handling: Use when_any to handle both event and timeout
    event = ctx.wait_for_external_event(event_name)
    timeout = ctx.create_timer(timedelta(seconds=30))
    winner = yield when_any([event, timeout])

    if winner == timeout:
        print('Workflow timed out waiting for event')
        return 'Timeout'

    yield ctx.call_activity(hello_act, input=100)
    yield ctx.call_activity(hello_act, input=1000)
    return 'Completed'

查看 hello_world_wf 工作流的上下文。

接下来,向 WorkflowRuntime 类注册工作流并启动工作流运行时。

export default class WorkflowRuntime {

  //..
  // Register workflow implementation for handling orchestrations
  public registerWorkflow(workflow: TWorkflow): WorkflowRuntime {
    const name = getFunctionName(workflow);
    const workflowWrapper = (ctx: OrchestrationContext, input: any): any => {
      const workflowContext = new WorkflowContext(ctx);
      return workflow(workflowContext, input);
    };
    this.worker.addNamedOrchestrator(name, workflowWrapper);
    return this;
  }

  // Register workflow activities
  public registerActivity(fn: TWorkflowActivity<TInput, TOutput>): WorkflowRuntime {
    const name = getFunctionName(fn);
    const activityWrapper = (ctx: ActivityContext, input: TInput): TOutput => {
      const wfActivityContext = new WorkflowActivityContext(ctx);
      return fn(wfActivityContext, input);
    };
    this.worker.addNamedActivity(name, activityWrapper);
    return this;
  }

  // Start the workflow runtime processing items and block.
  public async start() {
    await this.worker.start();
  }

}

查看 WorkflowRuntime 的上下文。

OrderProcessingWorkflow 类是从名为 Workflow 的基类派生的,具有输入和输出参数类型。它还包括一个 RunAsync 方法,该方法完成工作流的核心逻辑并调用工作流活动。

 class OrderProcessingWorkflow : Workflow<OrderPayload, OrderResult>
    {
        public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderPayload order)
        {
            //...

            await context.CallActivityAsync(
                nameof(NotifyActivity),
                new Notification($"Received order {orderId} for {order.Name} at {order.TotalCost:c}"));

            //...

            InventoryResult result = await context.CallActivityAsync<InventoryResult>(
                nameof(ReserveInventoryActivity),
                new InventoryRequest(RequestId: orderId, order.Name, order.Quantity));
            //...
            
            await context.CallActivityAsync(
                nameof(ProcessPaymentActivity),
                new PaymentRequest(RequestId: orderId, order.TotalCost, "USD"));

            await context.CallActivityAsync(
                nameof(NotifyActivity),
                new Notification($"Order {orderId} processed successfully!"));

            // End the workflow with a success result
            return new OrderResult(Processed: true);
        }
    }

查看 OrderProcessingWorkflow.cs 中的完整工作流示例。

接下来,向 WorkflowRuntimeBuilder 注册工作流并启动工作流运行时。

public class DemoWorkflowWorker {

  public static void main(String[] args) throws Exception {

    // Register the Workflow with the builder.
    WorkflowRuntimeBuilder builder = new WorkflowRuntimeBuilder().registerWorkflow(DemoWorkflow.class);
    builder.registerActivity(DemoWorkflowActivity.class);

    // Build and then start the workflow runtime pulling and executing tasks
    try (WorkflowRuntime runtime = builder.build()) {
      System.out.println("Start workflow runtime");
      runtime.start();
    }

    System.exit(0);
  }
}

查看 Java SDK 工作流的上下文。

使用参数 ctx *workflow.WorkflowContext 定义工作流函数,并返回 any 和 error。从工作流内部调用你定义的活动。

func BusinessWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input int
	if err := ctx.GetInput(&input); err != nil {
		return nil, err
	}
	var output string
	if err := ctx.CallActivity(BusinessActivity, workflow.ActivityInput(input)).Await(&output); err != nil {
		return nil, err
	}
	if err := ctx.WaitForExternalEvent("businessEvent", time.Minute*60).Await(&output); err != nil {
		return nil, err
	}
	
	if err := ctx.CreateTimer(time.Second).Await(nil); err != nil {
		return nil, nil
	}
	return output, nil
}

查看 Go SDK 工作流的上下文。

编写应用程序

最后,使用工作流编写应用程序。

在以下示例中,对于使用 Python SDK 的基本 Python hello world 应用程序,你的项目代码应包括:

  • 一个名为 DaprClient 的 Python 包,用于接收 Python SDK 功能。
  • 一个带有扩展的构建器:
  • API 调用。在以下示例中,这些调用启动、暂停、恢复、清除和完成工作流。
from datetime import timedelta
from time import sleep
from dapr.ext.workflow import (
    WorkflowRuntime,
    DaprWorkflowContext,
    WorkflowActivityContext,
    RetryPolicy,
    DaprWorkflowClient,
    when_any,
)
from dapr.conf import Settings
from dapr.clients.exceptions import DaprInternalError

settings = Settings()

counter = 0
retry_count = 0
child_orchestrator_count = 0
child_orchestrator_string = ''
child_act_retry_count = 0
instance_id = 'exampleInstanceID'
child_instance_id = 'childInstanceID'
workflow_name = 'hello_world_wf'
child_workflow_name = 'child_wf'
input_data = 'Hi Counter!'
event_name = 'event1'
event_data = 'eventData'
non_existent_id_error = 'no such instance exists'

retry_policy = RetryPolicy(
    first_retry_interval=timedelta(seconds=1),
    max_number_of_attempts=3,
    backoff_coefficient=2,
    max_retry_interval=timedelta(seconds=10),
    retry_timeout=timedelta(seconds=100),
)

wfr = WorkflowRuntime()


@wfr.workflow(name='hello_world_wf')
def hello_world_wf(ctx: DaprWorkflowContext, wf_input):
    print(f'{wf_input}')
    yield ctx.call_activity(hello_act, input=1)
    yield ctx.call_activity(hello_act, input=10)
    yield ctx.call_activity(hello_retryable_act, retry_policy=retry_policy)
    yield ctx.call_child_workflow(child_retryable_wf, retry_policy=retry_policy)

    # Change in event handling: Use when_any to handle both event and timeout
    event = ctx.wait_for_external_event(event_name)
    timeout = ctx.create_timer(timedelta(seconds=30))
    winner = yield when_any([event, timeout])

    if winner == timeout:
        print('Workflow timed out waiting for event')
        return 'Timeout'

    yield ctx.call_activity(hello_act, input=100)
    yield ctx.call_activity(hello_act, input=1000)
    return 'Completed'


@wfr.activity(name='hello_act')
def hello_act(ctx: WorkflowActivityContext, wf_input):
    global counter
    counter += wf_input
    print(f'New counter value is: {counter}!', flush=True)


@wfr.activity(name='hello_retryable_act')
def hello_retryable_act(ctx: WorkflowActivityContext):
    global retry_count
    if (retry_count % 2) == 0:
        print(f'Retry count value is: {retry_count}!', flush=True)
        retry_count += 1
        raise ValueError('Retryable Error')
    print(f'Retry count value is: {retry_count}! This print statement verifies retry', flush=True)
    retry_count += 1


@wfr.workflow(name='child_retryable_wf')
def child_retryable_wf(ctx: DaprWorkflowContext):
    global child_orchestrator_string, child_orchestrator_count
    if not ctx.is_replaying:
        child_orchestrator_count += 1
        print(f'Appending {child_orchestrator_count} to child_orchestrator_string!', flush=True)
        child_orchestrator_string += str(child_orchestrator_count)
    yield ctx.call_activity(
        act_for_child_wf, input=child_orchestrator_count, retry_policy=retry_policy
    )
    if child_orchestrator_count < 3:
        raise ValueError('Retryable Error')


@wfr.activity(name='act_for_child_wf')
def act_for_child_wf(ctx: WorkflowActivityContext, inp):
    global child_orchestrator_string, child_act_retry_count
    inp_char = chr(96 + inp)
    print(f'Appending {inp_char} to child_orchestrator_string!', flush=True)
    child_orchestrator_string += inp_char
    if child_act_retry_count % 2 == 0:
        child_act_retry_count += 1
        raise ValueError('Retryable Error')
    child_act_retry_count += 1


def main():
    wfr.start()
    wf_client = DaprWorkflowClient()

    print('==========Start Counter Increase as per Input:==========')
    wf_client.schedule_new_workflow(
        workflow=hello_world_wf, input=input_data, instance_id=instance_id
    )

    wf_client.wait_for_workflow_start(instance_id)

    # Sleep to let the workflow run initial activities
    sleep(12)

    assert counter == 11
    assert retry_count == 2
    assert child_orchestrator_string == '1aa2bb3cc'

    # Pause Test
    wf_client.pause_workflow(instance_id=instance_id)
    metadata = wf_client.get_workflow_state(instance_id=instance_id)
    print(f'Get response from {workflow_name} after pause call: {metadata.runtime_status.name}')

    # Resume Test
    wf_client.resume_workflow(instance_id=instance_id)
    metadata = wf_client.get_workflow_state(instance_id=instance_id)
    print(f'Get response from {workflow_name} after resume call: {metadata.runtime_status.name}')

    sleep(2)  # Give the workflow time to reach the event wait state
    wf_client.raise_workflow_event(instance_id=instance_id, event_name=event_name, data=event_data)

    print('========= Waiting for Workflow completion', flush=True)
    try:
        state = wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)
        if state.runtime_status.name == 'COMPLETED':
            print('Workflow completed! Result: {}'.format(state.serialized_output.strip('"')))
        else:
            print(f'Workflow failed! Status: {state.runtime_status.name}')
    except TimeoutError:
        print('*** Workflow timed out!')

    wf_client.purge_workflow(instance_id=instance_id)
    try:
        wf_client.get_workflow_state(instance_id=instance_id)
    except DaprInternalError as err:
        if non_existent_id_error in err._message:
            print('Instance Successfully Purged')

    sleep(10000)
    wfr.shutdown()


if __name__ == '__main__':
    main()

以下示例 是使用 JavaScript SDK 的基本 JavaScript 应用程序。与此示例一样,你的项目代码应包括:

  • 一个带有扩展的构建器:
  • API 调用。以下示例是一个使用工作流 API 的简单项目:
mkdir my-wf && cd my-wf
npm init -y
npm i @dapr/dapr @microsoft/durabletask-js
npm i -D typescript ts-node @types/node

创建以下 tsconfig.json 文件:

{
  "compilerOptions": {
    "target": "ES2020",
    "module": "CommonJS",
    "moduleResolution": "Node",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "outDir": "dist"
  },
  "include": ["src"]
}

创建以下 src/app.ts 文件:

import {
  WorkflowRuntime,
  WorkflowActivityContext,
  WorkflowContext,
  DaprWorkflowClient,
  TWorkflow
} from "@dapr/dapr";

const workflowClient = new DaprWorkflowClient();
const workflowRuntime = new WorkflowRuntime();

// simple activity
const hello = async (_: WorkflowActivityContext, name: string) => `Hello ${name}!`;

// simple workflow: call the activity 3 times
const sequence: TWorkflow = async function* (ctx: WorkflowContext): any {
  const out: string[] = [];
  out.push(yield ctx.callActivity(hello, "Tokyo"));
  out.push(yield ctx.callActivity(hello, "Seattle"));
  out.push(yield ctx.callActivity(hello, "London"));
  out.push(yield ctx.waitForExternalEvent("continue"));
  return out;
};

async function main() {
  workflowRuntime.registerWorkflow(sequence).registerActivity(hello);
  await workflowRuntime.start();

  const id = await workflowClient.scheduleNewWorkflow(sequence);
  console.log("Scheduled:", id);

  workflowClient.raiseEvent(id, "continue", "Go go go!");

  const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);
  console.log("Done:", state?.runtimeStatus, "output:", state?.serializedOutput);

  await new Promise(f => setTimeout(f, 100000));

  await workflowRuntime.stop();
  await workflowClient.stop();

}

main().catch((e) => { console.error(e); });

在以下 Program.cs 示例中,对于使用 .NET SDK 的基本 ASP.NET 订单处理应用程序,你的项目代码应包括:

  • 一个名为 Dapr.Workflow 的 NuGet 包,用于接收 .NET SDK 功能
  • 一个带有扩展方法 AddDaprWorkflow 的构建器
    • 这允许你注册工作流和工作流活动(工作流可以调度的任务)
  • HTTP API 调用
    • 一个用于提交新订单
    • 一个用于检查现有订单的状态
using Dapr.Workflow;
//...

// Dapr Workflows are registered as part of the service configuration
builder.Services.AddDaprWorkflow(options =>
{
    // Note that it's also possible to register a lambda function as the workflow
    // or activity implementation instead of a class.
    options.RegisterWorkflow<OrderProcessingWorkflow>();

    // These are the activities that get invoked by the workflow(s).
    options.RegisterActivity<NotifyActivity>();
    options.RegisterActivity<ReserveInventoryActivity>();
    options.RegisterActivity<ProcessPaymentActivity>();
});

WebApplication app = builder.Build();

// POST starts new order workflow instance
app.MapPost("/orders", async (DaprWorkflowClient client, [FromBody] OrderPayload orderInfo) =>
{
    if (orderInfo?.Name == null)
    {
        return Results.BadRequest(new
        {
            message = "Order data was missing from the request",
            example = new OrderPayload("Paperclips", 99.95),
        });
    }

//...
});

// GET fetches state for order workflow to report status
app.MapGet("/orders/{orderId}", async (string orderId, DaprWorkflowClient client) =>
{
    WorkflowState state = await client.GetWorkflowStateAsync(orderId, true);
    if (!state.Exists)
    {
        return Results.NotFound($"No order with ID = '{orderId}' was found.");
    }

    var httpResponsePayload = new
    {
        details = state.ReadInputAs<OrderPayload>(),
        status = state.RuntimeStatus.ToString(),
        result = state.ReadOutputAs<OrderResult>(),
    };

//...
}).WithName("GetOrderInfoEndpoint");

app.Run();

如以下示例所示,使用 Java SDK 和 Dapr Workflow 的 hello-world 应用程序应包括:

  • 一个名为 io.dapr.workflows.client 的 Java 包,用于接收 Java SDK 客户端功能。
  • 导入 io.dapr.workflows.Workflow
  • DemoWorkflow 类,它扩展了 Workflow
  • 使用输入和输出创建工作流。
  • API 调用。在以下示例中,这些调用启动并调用工作流活动。
package io.dapr.examples.workflows;

import com.microsoft.durabletask.CompositeTaskFailedException;
import com.microsoft.durabletask.Task;
import com.microsoft.durabletask.TaskCanceledException;
import io.dapr.workflows.Workflow;
import io.dapr.workflows.WorkflowStub;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

/**
 * Implementation of the DemoWorkflow for the server side.
 */
public class DemoWorkflow extends Workflow {
  @Override
  public WorkflowStub create() {
    return ctx -> {
      ctx.getLogger().info("Starting Workflow: " + ctx.getName());
      // ...
      ctx.getLogger().info("Calling Activity...");
      var input = new DemoActivityInput("Hello Activity!");
      var output = ctx.callActivity(DemoWorkflowActivity.class.getName(), input, DemoActivityOutput.class).await();
      // ...
    };
  }
}

查看完整的 Java SDK 工作流示例的上下文。

如以下示例所示,使用 Go SDK 和 Dapr Workflow 的 hello-world 应用程序应包括:

  • 一个名为 client 的 Go 包,用于接收 Go SDK 客户端功能。
  • BusinessWorkflow 方法
  • 使用输入和输出创建工作流。
  • API 调用。在以下示例中,这些调用启动并调用工作流活动。
package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"strconv"
	"time"

	"github.com/dapr/durabletask-go/workflow"
	"github.com/dapr/go-sdk/client"
)

var stage = 0
var failActivityTries = 0

func main() {
	r := workflow.NewRegistry()

	if err := r.AddWorkflow(BusinessWorkflow); err != nil {
		log.Fatal(err)
	}
	fmt.Println("BusinessWorkflow registered")

	if err := r.AddActivity(BusinessActivity); err != nil {
		log.Fatal(err)
	}
	fmt.Println("BusinessActivity registered")

	if err := r.AddActivity(FailActivity); err != nil {
		log.Fatal(err)
	}
	fmt.Println("FailActivity registered")

	wclient, err := client.NewWorkflowClient()
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("Worker initialized")

	ctx, cancel := context.WithCancel(context.Background())
	if err = wclient.StartWorker(ctx, r); err != nil {
		log.Fatal(err)
	}
	fmt.Println("runner started")

	// Start workflow test
	// Set the start time to the current time to not wait for the workflow to
	// "start". This is useful for increasing the throughput of creating
	// workflows.
	// workflow.WithStartTime(time.Now())
	instanceID, err := wclient.ScheduleWorkflow(ctx, "BusinessWorkflow", workflow.WithInstanceID("a7a4168d-3a1c-41da-8a4f-e7f6d9c718d9"), workflow.WithInput("1"))
	if err != nil {
		log.Fatalf("failed to start workflow: %v", err)
	}
	fmt.Printf("workflow started with id: %v\n", instanceID)

	// Pause workflow test
	err = wclient.SuspendWorkflow(ctx, instanceID, "")
	if err != nil {
		log.Fatalf("failed to pause workflow: %v", err)
	}

	respFetch, err := wclient.FetchWorkflowMetadata(ctx, instanceID, workflow.WithFetchPayloads(true))
	if err != nil {
		log.Fatalf("failed to fetch workflow: %v", err)
	}

	if respFetch.RuntimeStatus != workflow.StatusSuspended {
		log.Fatalf("workflow not paused: %s: %v", respFetch.RuntimeStatus, respFetch)
	}

	fmt.Printf("workflow paused\n")

	// Resume workflow test
	err = wclient.ResumeWorkflow(ctx, instanceID, "")
	if err != nil {
		log.Fatalf("failed to resume workflow: %v", err)
	}

	respFetch, err = wclient.FetchWorkflowMetadata(ctx, instanceID, workflow.WithFetchPayloads(true))
	if err != nil {
		log.Fatalf("failed to get workflow: %v", err)
	}

	if respFetch.RuntimeStatus != workflow.StatusRunning {
		log.Fatalf("workflow not running")
	}

	fmt.Println("workflow resumed")

	fmt.Printf("stage: %d\n", stage)

	// Raise Event
	err = wclient.RaiseEvent(ctx, instanceID, "businessEvent", workflow.WithEventPayload("testData"))
	if err != nil {
		fmt.Printf("failed to raise event: %v", err)
	}

	fmt.Println("workflow event raised")

	time.Sleep(time.Second) // allow workflow to advance

	fmt.Printf("stage: %d\n", stage)

	_, err = wclient.WaitForWorkflowCompletion(ctx, instanceID)
	if err != nil {
		log.Fatalf("failed to wait for workflow: %v", err)
	}

	fmt.Printf("fail activity executions: %d\n", failActivityTries)

	respFetch, err = wclient.FetchWorkflowMetadata(ctx, instanceID, workflow.WithFetchPayloads(true))
	if err != nil {
		log.Fatalf("failed to get workflow: %v", err)
	}

	fmt.Printf("workflow status: %v\n", respFetch.String())

	// Purge workflow test
	err = wclient.PurgeWorkflowState(ctx, instanceID)
	if err != nil {
		log.Fatalf("failed to purge workflow: %v", err)
	}

	respFetch, err = wclient.FetchWorkflowMetadata(ctx, instanceID, workflow.WithFetchPayloads(true))
	if err == nil || respFetch != nil {
		log.Fatalf("failed to purge workflow: %v", err)
	}

	fmt.Println("workflow purged")

	fmt.Printf("stage: %d\n", stage)

	// Terminate workflow test
	id, err := wclient.ScheduleWorkflow(ctx, "BusinessWorkflow", workflow.WithInstanceID("a7a4168d-3a1c-41da-8a4f-e7f6d9c718d9"), workflow.WithInput("1"))
	if err != nil {
		log.Fatalf("failed to start workflow: %v", err)
	}
	fmt.Printf("workflow started with id: %v\n", instanceID)

	metadata, err := wclient.WaitForWorkflowStart(ctx, id)
	if err != nil {
		log.Fatalf("failed to get workflow: %v", err)
	}
	fmt.Printf("workflow status: %s\n", metadata.String())

	err = wclient.TerminateWorkflow(ctx, id)
	if err != nil {
		log.Fatalf("failed to terminate workflow: %v", err)
	}
	fmt.Println("workflow terminated")

	err = wclient.PurgeWorkflowState(ctx, id)
	if err != nil {
		log.Fatalf("failed to purge workflow: %v", err)
	}
	fmt.Println("workflow purged")

	<-ctx.Done()
	cancel()

	fmt.Println("workflow worker successfully shutdown")
}

func BusinessWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var input string
	if err := ctx.GetInput(&input); err != nil {
		return nil, err
	}
	var output string
	if err := ctx.CallActivity(BusinessActivity, workflow.WithActivityInput(input)).Await(&output); err != nil {
		return nil, err
	}

	err := ctx.WaitForExternalEvent("businessEvent", time.Minute*60).Await(&output)
	if err != nil {
		return nil, err
	}

	if err := ctx.CallActivity(BusinessActivity, workflow.WithActivityInput(input)).Await(&output); err != nil {
		return nil, err
	}

	if err := ctx.CallActivity(FailActivity, workflow.WithActivityRetryPolicy(&workflow.RetryPolicy{
		MaxAttempts:          3,
		InitialRetryInterval: 100 * time.Millisecond,
		BackoffCoefficient:   2,
		MaxRetryInterval:     1 * time.Second,
	})).Await(nil); err == nil {
		return nil, fmt.Errorf("unexpected no error executing fail activity")
	}

	return output, nil
}

func BusinessActivity(ctx workflow.ActivityContext) (any, error) {
	var input string
	if err := ctx.GetInput(&input); err != nil {
		return "", err
	}

	iinput, err := strconv.Atoi(input)
	if err != nil {
		return "", err
	}

	stage += iinput

	return fmt.Sprintf("Stage: %d", stage), nil
}

func FailActivity(ctx workflow.ActivityContext) (any, error) {
	failActivityTries += 1
	return nil, errors.New("dummy activity error")
}

查看完整的 Go SDK 工作流示例的上下文。

运行工作流并使用 Diagrid Dashboard 检查工作流执行

通过你的 IDE 或 Dapr CLI 启动工作流应用程序(如果你想启动多个应用程序,请使用 Dapr 多应用运行;如果只启动一个应用程序,请使用常规 Dapr run 命令),然后调度一个新的工作流实例。

使用本地 Diagrid Dashboard 可视化和检查你的工作流状态,并深入查看详细的工作流执行历史。仪表板作为容器运行,连接到 Dapr 工作流使用的状态存储(默认情况下是本地 Redis 实例)。

Diagrid Dashboard showing local workflow executions

使用 Docker 启动 Diagrid Dashboard 容器:

docker run -p 8080:8080 ghcr.io/diagridio/diagrid-dashboard:latest

在浏览器中打开仪表板,地址为 http://localhost:8080

通过 Dapr CLI 测试工作流

编写工作流后,你可以使用 Dapr CLI 进行测试:

运行工作流应用程序

dapr run --app-id workflow-app python3 app.py

确保应用程序正在运行:

dapr list

运行工作流

dapr workflow run hello_world_wf --app-id workflow-app --input 'hello world' --instance-id test-run

检查工作流状态

dapr workflow list --app-id workflow-app -o wide

查看已完成的工作流

dapr workflow list --app-id workflow-app --filter-status COMPLETED -o wide

查看工作流历史

dapr workflow history --app-id workflow-app test-run

运行工作流应用程序

dapr run --app-id workflow-app npx ts-node src/app.ts

确保应用程序正在运行:

dapr list

运行工作流

dapr workflow run sequence --app-id workflow-app --input 'hello world' --instance-id test-run

检查工作流状态

dapr workflow list --app-id workflow-app -o wide

触发等待的外部事件

dapr workflow raise-event --app-id workflow-app test-run/businessEvent

查看已完成的工作流

dapr workflow list --app-id workflow-app --filter-status COMPLETED -o wide

查看工作流历史

dapr workflow history --app-id workflow-app test-run

运行工作流应用程序

dapr run --app-id workflow-app dotnet run

确保应用程序正在运行:

dapr list

运行工作流

dapr workflow run OrderProcessingWorkflow --app-id workflow-app  --instance-id test-run --input '{"name": "Paperclips", "totalCost": 99.95}'

检查工作流状态

dapr workflow list --app-id workflow-app -o wide

触发等待的外部事件

dapr workflow raise-event --app-id workflow-app test-run/incoming-purchase-order --input '{"name": "Paperclips", "totalCost": 99.95}'

查看已完成的工作流

dapr workflow list --app-id workflow-app --filter-status COMPLETED -o wide

查看工作流历史

dapr workflow history --app-id workflow-app test-run

运行工作流应用程序

dapr run --app-id workflow-app -- java -jar target/WorkflowService-0.0.1-SNAPSHOT.jar

确保应用程序正在运行:

dapr list

运行工作流

dapr workflow run DemoWorkflow --app-id workflow-app  --instance-id test-run --input "input data"

检查工作流状态

dapr workflow list --app-id workflow-app -o wide

触发等待的外部事件

dapr workflow raise-event --app-id workflow-app test-run/TestEvent --input 'TestEventPayload'
dapr workflow raise-event --app-id workflow-app test-run/event1 --input 'TestEvent 1 Payload'
dapr workflow raise-event --app-id workflow-app test-run/event2 --input 'TestEvent 2 Payload'
dapr workflow raise-event --app-id workflow-app test-run/event3 --input 'TestEvent 3 Payload'

查看已完成的工作流

dapr workflow list --app-id workflow-app --filter-status COMPLETED -o wide

查看工作流历史

dapr workflow history --app-id workflow-app test-run

运行工作流应用程序

dapr run --app-id workflow-app go run main.go

确保应用程序正在运行:

dapr list

运行工作流

dapr workflow run BusinessWorkflow --app-id workflow-app --input '1' --instance-id test-run

检查工作流状态

dapr workflow list --app-id workflow-app -o wide

触发等待的外部事件

dapr workflow raise-event --app-id workflow-app test-run/businessEvent

查看已完成的工作流

dapr workflow list --app-id workflow-app --filter-status COMPLETED -o wide

查看工作流历史

dapr workflow history test-run --app-id workflow-app

监控工作流执行

dapr workflow list --app-id workflow-app --filter-status RUNNING -o wide
dapr workflow list --app-id workflow-app --filter-status FAILED -o wide
dapr workflow list --app-id workflow-app --filter-status COMPLETED -o wide

测试外部事件

# 触发你的工作流正在等待的事件
dapr workflow raise-event <instance-id>/ApprovalReceived \
  --app-id workflow-app \
  --input '{"approved": true, "approver": "manager@company.com"}'

调试失败的工作流

# 列出失败的工作流
dapr workflow list --app-id workflow-app --filter-status FAILED --output wide

# 获取失败工作流的详细历史
dapr workflow history <failed-instance-id> --app-id workflow-app --output json

# 修复问题后重新运行工作流
dapr workflow rerun <failed-instance-id> --app-id workflow-app --input '<new-input-json-data>'

下一步

现在你已经编写了一个工作流,学习如何管理它。

管理工作流 >>

相关链接

7 - 操作指南:管理工作流

管理和运行工作流

现在您已经在应用程序中编写了工作流及其活动,可以使用 CLI 或 API 调用来启动、终止、重运行和获取工作流信息。

使用 Dapr CLI 管理工作流

Dapr CLI 提供了在自托管和 Kubernetes 环境中管理工作流实例的命令。

另请参阅工作流保留策略了解如何为已完成的工作流配置保留策略。

基本工作流操作

启动工作流

# 使用 `orderprocessing` 应用程序,启动一个新的工作流实例并传入输入数据
dapr workflow run OrderProcessingWorkflow \
  --app-id orderprocessing \
  --input '{"orderId": "12345", "amount": 100.50}'

# 使用特定实例 ID 启动新工作流
dapr workflow run OrderProcessingWorkflow \
  --app-id orderprocessing \
  --instance-id order-12345 \
  --input '{"orderId": "12345"}'

# 计划在 2024 年 12 月 25 日上午 10:00:00(协调世界时 UTC)启动新工作流
dapr workflow run OrderProcessingWorkflow \
  --app-id orderprocessing \
  --start-time "2024-12-25T10:00:00Z"

列出工作流实例

# 列出应用的所有工作流
dapr workflow list

# 按状态筛选
dapr workflow list --filter-status RUNNING

# 按工作流名称和应用 ID 筛选
dapr workflow list --app-id orderprocessing --filter-name OrderProcessingWorkflow

# 按时间筛选(过去 24 小时内启动的工作流)
dapr workflow list --filter-max-age 24h

# 获取详细输出
dapr workflow list -o wide

查看工作流历史

# 获取执行历史
dapr workflow history order-12345

# 在特定应用 ID 上以 JSON 格式获取历史
dapr workflow history order-12345 --app-id orderprocessing --output json

控制工作流执行

# 暂停正在运行的工作流
dapr workflow suspend order-12345 \
  --app-id orderprocessing \
  --reason "Waiting for manual approval"

# 恢复已暂停的工作流
dapr workflow resume order-12345 \
  --app-id orderprocessing \
  --reason "Approved by manager"

# 终止工作流
dapr workflow terminate order-12345 \
  --app-id orderprocessing \
  --output '{"reason": "Cancelled by customer"}'

触发外部事件

# 为等待中的工作流触发事件
dapr workflow raise-event order-12345/PaymentReceived \
  --app-id orderprocessing \
  --input '{"paymentId": "pay-67890", "amount": 100.50}'

重运行工作流

# 从头重运行
dapr workflow rerun order-12345

# 从特定事件 ID 重运行(通过 history 命令发现)
dapr workflow rerun order-12345 --event-id 5

# 使用新的指定实例 ID 重运行
dapr workflow rerun order-12345 --new-instance-id order-12345-retry

清理已完成的工作流

请注意,从 CLI 清理工作流也会删除所有关联的 Scheduler 提醒。

# 清理特定实例
dapr workflow purge order-12345

# 清理 30 天前已完成的所有工作流
dapr workflow purge --all-older-than 720h

# 清理所有终结状态的工作流(谨慎使用!)
dapr workflow purge --app-id orderprocessing --all

# 在没有运行工作流客户端的情况下强制清理(极度谨慎使用!)
dapr workflow purge order-12345 --force

Kubernetes 操作

所有命令都支持 Kubernetes 部署的 -k 标志:

# 在 Kubernetes 中列出工作流
dapr workflow list \
  --kubernetes \
  --namespace production \
  --app-id orderprocessing

# 在 Kubernetes 中暂停工作流
dapr workflow suspend order-12345 \
  --kubernetes \
  --namespace production \
  --app-id orderprocessing \
  --reason "Maintenance window"

列出工作流

在自托管模式下,只需运行:

dapr workflow list

在 Kubernetes 模式下,需要指定 --kubernetes/-k 标志以及命名空间和应用 ID:

dapr workflow list -k

工作流管理最佳实践

  1. 监控正在运行的工作流:使用筛选列表跟踪长期运行的实例

    dapr workflow list --app-id orderprocessing --filter-status RUNNING --filter-max-age 24h
    
  2. 使用实例 ID:分配有意义的实例 ID 以便更轻松地跟踪

    dapr workflow run OrderWorkflow --app-id orderprocessing --instance-id "order-$(date +%s)"
    
  3. 导出以供分析:导出工作流数据以供分析

    dapr workflow list --app-id orderprocessing --output json > workflows.json
    

使用 Dapr CLI 管理工作流提醒

工作流提醒存储在 Scheduler 中,可以使用 dapr scheduler CLI 进行管理。

列出工作流提醒

dapr scheduler list --filter workflow
NAME                                           BEGIN     COUNT  LAST TRIGGER
workflow/my-app/instance1/timer-0-ABC123       +50.0h    0
workflow/my-app/instance2/timer-0-XYZ789       +50.0h    0

获取提醒详情

dapr scheduler get workflow/my-app/instance1/timer-0-ABC123 -o yaml

删除工作流提醒

删除单个提醒:

dapr scheduler delete workflow/my-app/instance1/timer-0-ABC123

删除给定工作流应用的所有提醒:

dapr scheduler delete-all workflow/my-app

删除特定工作流实例的所有提醒:

dapr scheduler delete-all workflow/my-app/instance1

备份和恢复提醒

导出所有提醒:

dapr scheduler export -o workflow-reminders-backup.bin

从备份文件恢复:

dapr scheduler import -f workflow-reminders-backup.bin

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

  • schedule_new_workflow:启动工作流实例
  • get_workflow_state:获取工作流状态信息
  • pause_workflow:暂停或挂起工作流实例,以便后续恢复
  • resume_workflow:恢复已暂停的工作流实例
  • raise_workflow_event:向工作流触发事件
  • purge_workflow:移除与特定工作流实例相关的所有元数据
  • wait_for_workflow_completion:等待特定工作流实例完成
from dapr.ext.workflow import WorkflowRuntime, DaprWorkflowContext, WorkflowActivityContext
from dapr.clients import DaprClient

# 合理的参数
instanceId = "exampleInstanceID"
workflowComponent = "dapr"
workflowName = "hello_world_wf"
eventName = "event1"
eventData = "eventData"

# 启动工作流
wf_client.schedule_new_workflow(
        workflow=hello_world_wf, input=input_data, instance_id=instance_id
    )

# 获取工作流信息
wf_client.get_workflow_state(instance_id=instance_id)

# 暂停工作流
wf_client.pause_workflow(instance_id=instance_id)
    metadata = wf_client.get_workflow_state(instance_id=instance_id)

# 恢复工作流
wf_client.resume_workflow(instance_id=instance_id)

# 向工作流触发事件
wf_client.raise_workflow_event(instance_id=instance_id, event_name=event_name, data=event_data)

# 清理工作流
wf_client.purge_workflow(instance_id=instance_id)

# 等待工作流完成
wf_client.wait_for_workflow_completion(instance_id, timeout_in_seconds=30)

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

  • client.workflow.start:启动工作流实例
  • client.workflow.get:获取工作流状态信息
  • client.workflow.pause:暂停或挂起工作流实例,以便后续恢复
  • client.workflow.resume:恢复已暂停的工作流实例
  • client.workflow.purge:移除与特定工作流实例相关的所有元数据
  • client.workflow.terminate:终止或停止工作流的特定实例
import { DaprClient } from "@dapr/dapr";

async function printWorkflowStatus(client: DaprClient, instanceId: string) {
  const workflow = await client.workflow.get(instanceId);
  console.log(
    `Workflow ${workflow.workflowName}, created at ${workflow.createdAt.toUTCString()}, has status ${
      workflow.runtimeStatus
    }`,
  );
  console.log(`Additional properties: ${JSON.stringify(workflow.properties)}`);
  console.log("--------------------------------------------------\n\n");
}

async function start() {
  const client = new DaprClient();

  // 启动新的工作流实例
  const instanceId = await client.workflow.start("OrderProcessingWorkflow", {
    Name: "Paperclips",
    TotalCost: 99.95,
    Quantity: 4,
  });
  console.log(`Started workflow instance ${instanceId}`);
  await printWorkflowStatus(client, instanceId);

  // 暂停工作流实例
  await client.workflow.pause(instanceId);
  console.log(`Paused workflow instance ${instanceId}`);
  await printWorkflowStatus(client, instanceId);

  // 恢复工作流实例
  await client.workflow.resume(instanceId);
  console.log(`Resumed workflow instance ${instanceId}`);
  await printWorkflowStatus(client, instanceId);

  // 终止工作流实例
  await client.workflow.terminate(instanceId);
  console.log(`Terminated workflow instance ${instanceId}`);
  await printWorkflowStatus(client, instanceId);

  // 等待工作流完成,30 秒!
  await new Promise((resolve) => setTimeout(resolve, 30000));
  await printWorkflowStatus(client, instanceId);

  // 清理工作流实例
  await client.workflow.purge(instanceId);
  console.log(`Purged workflow instance ${instanceId}`);
  // 这会抛出错误,因为工作流实例已不存在
  await printWorkflowStatus(client, instanceId);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

在代码中管理工作流。在编写工作流指南的 OrderProcessingWorkflow 示例中,工作流已注册到代码中。现在您可以启动、终止和获取运行中工作流的信息:

string orderId = "exampleOrderId";
OrderPayload input = new OrderPayload("Paperclips", 99.95);
Dictionary<string, string> workflowOptions; // 这是一个可选参数

// 使用 orderId 作为工作流 ID 启动工作流。这返回一个字符串,包含特定工作流实例的实例 ID,无论我们是自行提供还是由系统生成
await daprWorkflowClient.ScheduleNewWorkflowAsync(nameof(OrderProcessingWorkflow), orderId, input, workflowOptions);

// 获取工作流信息。此响应包含工作流状态、启动时间等信息!
WorkflowState currentState = await daprWorkflowClient.GetWorkflowStateAsync(orderId, orderId);

// 终止工作流
await daprWorkflowClient.TerminateWorkflowAsync(orderId);

// 触发事件(传入的采购订单),工作流将等待该事件
await daprWorkflowClient.RaiseEventAsync(orderId, "incoming-purchase-order", input);

// 暂停
await daprWorkflowClient.SuspendWorkflowAsync(orderId);

// 恢复
await daprWorkflowClient.ResumeWorkflowAsync(orderId);

// 清理工作流,从关联实例中移除所有收件箱和历史信息
await daprWorkflowClient.PurgeInstanceAsync(orderId);

在代码中管理工作流。在 Java SDK 的工作流示例中,工作流使用以下 API 注册到代码中:

  • scheduleNewWorkflow:启动新的工作流实例
  • getInstanceState:获取工作流状态信息
  • waitForInstanceStart:暂停或挂起工作流实例,以便后续恢复
  • raiseEvent:为运行中的工作流实例触发事件/任务
  • waitForInstanceCompletion:等待工作流完成任务
  • purgeInstance:移除与特定工作流实例相关的所有元数据
  • terminateWorkflow:终止工作流
  • purgeInstance:移除与特定工作流相关的所有元数据
package io.dapr.examples.workflows;

import io.dapr.workflows.client.DaprWorkflowClient;
import io.dapr.workflows.client.WorkflowInstanceStatus;

// ...
public class DemoWorkflowClient {

  // ...
  public static void main(String[] args) throws InterruptedException {
    DaprWorkflowClient client = new DaprWorkflowClient();

    try (client) {
      // 启动工作流
      String instanceId = client.scheduleNewWorkflow(DemoWorkflow.class, "input data");
      
      // 获取工作流的状态信息
      WorkflowInstanceStatus workflowMetadata = client.getInstanceState(instanceId, true);

      // 等待或暂停工作流实例启动
      try {
        WorkflowInstanceStatus waitForInstanceStartResult =
            client.waitForInstanceStart(instanceId, Duration.ofSeconds(60), true);
      }

      // 为工作流触发事件;您可以并行触发多个事件
      client.raiseEvent(instanceId, "TestEvent", "TestEventPayload");
      client.raiseEvent(instanceId, "event1", "TestEvent 1 Payload");
      client.raiseEvent(instanceId, "event2", "TestEvent 2 Payload");
      client.raiseEvent(instanceId, "event3", "TestEvent 3 Payload");

      // 等待工作流完成任务
      try {
        WorkflowInstanceStatus waitForInstanceCompletionResult =
            client.waitForInstanceCompletion(instanceId, Duration.ofSeconds(60), true);
      } 

      // 清理工作流实例,移除与其关联的所有元数据
      boolean purgeResult = client.purgeInstance(instanceId);

      // 终止工作流实例
      client.terminateWorkflow(instanceToTerminateId, null);

    System.exit(0);
  }
}

在代码中管理工作流。在 Go SDK 的工作流示例中,工作流使用以下 API 注册到代码中:

  • StartWorkflow:启动新的工作流实例
  • GetWorkflow:获取工作流状态信息
  • PauseWorkflow:暂停或挂起工作流实例,以便后续恢复
  • RaiseEventWorkflow:为运行中的工作流实例触发事件/任务
  • ResumeWorkflow:等待工作流完成任务
  • PurgeWorkflow:移除与特定工作流实例相关的所有元数据
  • TerminateWorkflow:终止工作流
// 启动工作流
type StartWorkflowRequest struct {
	InstanceID        string // 可选的实例标识符
	WorkflowComponent string
	WorkflowName      string
	Options           map[string]string // 可选的元数据
	Input             any               // 可选的输入
	SendRawInput      bool              // 设置为 True 以禁用输入上的序列化
}

type StartWorkflowResponse struct {
	InstanceID string
}

// 获取工作流状态
type GetWorkflowRequest struct {
	InstanceID        string
	WorkflowComponent string
}

type GetWorkflowResponse struct {
	InstanceID    string
	WorkflowName  string
	CreatedAt     time.Time
	LastUpdatedAt time.Time
	RuntimeStatus string
	Properties    map[string]string
}

// 清理工作流
type PurgeWorkflowRequest struct {
	InstanceID        string
	WorkflowComponent string
}

// 终止工作流
type TerminateWorkflowRequest struct {
	InstanceID        string
	WorkflowComponent string
}

// 暂停工作流
type PauseWorkflowRequest struct {
	InstanceID        string
	WorkflowComponent string
}

// 恢复工作流
type ResumeWorkflowRequest struct {
	InstanceID        string
	WorkflowComponent string
}

// 为运行中的工作流触发事件
type RaiseEventWorkflowRequest struct {
	InstanceID        string
	WorkflowComponent string
	EventName         string
	EventData         any
	SendRawData       bool // 设置为 True 以禁用数据上的序列化
}

使用 HTTP 调用管理工作流。下面的示例将编写工作流示例中的属性与随机实例 ID 结合使用。

启动工作流

使用 ID 12345678 启动工作流,运行:

curl -X POST "http://localhost:3500/v1.0/workflows/dapr/OrderProcessingWorkflow/start?instanceID=12345678"

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

终止工作流

使用 ID 12345678 终止工作流,运行:

curl -X POST "http://localhost:3500/v1.0/workflows/dapr/12345678/terminate"

触发事件

对于支持订阅外部事件的工作流组件(如 Dapr Workflow engine),您可以使用以下"触发事件" API 将命名事件传递到特定的工作流实例。

curl -X POST "http://localhost:3500/v1.0/workflows/<workflowComponentName>/<instanceID>/raiseEvent/<eventName>"

eventName 可以是任意函数。

暂停或恢复工作流

为了计划停机时间、等待输入等,您可以暂停然后恢复工作流。使用 ID 12345678 暂停工作流直到触发恢复,运行:

curl -X POST "http://localhost:3500/v1.0/workflows/dapr/12345678/pause"

使用 ID 12345678 恢复工作流,运行:

curl -X POST "http://localhost:3500/v1.0/workflows/dapr/12345678/resume"

清理工作流

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

只有处于 COMPLETED、FAILED 或 TERMINATED 状态的工作流实例才能被清理。如果工作流处于任何其他状态,调用清理将返回错误。

curl -X POST "http://localhost:3500/v1.0/workflows/dapr/12345678/purge"

获取工作流信息

使用 ID 12345678 获取工作流信息(输出和输入),运行:

curl -X GET "http://localhost:3500/v1.0/workflows/dapr/12345678"

下一步

现在您已了解如何管理工作流,学习如何在多个应用程序中执行工作流

多应用程序工作流>>

相关链接

8 - 多应用工作流

跨多个应用程序执行工作流

单个工作流通常跨越多个应用程序、微服务或编程语言。 在这种情况下,活动或子工作流将在与托管父工作流不同的应用程序上执行。

这非常有用的一些场景包括:

  • 机器学习(ML)训练活动必须在启用 GPU 的机器上执行,而工作流的其余部分在仅支持 CPU 的编排机器上运行。
  • 活动需要访问仅对特定身份或区域可用的敏感数据或凭据。
  • 工作流的不同部分需要在不同的信任区域或网络中执行。
  • 由于数据驻留要求,工作流的不同部分需要在不同的地理区域中执行。
  • 涉及的业务流程跨越多个团队或部门,每个团队或部门拥有自己的应用程序。
  • 基于团队专业知识或现有代码库,工作流的实现跨越不同的编程语言。
  • 不同的团队边界或微服务所有权。
Diagram showing multi-application complex workflow

下图显示了一个复杂工作流的示例场景,该工作流跨多个用不同语言编写的应用程序进行编排。每个应用程序的主要步骤和活动包括:

App1: 主工作流服务 - 协调整个 ML 管道的顶级编排器

  • 启动流程
  • 调用 App2 上的数据处理活动
  • 调用 App3 上的 ML 训练子工作流
  • 调用 App4 上的模型部署
  • 结束完整工作流
  • 语言: Java

App2: 数据处理管道 - 仅限 GPU 活动

  • 数据摄取活动(GPU 加速)
  • 特征工程活动(GPU 加速)
  • 向主工作流返回完成信号
  • 语言: Go

App3: ML 训练子工作流 - 包含子工作流和活动

  • 子工作流编排:
    • 数据处理活动
    • 模型训练活动(GPU 密集型)
    • 模型验证活动
  • 由 App2 的活动完成触发
  • 向主工作流返回完成信号
  • 语言: Java

App4: 模型服务 - 强大的 GPU 应用程序,仅包含活动

  • 模型加载活动(GPU 内存密集型)
  • 推理设置活动(GPU 加速推理)
  • 由 App3 的工作流完成触发
  • 向主工作流返回完成信号
  • 语言: Go

多应用工作流

工作流执行路由基于托管 Dapr 应用程序的 App ID。 默认情况下,完整的工作流执行托管在启动该工作流的 App ID 上。该工作流可以在该 App ID 的任何副本上执行,而不仅仅是调度该工作流的单个副本。

可以通过在工作流执行代码中指定目标 App ID 参数,在不同的 App ID 上执行活动和子工作流。 执行时,目标 App ID 执行活动或子工作流,并将结果返回给原始 App ID 的父工作流。

整个工作流执行可以分布在多个 App ID 上,没有限制,每个活动或子工作流都可以指定目标 App ID。 工作流的最终历史记录将由托管最顶层父工作流(或可将其视为根工作流)的 App ID 保存。

错误处理

调用多应用活动或子工作流时:

  • 如果目标应用程序不存在,将使用提供的重试策略重试调用。
  • 如果目标应用程序存在但不包含指定的活动或工作流,调用将返回错误。
  • 标准工作流重试策略适用于多应用调用。

拥有不同 App ID 的团队之间必须进行协调,以确保活动和子工作流在需要时已定义并可用,这一点至关重要。

持久活动结果

活动通常需要一定的时间来完成,或者在资源或美元成本上执行起来很昂贵。 因此,即使在异常路径中,也不希望对同一轮次执行这些活动超过一次。 在 1.17 之前的多应用场景中,活动会通过网络调用将响应发布给托管拥有工作流的其他应用程序。 在托管工作流应用程序关闭或无法访问的情况下,结果将丢失,活动将被重试,从而导致活动的重复执行。

在 1.17 中,启用 `WorkflowsRemoteActivityReminder feature gate 将使活动结果在托管工作流应用程序处于离线或无法访问时,通过提醒发送给拥有该工作流的应用程序,从而确保结果不会丢失并避免重复执行。 在所有应用程序上使用 Dapr 1.17 版本的所有用户都应启用此选项。 为了在 Dapr 版本之间保持向后兼容性,该选项默认处于 禁用 状态,但将在未来的版本中默认启用。

多应用活动示例

Diagram showing multi-application call activity workflow pattern

以下示例展示如何在目标应用程序 App2 上执行活动 ActivityA

func BusinessWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var output string
	err := ctx.CallActivity("ActivityA",
		workflow.WithActivityInput("my-input"),
		workflow.WithActivityAppID("App2"), // 这里我们设置将执行此活动的目标 App ID。
	).Await(&output)

	if err != nil {
		return nil, err
	}

	return output, nil
}
public class BusinessWorkflow implements Workflow {
  @Override
  public WorkflowStub create() {
      return ctx -> {
          String output = ctx.callActivity(
                  ActivityA.class.getName(),
                  "my-input",
                  new WorkflowTaskOptions("App2"), // 这里我们设置将执行此活动的目标 App ID。
                  String.class
          ).await();

          ctx.complete(output);
      };
  }
}
@wfr.workflow
def app1_workflow(ctx: wf.DaprWorkflowContext):
  output = yield ctx.call_activity('ActivityA', input='my-input', app_id='App2')
  return output
public sealed class BusinessWorkflow : Workflow<string, string>
{
    public override async Task<string> RunAsync(WorkflowContext context, string input)
    {
        var options = new WorkflowTaskOptions { TargetAppId = "App2" };
        var output = await context.CallActivityAsync<string>(nameof(ActivityA), input, options);
        return output;
    }
}

多应用子工作流示例

Diagram showing multi-application child workflow pattern

以下示例展示如何在目标应用程序 App2 上执行子工作流 Workflow2

func BusinessWorkflow(ctx *workflow.WorkflowContext) (any, error) {
	var output string
	err := ctx.CallChildWorkflow("Workflow2",
		workflow.WithChildWorkflowInput("my-input"),
		workflow.WithChildWorkflowAppID("App2"), // 这里我们设置将执行此子工作流的目标 App ID。
	).Await(&output)

	if err != nil {
		return nil, err
	}

	return output, nil
}
@wfr.workflow
def workflow1(ctx: wf.DaprWorkflowContext):
  output = yield ctx.call_child_workflow(workflow='Workflow2', input='my-input', app_id='App2')
  return output
public sealed class BusinessWorkflow : Workflow<string, string>
{
    public override async Task<string> RunAsync(WorkflowContext context, string input)
    {
        var options = new ChildWorkflowTaskOptions { TargetAppId = "App2" };
        var output = await context.CallChildWorkflowAsync<string>(nameof(Workflow2), input, options);
        return output;
    }
}

相关链接

9 - 历史保留策略

定义保留策略以管理工作流状态历史记录

Dapr 工作流状态存储在[执行器状态存储]{{ %ref workflow-architecture.md#state-store-usage %}}中。 默认情况下,Dapr Workflows 会无限期保留工作流状态变更的完整历史记录。 这意味着历史工作流可以随时被查询和检查。 运行大量工作流或生成大量状态变更历史记录可能导致存储使用量增加,最终可能填满状态存储磁盘空间。

为帮助管理存储使用,Dapr Workflows 支持配置历史保留策略。 该保留策略定义了工作流状态变更历史在被删除前保留多长时间。 工作流只有在达到终态CompletedFailedTerminated)后才有资格被删除。 每个工作流终态都可以设置自定义的保留时长,也可以为未明确配置的终态设置默认保留时长。 时长定义为 Go duration string(例如,72h 表示 72 小时,30m 表示 30 分钟)。

Completed 工作流配置较短的保留时长可能很有用,同时为 FailedTerminated 工作流保留较长时间以便进行调查。

以下示例配置设置了每个终态。 这里设置的 anyTerminal 属性不会生效,因为所有终态都已明确配置,但它包含在此处作为参考。

有关如何将配置应用到 Dapr 应用程序的更多信息,请参阅 Dapr 配置文档

kind: Configuration
metadata:
  name: appconfig
spec:
  workflow:
    stateRetentionPolicy:
      anyTerminal: "360h"
      completed: "1m"
      failed: "720h"
      terminated: "360h"

相关链接

10 - 工作流执行并发

为 Dapr 工作流配置并发,以限制工作流和活动的执行速率。

您可以使用以下配置来设置在任何时候可以执行的最大并发工作流和活动数量。 这些限制是按每个边车实例实施的,这意味着如果您的工作流应用有 10 个副本,则有效限制将是配置值的 10 倍。

设置这些限制可以帮助防止您的 Dapr 边车和应用程序出现资源耗尽,或者在活动突发导致资源争用时帮助减少积压的工作流。 这些限制不会区分不同的工作流或活动定义,因此它们适用于在边车中运行的所有工作流和活动。

有关如何将配置应用到您的 Dapr 应用程序的更多信息,请参阅 Dapr 配置文档

apiVersion: dapr.io/v1alpha1
kind: Configuration
metadata:
  name: appconfig
spec:
  workflow:
    maxConcurrentWorkflowInvocations: 100 # 默认为无限
    maxConcurrentActivityInvocations: 1000 # 默认为无限

相关链接