1 - DaprWorkflowClient 生命周期管理与注册

了解如何配置 DaprWorkflowClient 生命周期管理与依赖注入

生命周期管理

DaprWorkflowClient 掌控着用于与 Dapr 边车通信的 TCP 套接字形式持有的网络资源,以及其他用于工作流管理和操作的类型。DaprWorkflowClient 实现了 IAsyncDisposable 以支持资源的主动清理。

依赖注入

AddDaprWorkflow() 方法会将 Dapr 工作流服务注册到 ASP.NET Core 的依赖注入容器中。这个方法需要一个选项委托,用于定义你希望在应用中注册和使用的每个工作流和活动。

修改 gRPC 消息大小限制

在注册时,你还可以为工作流客户端配置 gRPC 消息大小限制。当工作流负载超出默认 gRPC 限制时,这会很有用。

services
    .AddDaprWorkflowClient()
    .WithGrpcMessageSizeLimits( 
        maxReceiveMessageSize: 16 * 1024 * 1024, 
        maxSendMessageSize: 16 * 1024 * 1024);

单例注册

默认情况下,AddDaprWorkflow 方法以单例生命周期注册 DaprWorkflowClient 及其相关服务。这意味着这些服务只会被实例化一次。

以下示例展示了如何在典型的 Program.cs 文件中注册 DaprWorkflowClient

builder.Services.AddDaprWorkflow(options => {
    options.RegisterWorkflow<YourWorkflow>();
    options.RegisterActivity<YourActivity>();
});

var app = builder.Build();
await app.RunAsync();

作用域注册

虽然默认行为在你的场景中通常是可以接受的,但你可能希望覆盖指定的生命周期。这可以通过在 AddDaprWorkflow 中传递 ServiceLifetime 参数来实现。例如,你可能希望在 ASP.NET Core 处理管道中注入另一个需要 DaprClient 所用上下文的作用域服务,如果该服务被注册为单例,这些上下文将无法获取。

以下示例展示了如何操作:

builder.Services.AddDaprWorkflow(options => {
    options.RegisterWorkflow<YourWorkflow>();
    options.RegisterActivity<YourActivity>();
}, ServiceLifecycle.Scoped);

var app = builder.Build();
await app.RunAsync();

瞬态注册

最后,Dapr 服务也可以注册为瞬态生命周期,这意味着每次注入时都会重新初始化。以下示例展示了如何操作:

builder.Services.AddDaprWorkflow(options => {
    options.RegisterWorkflow<YourWorkflow>();
    options.RegisterActivity<YourActivity>();
}, ServiceLifecycle.Transient);

var app = builder.Build();
await app.RunAsync();

创建 DaprWorkflowClient 实例

在 ASP.Net Core 应用中,你可以通过方法注入或构造函数注入将 DaprWorkflowClient 注入到方法或控制器中。本示例展示了在最小 API 场景下的方法注入:

app.MapPost("/start", async (
    [FromServices] DaprWorkflowClient daprWorkflowClient,
    Order order
    ) => {
        var instanceId = await daprWorkflowClient.ScheduleNewWorkflowAsync(
            nameof(OrderProcessingWorkflow),
            input: order);

        return Results.Accepted(instanceId);
});

要在控制台应用中创建 DaprWorkflowClient 实例,请从 ServiceProvider 中获取它:

using var scope = host.Services.CreateAsyncScope();
var daprWorkflowClient = scope.ServiceProvider.GetRequiredService<DaprWorkflowClient>();

现在,你可以使用此客户端执行工作流管理操作,例如启动、暂停、恢复和终止工作流实例。有关这些操作的更多信息,请参阅使用 DaprWorkflowClient 进行工作流管理操作

将服务注入到工作流活动中

工作流活动支持开发者对现代 C# 应用程序所期望的相同依赖注入方式。假设在启动时进行了正确的注册,任何此类类型都可以注入到工作流活动的构造函数中,并在工作流执行期间供其使用。这使得通过注入 ILogger 轻松添加日志记录,或通过注入 DaprClientDaprJobsClient 访问其他 Dapr 构建块变得简单。

internal sealed class SquareNumberActivity : WorkflowActivity<int, int>
{
    private readonly ILogger _logger;
    
    public MyActivity(ILogger logger)
    {
        this._logger = logger;
    }
    
    public override Task<int> RunAsync(WorkflowActivityContext context, int input) 
    {
        this._logger.LogInformation("Squaring the value {number}", input);
        var result = input * input;
        this._logger.LogInformation("Got a result of {squareResult}", result);
        
        return Task.FromResult(result);
    }
}

活动任务执行标识符

从 Dapr .NET SDK v1.17.0 开始,WorkflowActivityContext 暴露了一个任务执行标识符,该标识符具有以下特性:

  • 对每个活动任务唯一
  • 在重试期间保持稳定

这使得它可用于幂等键、任务级状态跟踪和日志关联。

internal sealed class IdempotentActivity : WorkflowActivity<int, int>
{
    public override Task<int> RunAsync(WorkflowActivityContext context, int input)
    {
        var executionId = context.TaskExecutionId;
        // 将 executionId 用作幂等键或任务状态键。

        return Task.FromResult(input * input);
    }
}

在工作流中使用 ILogger

由于工作流必须是确定性的,因此无法向其注入任意服务。例如,如果能够将标准 ILogger 注入到工作流中,并且由于错误需要重放工作流,那么从事件源日志进行的后续重放将导致日志记录额外的操作,这些操作实际上并没有发生第二次或第三次,因为它们的结果是从日志中获取的。这可能会引入大量的混淆。相反,提供了一个重放安全的记录器供在工作流内使用。它只会在工作流首次运行时记录事件,而在重放工作流时不会记录任何内容。

此记录器可以从工作流实例上可用的 WorkflowContext 中存在的方法获取,并且可以像使用 ILogger 实例一样精确使用。

.NET SDK 仓库中可以看到演示此功能的完整示例,但下面提供了该示例的简要摘录。

public class OrderProcessingWorkflow : Workflow<OrderPayload, OrderResult>
{
    public override async Task<OrderResult> RunAsync(WorkflowContext context, OrderPayload order)
    {
        string orderId = context.InstanceId;
        var logger = context.CreateReplaySafeLogger<OrderProcessingWorkflow>(); // 使用此方法访问记录器实例

        logger.LogInformation("Received order {orderId} for {quantity} {name} at ${totalCost}", orderId, order.Quantity, order.Name, order.TotalCost);
        
        //...
    }
}

后续步骤

2 - .NET SDK 中的工作流序列化

配置 Dapr .NET SDK 的工作流序列化

概述

从 Dapr .NET SDK v1.17.0 开始,Dapr.Workflow 支持可插拔序列化。SDK 默认继续使用 System.Text.Json,但现在您可以:

  • 覆盖默认的 System.Text.Json 设置。
  • 注册自定义序列化器(例如,MessagePack 或 BSON)。

序列化配置完全在客户端进行,不需要特定版本的 Dapr 运行时。

兼容性和重大变更

默认 JSON 序列化

默认情况下,.NET SDK 使用 System.Text.Json 并采用 JsonSerializerDefaults.Web(请参阅 JsonSerializerDefaults.Web 参考)。 这意味着:

  • 属性名称不区分大小写。
  • 属性名称使用 “camelCase” 格式。
  • 读取时允许带引号的数字(数字属性的 JSON 字符串)。

此默认约定旨在与其他 Dapr 语言 SDK 兼容,以支持多应用工作流。

覆盖 System.Text.Json 默认值

要覆盖默认 JSON 设置,请使用工作流构建器注册工作流客户端,以便您可以提供自定义的 JsonSerializerOptions

builder.Services
    .AddDaprWorkflowBuilder(options =>
    {
        options.RegisterWorkflow<MyWorkflow>();
        options.RegisterActivity<MyActivity>();
    })
    .WithJsonSerializer(new JsonSerializerOptions { PropertyNamingPolicy = null });

从 DI 解析的所有 DaprWorkflowClient 实例都将使用提供的 JsonSerializerOptions 来处理工作流和活动负载。

自定义序列化提供程序

自定义序列化器必须实现 IWorkflowSerializer 接口。以下示例展示了基于 MessagePack 的实现,该实现将数据编码为 Base64 字符串进行传输:

public sealed class MessagePackWorkflowSerializer : IWorkflowSerializer
{
    private readonly MessagePackSerializerOptions _options;

    public MessagePackWorkflowSerializer(MessagePackSerializerOptions options)
    {
        _options = options;
    }

    /// <inheritdoc/>
    public string Serialize(object? value, Type? inputType = null)
    {
        if (value == null)
        {
            return string.Empty;
        }

        var targetType = inputType ?? value.GetType();
        var bytes = MessagePackSerializer.Serialize(targetType, value, _options);
        return Convert.ToBase64String(bytes);
    }

    /// <inheritdoc/>
    public T? Deserialize<T>(string? data)
    {
        return (T?)Deserialize(data, typeof(T));
    }

    /// <inheritdoc/>
    public object? Deserialize(string? data, Type returnType)
    {
        if (returnType == null)
        {
            throw new ArgumentNullException(nameof(returnType));
        }

        if (string.IsNullOrEmpty(data))
        {
            return default;
        }

        try
        {
            var bytes = Convert.FromBase64String(data);
            return MessagePackSerializer.Deserialize(returnType, bytes, _options);
        }
        catch (FormatException ex)
        {
            throw new InvalidOperationException(
                "Failed to decode Base64 data. The input may not be valid MessagePack-serialized data.",
                ex);
        }
        catch (MessagePackSerializationException ex)
        {
            throw new InvalidOperationException(
                $"Failed to deserialize data to type {returnType.FullName}.",
                ex);
        }
    }
}

注册自定义序列化器

使用工作流构建器注册序列化器:

builder.Services
    .AddDaprWorkflowBuilder(options =>
    {
        options.RegisterWorkflow<MyWorkflow>();
        options.RegisterActivity<MyActivity>();
    })
    .WithSerializer(new MessagePackWorkflowSerializer(MessagePackSerializerOptions.Standard));

如果需要 DI 提供的配置,请使用接收 IServiceProvider 的重载:

builder.Services
    .AddDaprWorkflowBuilder(options =>
    {
        options.RegisterWorkflow<MyWorkflow>();
        options.RegisterActivity<MyActivity>();
    })
    .WithSerializer(serviceProvider =>
    {
        var options = serviceProvider
            .GetRequiredService<IOptions<MessagePackSerializerOptions>>()
            .Value;

        return new MessagePackWorkflowSerializer(options);
    });

3 - .NET SDK 中的多应用程序工作流

使用 .NET SDK 调用托管在其他 Dapr 应用程序中的活动和子工作流

概述

Dapr 工作流可以调用托管在不同 Dapr 应用程序中的活动或子工作流。在 .NET 中,多应用程序工作流的支持从以下版本开始:

  • Dapr 运行时 v1.16.0+
  • Dapr .NET SDK v1.17.0+

概念性指导和约束在 多应用程序工作流 中介绍。

要求

多应用程序工作流调用需要:

  • 目标应用 ID 必须存在,并且必须注册你调用的活动或工作流。
  • 所有参与的应用 ID 必须位于同一命名空间中。
  • 所有参与的应用 ID 必须使用相同的工作流(actor)状态存储。

在另一个应用程序中调用活动

在调用活动时,在 WorkflowTaskOptions 上设置 TargetAppId 以在另一个应用程序中执行它:

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;
    }
}

父工作流继续在本地进行编排并接收活动结果。

在另一个应用程序中调用子工作流

在调用子工作流时,在 ChildWorkflowTaskOptions 上设置 TargetAppId 以在另一个应用程序中执行它:

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;
    }
}

后续步骤

4 - 使用 DaprWorkflowClient 进行工作流管理操作

了解如何使用 DaprWorkflowClient 管理工作流

使用 DaprWorkflowClient 进行工作流管理操作

DaprWorkflowClient 类提供了管理工作流实例的方法。以下是你可以使用 DaprWorkflowClient 执行的操作。

调度新的工作流实例

要启动新的工作流实例,请使用 ScheduleNewWorkflowAsync 方法。此方法需要工作流类型名称和工作流所需的输入参数。工作流的 instanceId 是一个可选参数;如果未提供,DaprWorkflowClient 会生成一个新的 GUID。最后一个可选参数是 DateTimeOffset 类型的 startTime,可用于定义工作流实例应该何时开始。该方法返回已调度工作流的 instanceId,用于其他工作流管理操作。

var instanceId = $"order-workflow-{Guid.NewGuid().ToString()[..8]}";
var input = new Order("Paperclips", 1000, 9.95);
await daprWorkflowClient.ScheduleNewWorkflowAsync(
  nameof(OrderProcessingWorkflow),
  instanceId,
  input);

获取工作流实例的状态

要获取工作流实例的当前状态,请使用 GetWorkflowStateAsync 方法。此方法需要工作流的实例 ID,并返回一个包含工作流当前状态详细信息的 WorkflowStatus 对象。

var workflowStatus = await daprWorkflowClient.GetWorkflowStateAsync(instanceId);

向正在运行的工作流实例发送事件

要向正在等待外部事件的运行中工作流实例发送事件,请使用 RaiseEventAsync 方法。此方法需要工作流的实例 ID、事件名称,以及可选的事件负载。

await daprWorkflowClient.RaiseEventAsync(instanceId, "Approval", true);

暂停正在运行的工作流实例

可以使用 SuspendWorkflowAsync 方法暂停正在运行的工作流实例。此方法需要工作流的实例 ID。你可以选择性地提供暂停工作流的原因。

await daprWorkflowClient.SuspendWorkflowAsync(instanceId);

恢复已暂停的工作流实例

可以使用 ResumeWorkflowAsync 方法恢复已暂停的工作流实例。此方法需要工作流的实例 ID。你可以选择性地提供恢复工作流的原因。

await daprWorkflowClient.ResumeWorkflowAsync(instanceId);

终止工作流实例

要终止工作流实例,请使用 TerminateWorkflowAsync 方法。此方法需要工作流的实例 ID。你可以选择性地提供一个 string 类型的 output 参数。终止工作流实例也会终止所有子工作流实例,但对正在执行的活动没有影响。

await daprWorkflowClient.TerminateWorkflowAsync(instanceId);

清除工作流实例

要从 Dapr Workflow 状态存储中删除工作流实例历史记录,请使用 PurgeWorkflowAsync 方法。此方法需要工作流的实例 ID。只有已完成、失败或已终止的工作流实例才能被清除。

await daprWorkflowClient.PurgeWorkflowAsync(instanceId);

后续步骤

5 - .NET SDK 中的工作流版本控制

了解如何在 Dapr .NET SDK 中使用基于补丁和基于名称的工作流版本控制

概述

Dapr 工作流版本控制允许你演进工作流,而不会中断进行中实例的确定性执行。 .NET SDK 支持两种方法:

  • 基于补丁的版本控制:引入由 context.IsPatched("patch-name") 守护的条件分支。
  • 基于名称的版本控制:创建一个新的工作流类型名称,并让版本控制策略选择最新版本。

对于小型就地更改,使用基于补丁的版本控制。对于需要全新的工作流类型的大型重构,使用基于名称的版本控制。

何时使用每种方法

基于补丁的版本控制 适用于以下情况:

  • 你需要在现有工作流中进行小型、增量式的更改。
  • 你希望现有实例在部署后保持确定性行为。
  • 你希望暂时避免引入新的工作流类型。

基于名称的版本控制 适用于以下情况:

  • 你想要一个没有累积补丁的全新工作流类型。
  • 你准备好删除旧的补丁块并更自由地重构。
  • 你希望基于命名约定自动选择版本。

基于补丁的版本控制

基于补丁的版本控制依赖于工作流内部的确定性开关。使用 WorkflowContext.IsPatched 来守护新行为:

public override async Task RunAsync(WorkflowContext context, OrderPayload input)
{
    await context.CallActivityAsync(nameof(ReserveInventoryActivity), input);

    if (context.IsPatched("v2"))
    {
        await context.CallActivityAsync(nameof(ChargePaymentActivityV2), input);
    }
    else
    {
        await context.CallActivityAsync(nameof(ChargePaymentActivity), input);
    }
}

补丁规则

  • 补丁名称可以在同一工作流中多次出现,并且可以嵌套。
  • 补丁名称在部署中必须唯一。例如,如果你部署了一个带有补丁名称 "v1" 的工作流,
  • 你不得在后续编辑中重复使用 "v1"。请使用新的标识符(如 "v2")以避免非确定性行为。
  • IsPatchedWorkflowContext 上可用;无需额外设置。

基于名称的版本控制

基于名称的版本控制允许你通过更改工作流类型名称来创建新版本的工作流。推荐的模式是将现有工作流复制到新文件中,重命名类,根据需要进行重构,然后在必要时再次开始打补丁。

例如,如果你有 OrderWorkflow,则创建 OrderWorkflowV2 并重构它。较旧的版本可以保留给进行中的实例,而新实例使用最新版本。

默认命名行为

默认情况下,基于名称的版本控制使用内置的 NumericVersionStrategy 和数字后缀。以下都是有效示例:

  • MyWorkflow(被视为版本 0
  • MyWorkflow2
  • MyWorkflowV2

默认策略假设较高的数字值是较新的版本(例如,MyWorkflowV10MyWorkflowV2 更新)。.NET SDK 还包含其他内置策略(Date、SemVer 和 Numeric)以及对自定义策略的支持。

内置策略和选项

.NET SDK 附带了几个内置的基于名称的策略。每个策略都支持允许你调整如何解析后缀以及当没有后缀时该如何处理的选项。

  • DateVersionStrategy:从尾随后缀派生基于日期的版本(例如,MyWorkflow20220611)。 选项包括:
    • 日期格式:使用标准 C# 日期格式规则;默认为 yyyyMMdd
    • 默认版本:当未提供后缀时使用;默认为 0
    • 前缀:在日期后缀之前匹配的可选前缀,带有可选区分大小写。
  • SemVerVersionStrategy:从尾随后缀派生 SemVer 版本(例如,MyWorkflow1.2.3)。 选项包括:
    • 前缀:在 SemVer 后缀之前匹配的可选前缀,带有可选区分大小写。
    • 预发布/生成支持:可以解析预发布注释和生成元数据。
    • 默认版本:当未提供后缀时的可选默认值(如果配置为允许缺少后缀)。
  • NumericVersionStrategy:从尾随后缀派生数字版本(例如,MyWorkflow42MyWorkflowV42)。 选项包括:
    • 前缀:在数字后缀之前匹配的可选前缀,带有可选区分大小写。
    • 零填充宽度:可选宽度,允许使用带前导零的固宽数字。
    • 默认版本:当未提供后缀时使用。

配置基于名称的版本控制

1. 安装版本控制包

Dapr.Workflow.Versioning 包添加到你的项目中。

2. 注册工作流版本控制

在启动期间将版本控制添加到 DI:

builder.Services.AddDaprWorkflowVersioning();

3. 选择策略(可选)

你可以在调用 AddDaprWorkflowVersioning 后在 DI 中注册策略来选择:

builder.Services.UseDefaultWorkflowStrategy<NumericVersionStrategy>("workflow-versioning-options");

可选字符串键用于定位策略选项。

4. 配置策略选项(可选)

使用相同的键注册策略选项:

builder.Services.ConfigureStrategyOptions<NumericVersionStrategyOptions>("workflow-versioning-options", o =>
{
    o.SuffixPrefix = "V";
});

5. 注册活动

活动照常注册。使用基于名称的版本控制时不需要注册工作流,因为源生成器会在构建时自动发现它们。

builder.Services.AddDaprWorkflow(w =>
{
    w.RegisterActivity<SendEmailActivity>();
});

配置完成后,命名工作流版本控制将在运行时自动应用。

跨程序集工作流发现

默认情况下,工作流版本控制源生成器仅扫描执行的程序集。如果你将工作流保留在单独的引用程序集中,则除非你选择加入引用扫描,否则不会发现这些实现。

默认情况下禁用引用扫描,因为它可能会增加构建时间(生成器必须检查所有引用的程序集以查找 Workflow<,> 实现)。要启用它,请将以下内容添加到执行应用程序的 .csproj 文件中:

<ItemGroup>
  <CompilerVisibleProperty Include="DaprWorkflowVersioningScanReferences" />
</ItemGroup>

启用后,源生成器会将引用程序集中发现的任何工作流实现添加到用于版本跟踪的内部注册表中。

覆盖名称和版本

如果你需要覆盖从工作流类型检测到的规范名称或版本,请将 [WorkflowVersion] 属性应用于实现工作流的类,并显式指定值。

最佳实践

  • 按规范名称调度:调度新的工作流实例时,使用规范工作流名称(未版本化的名称)。SDK 会自动发现版本化类型并将规范名称映射到最新版本。避免使用特定的版本化名称进行调度。
  • 保留旧版本:无限期保留较旧的工作流类型。只有在确定没有进行中的实例引用它们时才能删除它们。过早删除旧版本可能会导致长时间运行的工作流停滞。
  • 单向版本控制:始终向前推进版本。避免重命名或重复使用较旧的版本标识符。

当前限制

  • 单个项目不能为不同的工作流混合使用多个版本控制策略。
  • 没有从一个版本控制策略到另一个策略的自动迁移。
  • 工作流版本控制不能跨应用程序边界工作。如果你使用多应用工作流,必须根据目标应用的任何版本控制策略使用该应用期望的类型名称。调用此 .NET 应用程序的其他应用程序如果设置为使用基于名称的工作流版本控制,则可以简单地使用规范名称。

示例项目

端到端示例可在 Dapr .NET SDK 仓库的 examples/Workflow/WorkflowVersioning 中找到。

后续步骤

6 - .NET 工作流示例

探索 GitHub 上的 Dapr 工作流代码示例

Dapr Quickstarts 仓库中的工作流教程

GitHub 上的 Dapr Quickstarts 仓库包含许多工作流教程,展示各种工作流模式以及如何使用工作流管理操作。你可以在 quickstarts/tutorials/workflow/csharp 文件夹中找到这些教程。

.NET SDK 仓库中的工作流示例

GitHub 上的 Dapr .NET SDK 仓库包含多个示例,演示如何将 Dapr 工作流与 .NET 结合使用。你可以在 examples/Workflow 文件夹中找到这些示例。

后续步骤