Dapr Messaging .NET SDK

• 1: 如何操作：使用 .NET SDK 创建和管理 Dapr 流式订阅
• 2: DaprPublishSubscribeClient 用法

1 - 如何操作：使用 .NET SDK 创建和管理 Dapr 流式订阅

让我们使用流式处理能力创建一个对发布/订阅主题或队列的订阅。在接下来的演示中，我们将使用此处提供的简单示例，作为说明如何配置消息处理程序的指南，这些配置在运行时进行，无需预先配置端点。在本指南中，您将：

• 部署 .NET Web API 应用程序 (StreamingSubscriptionExample)
• 使用 Dapr .NET Messaging SDK 动态订阅发布/订阅主题。

前置条件

• Dapr CLI
• 已初始化的 Dapr 环境
• 已安装 .NET 8、.NET 9 或 .NET 10
• 已将 Dapr.Messaging NuGet 包安装到您的项目中

设置环境

克隆 .NET SDK 仓库。

git clone https://github.com/dapr/dotnet-sdk.git

从 .NET SDK 根目录导航到 Dapr 流式 PubSub 示例。

cd examples/Client/PublishSubscribe

在本地运行应用程序

要运行 Dapr 应用程序，您需要启动 .NET 程序和 Dapr 边车。导航到 StreamingSubscriptionExample 目录。

cd StreamingSubscriptionExample

我们将运行一个同时启动 Dapr 边车和 .NET 程序的命令。

dapr run --app-id pubsubapp --dapr-grpc-port 4001 --dapr-http-port 3500 -- dotnet run

Dapr 在 http://localhost:3500 监听 HTTP 请求，在 http://localhost:4001 监听内部 Jobs gRPC 请求。

使用依赖注入注册 Dapr PubSub 客户端

Dapr Messaging SDK 提供了一个扩展方法来简化 Dapr PubSub 客户端的注册。在完成 Program.cs 中的依赖注入注册之前，添加以下行：

var builder = WebApplication.CreateBuilder(args);

//Add anywhere between these two
builder.Services.AddDaprPubSubClient(); //That's it

var app = builder.Build();

您可能需要为 Dapr PubSub 客户端提供一些配置选项，这些选项应在每次对边车的调用时都存在，例如 Dapr API 令牌，或者您想使用非标准的 HTTP 或 gRPC 端点。这可以通过使用注册方法的重载来实现，该方法允许配置 DaprPublishSubscribeClientBuilder 实例：

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddDaprPubSubClient((_, daprPubSubClientBuilder) => {
    daprPubSubClientBuilder.UseDaprApiToken("abc123");
    daprPubSubClientBuilder.UseHttpEndpoint("http://localhost:8512"); //Non-standard sidecar HTTP endpoint
});

var app = builder.Build();

不过，您可能希望从其他来源检索要注入的值，而这些值本身已注册为依赖项。您还可以使用另一个重载将 IServiceProvider 注入到配置操作方法中。在以下示例中，我们注册一个虚拟的单例，该单例可以从某处检索机密，并将其传递给 AddDaprJobClient 的配置方法，以便我们可以从其他地方检索我们的 Dapr API 令牌以在此处进行注册：

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddSingleton<SecretRetriever>();
builder.Services.AddDaprPubSubClient((serviceProvider, daprPubSubClientBuilder) => {
    var secretRetriever = serviceProvider.GetRequiredService<SecretRetriever>();
    var daprApiToken = secretRetriever.GetSecret("DaprApiToken").Value;
    daprPubSubClientBuilder.UseDaprApiToken(daprApiToken);
    
    daprPubSubClientBuilder.UseHttpEndpoint("http://localhost:8512");
});

var app = builder.Build();

使用 IConfiguration 配置 Dapr PubSub 客户端

也可以使用已注册的 IConfiguration 中的值来配置 Dapr PubSub 客户端，而无需按照上一节演示的那样使用 DaprPublishSubscribeClientBuilder 显式指定每个值覆盖。相反，通过填充通过依赖注入提供的 IConfiguration，AddDaprPubSubClient() 注册将自动使用这些值而不是其各自的默认值。

首先填充配置中的值。可以通过以下几种不同方式来完成此操作，如下所示。

通过 ConfigurationBuilder 配置

可以在不使用配置源的情况下配置应用程序设置，而是使用 ConfigurationBuilder 实例在内存中填充值：

var builder = WebApplication.CreateBuilder();

//Create the configuration
var configuration = new ConfigurationBuilder()
    .AddInMemoryCollection(new Dictionary<string, string> {
            { "DAPR_HTTP_ENDPOINT", "http://localhost:54321" },
            { "DAPR_API_TOKEN", "abc123" }
        })
    .Build();

builder.Configuration.AddConfiguration(configuration);
builder.Services.AddDaprPubSubClient(); //This will automatically populate the HTTP endpoint and API token values from the IConfiguration

通过环境变量配置

可以从应用程序可用的环境变量访问应用程序设置。

以下环境变量将用于填充注册 Dapr PubSub 客户端所使用的 HTTP 端点和 API 令牌。

var builder = WebApplication.CreateBuilder();

builder.Configuration.AddEnvironmentVariables();
builder.Services.AddDaprPubSubClient();

Dapr PubSub 客户端将被配置为使用 HTTP 端点 http://localhost:54321，并在所有出站请求中填充 API 令牌头 abc123。

通过带前缀的环境变量配置

但是，在不使用容器的共享主机场景中（在同一台机器上运行多个应用程序）或在开发环境中，为环境变量添加前缀并不罕见。以下示例假定 HTTP 端点和 API 令牌都将从带有值 “myapp_” 前缀的环境变量中提取。在此场景中使用的两个环境变量如下：

这些环境变量将在以下示例中加载到已注册的配置中，并在不带前缀的情况下可用。

var builder = WebApplication.CreateBuilder();

builder.Configuration.AddEnvironmentVariables(prefix: "myapp_");
builder.Services.AddDaprPubSubClient();

Dapr PubSub 客户端将被配置为使用 HTTP 端点 http://localhost:54321，并在所有出站请求中填充 API 令牌头 abc123。

不依赖依赖注入使用 Dapr PubSub 客户端

虽然使用依赖注入简化了 .NET 中复杂类型的使用，并使处理复杂配置变得更加容易，但您不需要以这种方式注册 DaprPublishSubscribeClient。相反，您也可以选择从 DaprPublishSubscribeClientBuilder 实例创建它的实例，如下所示：

public class MySampleClass
{
    public void DoSomething()
    {
        var daprPubSubClientBuilder = new DaprPublishSubscribeClientBuilder();
        var daprPubSubClient = daprPubSubClientBuilder.Build();

        //Do something with the `daprPubSubClient`
    }
}

设置消息处理程序

Dapr 中的流式订阅实现通过将消息保留在 Dapr 运行时中，直到您的应用程序准备好接受它们为止，从而为您提供更大的控制权来处理来自事件的背压。.NET SDK 支持高性能队列，用于在处理挂起时在应用程序中维护这些消息的本地缓存。这些消息将保留在队列中，直到每个消息的处理超时或对每个消息执行响应操作（通常在处理成功或失败之后）。在 Dapr 运行时收到此响应操作之前，消息将由 Dapr 持久化，并在服务故障的情况下可用。

可用的各种响应操作如下：

处理程序一次将只接收一条消息，如果向订阅提供了取消令牌，则该令牌将在处理程序调用期间提供。

处理程序必须配置为返回 Task<TopicResponseAction>，以指示这些操作之一，即使来自 try/catch 块。如果您的处理程序未捕获异常，订阅将使用订阅注册期间在选项中配置的响应操作。

以下演示了示例中提供的示例消息处理程序：

Task<TopicResponseAction> HandleMessageAsync(TopicMessage message, CancellationToken cancellationToken = default)
{
    try
    {
        //Do something with the message
        Console.WriteLine(Encoding.UTF8.GetString(message.Data.Span));
        return Task.FromResult(TopicResponseAction.Success);
    }
    catch
    {
        return Task.FromResult(TopicResponseAction.Retry);
    }
}

配置和订阅 PubSub 主题

流式订阅的配置需要向 Dapr 注册的 PubSub 组件的名称、正在订阅的主题或队列的名称、提供订阅配置的 DaprSubscriptionOptions、消息处理程序和可选的取消令牌。DaprSubscriptionOptions 唯一必需的参数是默认的 MessageHandlingPolicy，它由每个事件的超时和在该超时时采取的 TopicResponseAction 组成。

其他选项如下：

然后，订阅将配置为以下示例：

var messagingClient = app.Services.GetRequiredService<DaprPublishSubscribeClient>();

var cancellationTokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(60)); //Override the default of 30 seconds
var options = new DaprSubscriptionOptions(new MessageHandlingPolicy(TimeSpan.FromSeconds(10), TopicResponseAction.Retry));
var subscription = await messagingClient.SubscribeAsync("pubsub", "mytopic", options, HandleMessageAsync, cancellationTokenSource.Token);

终止和清理订阅

当您完成订阅并希望停止接收新事件时，只需在订阅实例上等待对 DisposeAsync() 的调用。这将导致客户端取消注册其他事件，并在释放任何内部资源之前继续处理背压队列中仍然剩余的所有事件（如果有）。此清理将受到订阅注册时在 DaprSubscriptionOptions 中提供的超时间隔的限制，默认情况下，这设置为 30 秒。

2 - DaprPublishSubscribeClient 用法

生命周期管理

DaprPublishSubscribeClient 是 Dapr 客户端的一个版本，专门用于与 Dapr 消息传递 API 交互。 它可以与 DaprClient 和其他 Dapr 客户端一起注册而不会产生问题。

它维护对网络资源的访问，这些资源以用于与 Dapr 边车通信的 TCP 套接字形式存在，并实现 IAsyncDisposable 以支持资源的积极清理。

为获得最佳性能，应创建一个 DaprPublishSubscribeClient 的单一长期实例，并在整个应用程序中提供对该共享 实例的访问。DaprPublishSubscribeClient 实例是线程安全的，旨在共享使用。

可以利用依赖注入功能来辅助实现这一点。注册方法支持注册为 单例、作用域实例或瞬态（意味着每次注入时都会重新创建），但同时也支持 利用 IConfiguration 或其他注入服务的值进行注册，这在每次 从头创建客户端时是不切实际的。

避免为每个操作创建一个 DaprPublishSubscribeClient 并在操作完成时将其释放。 DaprPublishSubscribeClient 仅应在您不再希望在订阅上接收事件时才被释放，因为释放它将取消新事件的持续接收。

通过 DaprPublishSubscribeClientBuilder 配置 DaprPublishSubscribeClient

可以通过在调用 .Build() 创建客户端本身之前调用 DaprPublishSubscribeClientBuilder 类上的方法来配置 DaprPublishSubscribeClient。每个 DaprPublishSubscribeClient 的设置是独立的， 无法在调用 .Build() 后更改。

var daprPubsubClient = new DaprPublishSubscribeClientBuilder()
    .UseDaprApiToken("abc123") // 指定用于向其他 Dapr 边车进行身份验证的 API 令牌
    .Build();

DaprPublishSubscribeClientBuilder 包含以下设置：

• Dapr 边车的 HTTP 端点
• Dapr 边车的 gRPC 端点
• 用于配置 JSON 序列化的 JsonSerializerOptions 对象
• 用于配置 gRPC 的 GrpcChannelOptions 对象
• 用于向边车验证请求的 API 令牌
• 用于创建 SDK 使用的 HttpClient 实例的工厂方法
• 在向边车发出请求时 HttpClient 实例使用的超时时间

SDK 将读取以下环境变量来配置默认值：

• DAPR_HTTP_ENDPOINT：用于查找 Dapr 边车的 HTTP 端点，示例：https://dapr-api.mycompany.com
• DAPR_GRPC_ENDPOINT：用于查找 Dapr 边车的 gRPC 端点，示例：https://dapr-grpc-api.mycompany.com
• DAPR_HTTP_PORT：如果未设置 DAPR_HTTP_ENDPOINT，则使用此项来查找 Dapr 边车的 HTTP 本地端点
• DAPR_GRPC_PORT：如果未设置 DAPR_GRPC_ENDPOINT，则使用此项来查找 Dapr 边车的 gRPC 本地端点
• DAPR_API_TOKEN：用于设置 API 令牌

配置 gRPC 通道选项

Dapr 使用 CancellationToken 进行取消依赖于 gRPC 通道选项的配置。如果您 需要自己配置这些选项，请确保启用 ThrowOperationCanceledOnCancellation 设置。

var daprPubsubClient = new DaprPublishSubscribeClientBuilder()
    .UseGrpcChannelOptions(new GrpcChannelOptions { ... ThrowOperationCanceledOnCancellation = true })
    .Build();

在 DaprPublishSubscribeClient 中使用取消

DaprPublishSubscribeClient 上的 API 执行异步操作并接受一个可选的 CancellationToken 参数。这遵循 .NET 用于可取消操作的标准实践。请注意，当发生取消时，无法 保证远程端点停止处理请求，只能保证客户端已停止等待完成。

当操作被取消时，它将抛出 OperationCancelledException。

通过依赖注入配置 DaprPublishSubscribeClient

使用用于在依赖注入容器中注册 DaprPublishSubscribeClient 的内置扩展方法 可以带来以下好处：一次性注册长期服务、集中复杂配置，并通过确保类似的长期资源在可能的情况下被重用（例如 HttpClient 实例）来提高性能。

有三种重载可用，为开发人员在其场景中配置客户端提供最大的灵活性。 如果尚未注册，每个重载都将代表您注册 IHttpClientFactory，并配置 DaprPublishSubscribeClientBuilder 在创建 HttpClient 实例时使用它，以便尽可能重用同一实例并避免套接字耗尽和其他问题。

在第一种方法中，开发人员不进行任何配置，DaprPublishSubscribeClient 使用默认设置进行配置。

var builder = WebApplication.CreateBuilder(args);

builder.Services.DaprPublishSubscribeClient(); //注册 `DaprPublishSubscribeClient` 以便根据需要注入
var app = builder.Build();

有时开发人员需要使用上述各种配置选项来配置创建的客户端。这是通过传入 DaprJobsClientBuiler 的重载来完成的，该重载公开了配置必要选项的方法。

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddDaprJobsClient((_, daprPubSubClientBuilder) => {
   //设置 API 令牌
   daprPubSubClientBuilder.UseDaprApiToken("abc123");
   //指定非标准的 HTTP 端点
   daprPubSubClientBuilder.UseHttpEndpoint("http://dapr.my-company.com");
});

var app = builder.Build();

最后，开发人员可能需要从另一个服务检索信息以填充这些配置值。该值可以从 DaprClient 实例、供应商特定的 SDK 或某个本地服务提供，但只要它也在 DI 中注册，就可以通过最后一个重载注入到此配置操作中：

var builder = WebApplication.CreateBuilder(args);

//注册一个从某处检索机密的虚构服务
builder.Services.AddSingleton<SecretService>();

builder.Services.AddDaprPublishSubscribeClient((serviceProvider, daprPubSubClientBuilder) => {
    //从服务提供程序检索 `SecretService` 的实例
    var secretService = serviceProvider.GetRequiredService<SecretService>();
    var daprApiToken = secretService.GetSecret("DaprApiToken").Value;

    //配置 `DaprPublishSubscribeClientBuilder`
    daprPubSubClientBuilder.UseDaprApiToken(daprApiToken);
});

var app = builder.Build();