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

Return to the regular view of this page.

可插拔组件 SDK

用你喜欢的语言开发可插拔组件

1: Dapr 可插拔组件 .NET SDK 入门

1.1: 实现 .NET 输入/输出绑定组件

1.2: 实现 .NET 发布订阅组件

1.3: 实现 .NET 状态存储组件

1.4: Dapr 可插拔组件 .NET SDK 的高级用法

1.4.1: .NET Dapr 可插拔组件中的多个服务
1.4.2: .NET Dapr 可插拔组件的应用环境
1.4.3: .NET Dapr 可插拔组件的生命周期

2: Dapr 可插拔组件 Go SDK 入门

2.1: 实现 Go 输入/输出绑定组件

2.2: 实现 Go 发布订阅组件

2.3: 实现 Go 状态存储组件

2.4: Dapr 可插拔组件 Go SDK 的高级用法

Dapr SDK 是创建可插拔组件的最简单方式。选择你喜欢的语言，在几分钟内开始创建组件。

可插拔组件 SDK

语言	状态
Go	开发中
.NET	开发中

1 - Dapr 可插拔组件 .NET SDK 入门

如何开始使用 Dapr 可插拔组件 .NET SDK

Dapr 提供 NuGet 包以帮助开发 .NET 可插拔组件。

前置条件

.NET 6 SDK 或更高版本
Dapr 1.9 CLI 或更高版本
已初始化的 Dapr 环境
Linux、Mac 或 Windows（配合 WSL）

注意

在 Windows 上开发 Dapr 可插拔组件需要 WSL，因为某些开发平台在"原生" Windows 上不完全支持 Unix 域套接字。

创建项目

创建可插拔组件始于一个空的 ASP.NET 项目。

dotnet new web --name <project name>

添加 NuGet 包

添加 Dapr .NET 可插拔组件 NuGet 包。

dotnet add package Dapr.PluggableComponents.AspNetCore

创建应用和服务

创建 Dapr 可插拔组件应用类似于创建 ASP.NET 应用。在 Program.cs 中，将 WebApplication 相关代码替换为 Dapr 等效的 DaprPluggableComponentsApplication。

using Dapr.PluggableComponents;

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "<socket name>",
    serviceBuilder =>
    {
        // Register one or more components with this service.
    });

app.Run();

这将创建一个包含单个服务的应用。每个服务：

对应单个 Unix 域套接字
可以承载一种或多种组件类型

注意

单个服务只能注册每种类型的一个组件。但是，相同类型的多个组件可以分布在多个服务中。

实现和注册组件

本地测试组件

可以通过在命令行启动应用并配置 Dapr 边车来使用它，从而测试可插拔组件。

要启动组件，在应用目录中：

dotnet run

要配置 Dapr 使用该组件，在资源路径目录中：

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: <component name>
spec:
  type: state.<socket name>
  version: v1
  metadata:
  - name: key1
    value: value1
  - name: key2
    value: value2

当组件实例化时，任何 metadata 属性都将通过其 IPluggableComponent.InitAsync() 方法传递给组件。

要启动 Dapr（以及可选的，使用该服务的服务）：

dapr run --app-id <app id> --resources-path <resources path> ...

此时，Dapr 边车将启动并通过 Unix 域套接字连接到组件。然后您可以通过以下任一方式与组件交互：

通过使用该组件的服务（如果已启动），或
直接使用 Dapr HTTP 或 gRPC API

创建容器

有多种方法可以为您的组件创建容器以进行最终部署。

使用 .NET SDK

.NET 7 及更高版本的 SDK 使您能够在不使用 Dockerfile 的情况下为应用创建基于 .NET 的容器，即使是针对早期版本的 .NET SDK 的应用。这可能是目前为组件生成容器的最简单方法。

注意

目前，.NET 7 SDK 需要本地计算机上的 Docker Desktop、一个特殊的 NuGet 包以及本地计算机上的 Docker Desktop 来构建容器。.NET SDK 的未来版本计划消除这些要求。

本地计算机上可以同时安装多个版本的 .NET SDK。

将 Microsoft.NET.Build.Containers NuGet 包添加到组件项目。

dotnet add package Microsoft.NET.Build.Containers

将应用发布为容器：

dotnet publish --os linux --arch x64 /t:PublishContainer -c Release

注意

确保架构参数 --arch x64 与组件的最终部署目标相匹配。默认情况下，生成的容器架构与本地计算机的架构相匹配。例如，如果本地计算机基于 ARM64（例如 M1 或 M2 Mac）并且省略了该参数，则会生成 ARM64 容器，该容器可能与期望 AMD64 容器的部署目标不兼容。

有关更多配置选项，例如控制容器名称、标签和基础镜像，请参阅 .NET 发布为容器指南。

使用 Dockerfile

虽然有工具可以为 .NET 应用生成 Dockerfile，但 .NET SDK 本身不会。典型的 Dockerfile 可能如下所示：

FROM mcr.microsoft.com/dotnet/aspnet:<runtime> AS base
WORKDIR /app

# Creates a non-root user with an explicit UID and adds permission to access the /app folder
# For more info, please refer to https://aka.ms/vscode-docker-dotnet-configure-containers
RUN adduser -u 5678 --disabled-password --gecos "" appuser && chown -R appuser /app
USER appuser

FROM mcr.microsoft.com/dotnet/sdk:<runtime> AS build
WORKDIR /src
COPY ["<application>.csproj", "<application folder>/"]
RUN dotnet restore "<application folder>/<application>.csproj"
COPY . .
WORKDIR "/src/<application folder>"
RUN dotnet build "<application>.csproj" -c Release -o /app/build

FROM build AS publish
RUN dotnet publish "<application>.csproj" -c Release -o /app/publish /p:UseAppHost=false

FROM base AS final
WORKDIR /app
COPY --from=publish /app/publish .
ENTRYPOINT ["dotnet", "<application>.dll"]

构建镜像：

docker build -f Dockerfile -t <image name>:<tag> .

注意

Dockerfile 中 COPY 操作的路径相对于构建镜像时传递的 Docker 上下文，而 Docker 上下文本身会根据所构建项目的需求而变化（例如，如果它有引用的项目）。在上面的示例中，假设 Docker 上下文是组件项目目录。

演示

观看此视频，了解使用 .NET 构建可插拔组件的演示：

后续步骤

了解可插拔组件 .NET SDK 的高级步骤
了解有关使用可插拔组件 .NET SDK 的更多信息：

1.1 - 实现 .NET 输入/输出绑定组件

如何使用 Dapr 可插拔组件 .NET SDK 创建输入/输出绑定

创建绑定组件只需要几个基本步骤。

添加绑定命名空间

添加绑定相关命名空间的 using 语句。

using Dapr.PluggableComponents.Components;
using Dapr.PluggableComponents.Components.Bindings;

输入绑定：实现 `IInputBinding`

创建一个实现 IInputBinding 接口的类。

internal sealed class MyBinding : IInputBinding
{
    public Task InitAsync(MetadataRequest request, CancellationToken cancellationToken = default)
    {
        // 使用配置的元数据初始化组件时调用...
    }

    public async Task ReadAsync(MessageDeliveryHandler<InputBindingReadRequest, InputBindingReadResponse> deliveryHandler, CancellationToken cancellationToken = default)
    {
        // 直到被取消之前，检查底层存储中的消息并将其传递给 Dapr 运行时...
    }
}

对 ReadAsync() 方法的调用是"长期运行"的，因为该方法预期在取消之前不会返回（例如，通过 cancellationToken）。当从组件的底层存储读取消息时，这些消息通过 deliveryHandler 回调传递给 Dapr 运行时。传递操作允许组件在应用程序（由 Dapr 运行时服务）确认处理消息时接收通知。

    public async Task ReadAsync(MessageDeliveryHandler<InputBindingReadRequest, InputBindingReadResponse> deliveryHandler, CancellationToken cancellationToken = default)
    {
        TimeSpan pollInterval = // 轮询间隔（例如，来自初始化元数据）...

        // 轮询底层存储直到被取消...
        while (!cancellationToken.IsCancellationRequested)
        {
            var messages = // 从底层存储轮询消息...

            foreach (var message in messages)
            {
                // 将消息传递给 Dapr 运行时...
                await deliveryHandler(
                    new InputBindingReadResponse
                    {
                        // 设置消息内容...
                    },
                    // 当应用程序确认消息时调用的回调...
                    async request =>
                    {
                        // 处理响应数据或错误消息...
                    })
            }

            // 等待下一次轮询（或取消）...
            await Task.Delay(pollInterval, cancellationToken);
        }
    }

输出绑定：实现 `IOutputBinding`

创建一个实现 IOutputBinding 接口的类。

internal sealed class MyBinding : IOutputBinding
{
    public Task InitAsync(MetadataRequest request, CancellationToken cancellationToken = default)
    {
        // 使用配置的元数据初始化组件时调用...
    }

    public Task<OutputBindingInvokeResponse> InvokeAsync(OutputBindingInvokeRequest request, CancellationToken cancellationToken = default)
    {
        // 调用以执行特定操作...
    }

    public Task<string[]> ListOperationsAsync(CancellationToken cancellationToken = default)
    {
        // 调用以列出可执行的操作。
    }
}

输入和输出绑定组件

组件可以同时是输入和输出绑定，只需实现这两个接口即可。

internal sealed class MyBinding : IInputBinding, IOutputBinding
{
    // IInputBinding 实现...

    // IOutputBinding 实现...
}

注册绑定组件

在主程序文件（例如 Program.cs）中，在应用程序服务中注册绑定组件。

using Dapr.PluggableComponents;

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "<socket name>",
    serviceBuilder =>
    {
        serviceBuilder.RegisterBinding<MyBinding>();
    });

app.Run();

注意

同时实现 IInputBinding 和 IOutputBinding 的组件将被同时注册为输入和输出绑定。

后续步骤

了解可插拔组件 .NET SDK 的高级步骤
了解有关使用可插拔组件 .NET SDK 的更多信息：
- 发布订阅
- 状态存储

1.2 - 实现 .NET 发布订阅组件

如何使用 Dapr 可插拔组件 .NET SDK 创建发布订阅

创建发布订阅组件只需几个基本步骤。

添加发布订阅命名空间

为发布订阅相关的命名空间添加 using 语句。

using Dapr.PluggableComponents.Components;
using Dapr.PluggableComponents.Components.PubSub;

实现 `IPubSub`

创建一个实现 IPubSub 接口的类。

internal sealed class MyPubSub : IPubSub
{
    public Task InitAsync(MetadataRequest request, CancellationToken cancellationToken = default)
    {
        // 调用以使用配置的元数据初始化组件...
    }

    public Task PublishAsync(PubSubPublishRequest request, CancellationToken cancellationToken = default)
    {
        // 将消息发送到"topic"...
    }

    public Task PullMessagesAsync(PubSubPullMessagesTopic topic, MessageDeliveryHandler<string?, PubSubPullMessagesResponse> deliveryHandler, CancellationToken cancellationToken = default)
    {
        // 直到取消之前，检查 topic 中的消息并将其传递给 Dapr runtime...
    }
}

对 PullMessagesAsync() 方法的调用是"长期存在"的，也就是说该方法在取消之前（例如通过 cancellationToken）不会返回。应从中拉取消息的"topic"通过 topic 参数传递，而向 Dapr runtime 的传递则通过 deliveryHandler 回调执行。传递允许组件在应用程序（由 Dapr runtime 提供服务）确认已处理消息时接收通知。

    public async Task PullMessagesAsync(PubSubPullMessagesTopic topic, MessageDeliveryHandler<string?, PubSubPullMessagesResponse> deliveryHandler, CancellationToken cancellationToken = default)
    {
        TimeSpan pollInterval = // 轮询间隔（例如来自初始化元数据）...

        // 轮询 topic 直到取消...
        while (!cancellationToken.IsCancellationRequested)
        {
            var messages = // 从 topic 轮询消息...

            foreach (var message in messages)
            {
                // 将消息传递给 Dapr runtime...
                await deliveryHandler(
                    new PubSubPullMessagesResponse(topicName)
                    {
                        // 设置消息内容...
                    },
                    // 当应用程序确认消息时调用的回调...
                    async errorMessage =>
                    {
                        // 空消息表示应用程序成功处理了消息...
                        if (String.IsNullOrEmpty(errorMessage))
                        {
                            // 从 topic 中删除消息...
                        }
                    })
            }

            // 等待下一次轮询（或取消）...
            await Task.Delay(pollInterval, cancellationToken);
        }
    }

注册发布订阅组件

在主程序文件（例如 Program.cs）中，向应用程序服务注册发布订阅组件。

using Dapr.PluggableComponents;

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "<socket name>",
    serviceBuilder =>
    {
        serviceBuilder.RegisterPubSub<MyPubSub>();
    });

app.Run();

后续步骤

了解可插拔组件 .NET SDK 的高级步骤
了解有关使用可插拔组件 .NET SDK 的更多信息：
- 绑定
- 状态存储

1.3 - 实现 .NET 状态存储组件

如何使用 Dapr 可插拔组件 .NET SDK 创建状态存储

创建状态存储组件只需要几个基本步骤。

添加状态存储命名空间

添加状态存储相关命名空间的 using 语句。

using Dapr.PluggableComponents.Components;
using Dapr.PluggableComponents.Components.StateStore;

实现 `IStateStore`

创建一个实现 IStateStore 接口的类。

internal sealed class MyStateStore : IStateStore
{
    public Task DeleteAsync(StateStoreDeleteRequest request, CancellationToken cancellationToken = default)
    {
        // 从状态存储中删除请求的键...
    }

    public Task<StateStoreGetResponse?> GetAsync(StateStoreGetRequest request, CancellationToken cancellationToken = default)
    {
        // 从状态存储中获取请求的键值，否则返回 null...
    }

    public Task InitAsync(MetadataRequest request, CancellationToken cancellationToken = default)
    {
        // 调用以使用配置的元数据初始化组件...
    }

    public Task SetAsync(StateStoreSetRequest request, CancellationToken cancellationToken = default)
    {
        // 在状态存储中设置请求的键为指定值...
    }
}

注册状态存储组件

在主程序文件（例如 Program.cs）中，向应用程序服务注册状态存储。

using Dapr.PluggableComponents;

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "<socket name>",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<MyStateStore>();
    });

app.Run();

批量状态存储

旨在支持批量操作的状态存储应实现可选的 IBulkStateStore 接口。其方法镜像了基础 IStateStore 接口的方法，但包含多个请求值。

注意

对于未实现 IBulkStateStore 的状态存储，Dapr 运行时将通过单独调用其操作来模拟批量状态存储操作。

internal sealed class MyStateStore : IStateStore, IBulkStateStore
{
    // ...

    public Task BulkDeleteAsync(StateStoreDeleteRequest[] requests, CancellationToken cancellationToken = default)
    {
        // 从状态存储中删除所有请求的值...
    }

    public Task<StateStoreBulkStateItem[]> BulkGetAsync(StateStoreGetRequest[] requests, CancellationToken cancellationToken = default)
    {
        // 从状态存储中返回所有请求的值...
    }

    public Task BulkSetAsync(StateStoreSetRequest[] requests, CancellationToken cancellationToken = default)
    {
        // 在状态存储中设置所有请求的键的值...
    }
}

事务状态存储

旨在支持事务的状态存储应实现可选的 ITransactionalStateStore 接口。其 TransactAsync() 方法接收一个请求，其中包含要在事务中执行的一系列删除和/或设置操作。状态存储应遍历该序列并调用每个操作的 Visit() 方法，传入代表对每种操作类型要执行的操作的回调。

internal sealed class MyStateStore : IStateStore, ITransactionalStateStore
{
    // ...

    public async Task TransactAsync(StateStoreTransactRequest request, CancellationToken cancellationToken = default)
    {
        // 开始事务...

        try
        {
            foreach (var operation in request.Operations)
            {
                await operation.Visit(
                    async deleteRequest =>
                    {
                        // 处理删除请求...

                    },
                    async setRequest =>
                    {
                        // 处理设置请求...
                    });
            }
        }
        catch
        {
            // 回滚事务...

            throw;
        }

        // 提交事务...
    }
}

可查询状态存储

旨在支持查询的状态存储应实现可选的 IQueryableStateStore 接口。其 QueryAsync() 方法接收有关查询的详细信息，例如筛选器、结果限制和分页，以及结果的排序顺序。状态存储应使用这些详细信息生成一组值作为其响应的一部分返回。

internal sealed class MyStateStore : IStateStore, IQueryableStateStore
{
    // ...

    public Task<StateStoreQueryResponse> QueryAsync(StateStoreQueryRequest request, CancellationToken cancellationToken = default)
    {
        // 生成并返回结果...
    }
}

ETag 和其他语义错误处理

Dapr 运行时对某些状态存储操作导致的某些错误条件有额外的处理。状态存储可以通过从其操作逻辑中抛出特定异常来指示此类情况：

异常	适用操作	描述
`ETagInvalidException`	Delete、Set、Bulk Delete、Bulk Set	当 ETag 无效时
`ETagMismatchException`	Delete、Set、Bulk Delete、Bulk Set	当 ETag 与预期值不匹配时
`BulkDeleteRowMismatchException`	Bulk Delete	当受影响的行数与预期行数不匹配时

后续步骤

了解可插拔组件 .NET SDK 的高级步骤
了解有关使用可插拔组件 .NET SDK 的更多信息：
- 绑定
- 发布订阅

1.4 - Dapr 可插拔组件 .NET SDK 的高级用法

如何使用 Dapr 可插拔组件 .NET SDK 的高级技术

尽管大多数人通常不需要，但这些指南展示了配置 .NET 可插拔组件的高级方法。

1.4.1 - .NET Dapr 可插拔组件中的多个服务

如何从 .NET 可插拔组件公开多个服务

可插拔组件可以托管多种类型的多个组件。您可能需要这样做：

为了最小化集群中运行的边车数量
为了分组可能共享库和实现的相关组件，例如：
- 一个既作为通用状态存储公开的数据库，和
- 允许更特定操作的输出绑定。

每个 Unix 域套接字可以管理对每种类型的一个组件的调用。要托管相同类型的多个组件，您可以将这些类型分散到多个套接字上。SDK 将每个套接字绑定到一个"服务"，每个服务由一种或多种组件类型组成。

注册多个服务

每次调用 RegisterService() 都会将一个套接字绑定到一组已注册的组件，其中每个服务可以注册每种类型的组件中的一个。

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "service-a",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<MyDatabaseStateStore>();
        serviceBuilder.RegisterBinding<MyDatabaseOutputBinding>();
    });

app.RegisterService(
    "service-b",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<AnotherStateStore>();
    });

app.Run();

class MyDatabaseStateStore : IStateStore
{
    // ...
}

class MyDatabaseOutputBinding : IOutputBinding
{
    // ...
}

class AnotherStateStore : IStateStore
{
    // ...
}

配置多个组件

配置 Dapr 以使用托管组件与任何单个组件相同 - 组件 YAML 引用关联的套接字。

#
# 此组件使用与套接字 `state-store-a` 关联的状态存储
#
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: state-store-a
spec:
  type: state.service-a
  version: v1
  metadata: []

#
# 此组件使用与套接字 `state-store-b` 关联的状态存储
#
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: state-store-b
spec:
  type: state.service-b
  version: v1
  metadata: []

后续步骤

1.4.2 - .NET Dapr 可插拔组件的应用环境

如何配置 .NET 可插拔组件的环境

.NET Dapr 可插拔组件应用可以像 ASP.NET 应用一样配置依赖注入、日志和配置值。DaprPluggableComponentsApplication 暴露了一组与 WebApplicationBuilder 相似的配置属性。

依赖注入

注册到服务的组件可以参与依赖注入。在创建组件时，组件构造函数中的参数将被注入，前提是这些类型已在应用中注册。你可以通过 DaprPluggableComponentsApplication 暴露的 IServiceCollection 注册它们。

var app = DaprPluggableComponentsApplication.Create();

// 将 MyService 注册为 IService 的单例实现。
app.Services.AddSingleton<IService, MyService>();

app.RegisterService(
    "<service name>",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<MyStateStore>();
    });

app.Run();

interface IService
{
    // ...
}

class MyService : IService
{
    // ...
}

class MyStateStore : IStateStore
{
    // 在创建状态存储时注入 IService。
    public MyStateStore(IService service)
    {
        // ...
    }

    // ...
}

警告

不建议使用 IServiceCollection.AddScoped()。此类实例的生命周期绑定到单个 gRPC 方法调用，这与单个组件实例的生命周期不匹配。

日志

.NET Dapr 可插拔组件可以使用标准 .NET 日志机制。DaprPluggableComponentsApplication 暴露了一个 ILoggingBuilder，可以通过它进行配置。

注意

与 ASP.NET 一样，日志记录器服务（例如 ILogger<T>）已预先注册。

var app = DaprPluggableComponentsApplication.Create();

// 清除默认日志记录器并设置新的日志记录器。
app.Logging.ClearProviders();
app.Logging.AddConsole();

app.RegisterService(
    "<service name>",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<MyStateStore>();
    });

app.Run();

class MyStateStore : IStateStore
{
    // 在创建状态存储时注入日志记录器。
    public MyStateStore(ILogger<MyStateStore> logger)
    {
        // ...
    }

    // ...
}

配置值

由于 .NET 可插拔组件基于 ASP.NET 构建，它们可以使用其标准配置机制，并默认使用同一组预先注册的提供程序。DaprPluggableComponentsApplication 暴露了一个 IConfigurationManager，可以通过它进行配置。

var app = DaprPluggableComponentsApplication.Create();

// 清除默认配置提供程序并添加新的提供程序。
((IConfigurationBuilder)app.Configuration).Sources.Clear();
app.Configuration.AddEnvironmentVariables();

// 在启动时获取配置值。
const value = app.Configuration["<name>"];

app.RegisterService(
    "<service name>",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<MyStateStore>();
    });

app.Run();

class MyStateStore : IStateStore
{
    // 在创建状态存储时注入配置。
    public MyStateStore(IConfiguration configuration)
    {
        // ...
    }

    // ...
}

后续步骤

1.4.3 - .NET Dapr 可插拔组件的生命周期

如何控制 .NET 可插拔组件的生命周期

有两种方式注册组件：

组件作为单例运行，其生命周期由 SDK 管理
组件的生命周期由可插拔组件决定，可根据需要为多实例或单例

单例组件

_按类型_注册的组件是单例：一个实例将为与该 socket 关联的该类型的所有已配置组件提供服务。当该类型仅存在单个组件且在 Dapr 应用程序之间共享时，此方法最佳。

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "service-a",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore<SingletonStateStore>();
    });

app.Run();

class SingletonStateStore : IStateStore
{
    // ...
}

多实例组件

可以通过传递"工厂方法"来注册组件。对于与该 socket 关联的该类型的每个已配置组件，都会调用此方法。该方法返回要与该组件关联的实例（无论是否共享）。当同一类型的多个组件可能使用不同的元数据集进行配置，或需要将组件操作彼此隔离时，此方法最佳。

工厂方法将接收上下文，例如已配置的 Dapr 组件的 ID，可用于区分组件实例。

var app = DaprPluggableComponentsApplication.Create();

app.RegisterService(
    "service-a",
    serviceBuilder =>
    {
        serviceBuilder.RegisterStateStore(
            context =>
            {
                return new MultiStateStore(context.InstanceId);
            });
    });

app.Run();

class MultiStateStore : IStateStore
{
    private readonly string instanceId;

    public MultiStateStore(string instanceId)
    {
        this.instanceId = instanceId;
    }

    // ...
}

后续步骤

2 - Dapr 可插拔组件 Go SDK 入门

如何开始使用 Dapr 可插拔组件 Go SDK

Dapr 提供了用于帮助开发 Go 可插拔组件的软件包。

前置条件

Go 1.20 或更高版本
Dapr 1.9 CLI 或更高版本
已初始化的 Dapr 环境
Linux、Mac 或 Windows（需使用 WSL）

注意

在 Windows 上开发 Dapr 可插拔组件需要 WSL。并非所有语言和 SDK 在"原生"Windows 上都支持 Unix 域套接字。

应用程序创建

创建可插拔组件首先需要创建一个空的 Go 应用程序。

mkdir example
cd example
go mod init example

导入 Dapr 软件包

导入 Dapr 可插拔组件 SDK 软件包。

go get github.com/dapr-sandbox/components-go-sdk@v0.1.0

创建 main 包

在 main.go 中，导入 Dapr 可插拔组件软件包并运行应用程序。

package main

import (
	dapr "github.com/dapr-sandbox/components-go-sdk"
)

func main() {
	dapr.MustRun()
}

这将创建一个不包含任何组件的应用程序。你需要实现并注册一个或多个组件。

实现并注册组件

注意

单个服务只能注册每种类型的一个组件。但是，可以跨多个服务分发同类型的多个组件。

本地测试组件

创建 Dapr 组件套接字目录

Dapr 通过公共目录中的 Unix 域套接字文件与可插拔组件通信。默认情况下，Dapr 和可插拔组件都使用 /tmp/dapr-components-sockets 目录。如果该目录尚不存在，你应该创建它。

mkdir /tmp/dapr-components-sockets

启动可插拔组件

可以通过在命令行启动应用程序来测试可插拔组件。

要启动组件，在应用程序目录中：

go run main.go

配置 Dapr 以使用可插拔组件

要配置 Dapr 使用该组件，请在 resources 目录中创建一个组件 YAML 文件。例如，对于状态存储组件：

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: <component name>
spec:
  type: state.<socket name>
  version: v1
  metadata:
  - name: key1
    value: value1
  - name: key2
    value: value2

当组件实例化时，任何 metadata 属性都将通过组件的 Store.Init(metadata state.Metadata) 方法传递给组件。

启动 Dapr

要启动 Dapr（以及可选的，使用该服务的服务）：

dapr run --app-id <app id> --resources-path <resources path> ...

此时，Dapr 边车将启动并通过 Unix 域套接字连接到组件。然后你可以通过以下方式与组件交互：

通过使用该组件的服务（如果已启动），或
直接使用 Dapr HTTP 或 gRPC API

创建容器

可插拔组件作为容器部署，作为应用程序的边车运行（就像 Dapr 本身一样）。用于为 Go 应用程序创建 Docker 镜像的典型 Dockerfile 可能如下所示：

FROM golang:1.20-alpine AS builder

WORKDIR /usr/src/app

# 下载依赖
COPY go.mod go.sum ./
RUN go mod download && go mod verify

# 构建应用程序
COPY . .
RUN go build -v -o /usr/src/bin/app .

FROM alpine:latest

# 设置非 root 用户和权限
RUN addgroup -S app && adduser -S app -G app
RUN mkdir /tmp/dapr-components-sockets && chown app /tmp/dapr-components-sockets

# 将应用程序复制到运行时镜像
COPY --from=builder --chown=app /usr/src/bin/app /app

USER app

CMD ["/app"]

构建镜像：

docker build -f Dockerfile -t <image name>:<tag> .

注意

Dockerfile 中 COPY 操作的路径是相对于构建镜像时传递的 Docker 上下文而言的，而 Docker 上下文本身将根据所构建应用程序的需求而变化。在上面的示例中，假设 Docker 上下文是组件应用程序目录。

后续步骤

可插拔组件 Go SDK 的高级技巧
了解有关实现的更多信息：

2.1 - 实现 Go 输入/输出绑定组件

如何使用 Dapr 可插拔组件 Go SDK 创建输入/输出绑定

创建绑定组件只需几个基本步骤。

导入绑定包

创建文件 components/inputbinding.go 并添加与绑定相关包的 import 语句。

package components

import (
	"context"
	"github.com/dapr/components-contrib/bindings"
)

输入绑定：实现 `InputBinding` 接口

创建一个实现 InputBinding 接口的类型。

type MyInputBindingComponent struct {
}

func (component *MyInputBindingComponent) Init(meta bindings.Metadata) error {
	// 调用以使用配置的元数据初始化组件...
}

func (component *MyInputBindingComponent) Read(ctx context.Context, handler bindings.Handler) error {
	// 直到取消为止，检查底层存储中的消息并将其传递给 Dapr 运行时...
}

预期 Read() 方法调用会建立一个用于检索消息的长效机制，但立即返回 nil（或在无法设置该机制时返回错误）。该机制应在取消时结束（例如，通过 ctx.Done() or ctx.Err() != nil）。当从组件的底层存储读取消息时，它们通过 handler 回调传递给 Dapr 运行时，该回调在应用程序（由 Dapr 运行时提供服务）确认消息处理完成之前不会返回。

func (b *MyInputBindingComponent) Read(ctx context.Context, handler bindings.Handler) error {
	go func() {
		for {
			err := ctx.Err()

			if err != nil {
				return
			}
	
			messages := // 轮询消息...

            for _, message := range messages {
                handler(ctx, &bindings.ReadResponse{
                    // 设置消息内容...
                })
            }

			select {
				case <-ctx.Done():
				case <-time.After(5 * time.Second):
			} 
		}
	}()

	return nil
}

输出绑定：实现 `OutputBinding` 接口

创建一个实现 OutputBinding 接口的类型。

type MyOutputBindingComponent struct {
}

func (component *MyOutputBindingComponent) Init(meta bindings.Metadata) error {
	// 调用以使用配置的元数据初始化组件...
}

func (component *MyOutputBindingComponent) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
	// 调用以调用特定操作...
}

func (component *MyOutputBindingComponent) Operations() []bindings.OperationKind {
	// 调用以列出可被调用的操作。
}

输入和输出绑定组件

组件可以同时既是输入绑定又是输出绑定。只需实现两个接口并将组件注册为两种绑定类型。

注册绑定组件

在主应用程序文件（例如 main.go）中，向应用程序注册绑定组件。

package main

import (
	"example/components"
	dapr "github.com/dapr-sandbox/components-go-sdk"
	"github.com/dapr-sandbox/components-go-sdk/bindings/v1"
)

func main() {
	// 注册输入绑定...
	dapr.Register("my-inputbinding", dapr.WithInputBinding(func() bindings.InputBinding {
		return &components.MyInputBindingComponent{}
	}))

	// 注册输出绑定...
	dapr.Register("my-outputbinding", dapr.WithOutputBinding(func() bindings.OutputBinding {
		return &components.MyOutputBindingComponent{}
	}))

	dapr.MustRun()
}

后续步骤

可插拔组件 Go SDK 的高级技巧
了解更多关于实现：
- 状态存储
- 发布订阅

2.2 - 实现 Go 发布订阅组件

如何使用 Dapr 可插拔组件 Go SDK 创建发布订阅组件

创建发布订阅组件只需要几个基本步骤。

导入发布订阅包

创建文件 components/pubsub.go 并添加发布订阅相关包的 import 语句。

package components

import (
	"context"
	"github.com/dapr/components-contrib/pubsub"
)

实现 `PubSub` 接口

创建一个实现 PubSub 接口的类型。

type MyPubSubComponent struct {
}

func (component *MyPubSubComponent) Init(metadata pubsub.Metadata) error {
	// 调用此方法以使用配置的元数据初始化组件...
}

func (component *MyPubSubComponent) Close() error {
    // 不用于可插拔组件...
	return nil
}

func (component *MyPubSubComponent) Features() []pubsub.Feature {
	// 返回组件支持的功能列表...
}

func (component *MyPubSubComponent) Publish(req *pubsub.PublishRequest) error {
	// 将消息发送到 "topic"...
}

func (component *MyPubSubComponent) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, handler pubsub.Handler) error {
	// 在取消之前，持续检查 topic 是否有消息并将其传递给 Dapr 运行时...
}

对 Subscribe() 方法的调用预期会建立一个用于检索消息的长效机制，但立即返回 nil（或错误，如果无法建立该机制）。该机制应在取消时结束（例如，通过 ctx.Done() 或 ctx.Err() != nil）。应从中拉取消息的 “topic” 通过 req 参数传递，而传递给 Dapr 运行时则通过 handler 回调执行。回调在应用程序（由 Dapr 运行时提供服务）确认消息处理后才会返回。

func (component *MyPubSubComponent) Subscribe(ctx context.Context, req pubsub.SubscribeRequest, handler pubsub.Handler) error {
	go func() {
		for {
			err := ctx.Err()

			if err != nil {
				return
			}
	
			messages := // 轮询消息...

            for _, message := range messages {
                handler(ctx, &pubsub.NewMessage{
                    // 设置消息内容...
                })
            }

			select {
				case <-ctx.Done():
				case <-time.After(5 * time.Second):
			} 
		}
	}()

	return nil
}

注册发布订阅组件

在主应用程序文件（例如 main.go）中，向应用程序注册发布订阅组件。

package main

import (
	"example/components"
	dapr "github.com/dapr-sandbox/components-go-sdk"
	"github.com/dapr-sandbox/components-go-sdk/pubsub/v1"
)

func main() {
	dapr.Register("<socket name>", dapr.WithPubSub(func() pubsub.PubSub {
		return &components.MyPubSubComponent{}
	}))

	dapr.MustRun()
}

后续步骤

可插拔组件 Go SDK 的高级技巧
了解有关实现的更多信息：
- 绑定
- 状态

2.3 - 实现 Go 状态存储组件

如何使用 Dapr 可插拔组件 Go SDK 创建状态存储

创建状态存储组件只需要几个基本步骤。

导入状态存储包

创建文件 components/statestore.go 并添加状态存储相关包的 import 语句。

package components

import (
	"context"
	"github.com/dapr/components-contrib/state"
)

实现 `Store` 接口

创建一个实现 Store 接口的类型。

type MyStateStore struct {
}

func (store *MyStateStore) Init(metadata state.Metadata) error {
	// 使用配置的元数据初始化组件时调用...
}

func (store *MyStateStore) GetComponentMetadata() map[string]string {
    // 可插拔组件不使用此方法...
	return map[string]string{}
}

func (store *MyStateStore) Features() []state.Feature {
	// 返回状态存储支持的功能列表...
}

func (store *MyStateStore) Delete(ctx context.Context, req *state.DeleteRequest) error {
	// 从状态存储中删除请求的键...
}

func (store *MyStateStore) Get(ctx context.Context, req *state.GetRequest) (*state.GetResponse, error) {
	// 从状态存储中获取请求的键值，否则返回空响应...
}

func (store *MyStateStore) Set(ctx context.Context, req *state.SetRequest) error {
	// 在状态存储中将请求的键设置为指定值...
}

func (store *MyStateStore) BulkGet(ctx context.Context, req []state.GetRequest) (bool, []state.BulkGetResponse, error) {
	// 从状态存储中获取请求的键值...
}

func (store *MyStateStore) BulkDelete(ctx context.Context, req []state.DeleteRequest) error {
	// 从状态存储中删除请求的键...
}

func (store *MyStateStore) BulkSet(ctx context.Context, req []state.SetRequest) error {
	// 在状态存储中将请求的键设置为指定的值...
}

注册状态存储组件

在主应用程序文件（例如 main.go）中，将状态存储注册到应用程序服务。

package main

import (
	"example/components"
	dapr "github.com/dapr-sandbox/components-go-sdk"
	"github.com/dapr-sandbox/components-go-sdk/state/v1"
)

func main() {
	dapr.Register("<socket name>", dapr.WithStateStore(func() state.Store {
		return &components.MyStateStoreComponent{}
	}))

	dapr.MustRun()
}

批量状态存储

虽然状态存储需要支持批量操作，但其实现会顺序委托给单个操作方法。

事务性状态存储

支持事务的状态存储应该实现可选的 TransactionalStore 接口。其 Multi() 方法接收一个包含要在事务中执行的 delete 和/或 set 操作序列的请求。状态存储应遍历该序列并应用每个操作。

func (store *MyStateStoreComponent) Multi(ctx context.Context, request *state.TransactionalStateRequest) error {
    // 开始事务...

    for _, operation := range request.Operations {
		switch operation.Operation {
		case state.Delete:
			deleteRequest := operation.Request.(state.DeleteRequest)
			// 处理删除请求...
		case state.Upsert:
			setRequest := operation.Request.(state.SetRequest)
			// 处理设置请求...
		}
	}

    // 结束（或回滚）事务...

	return nil
}

可查询状态存储

支持查询的状态存储应该实现可选的 Querier 接口。其 Query() 方法接收有关查询的详细信息，例如过滤器、结果限制、分页和结果的排序顺序。状态存储使用这些详细信息生成一组值作为其响应的一部分返回。

func (store *MyStateStoreComponent) Query(ctx context.Context, req *state.QueryRequest) (*state.QueryResponse, error) {
	// 生成并返回结果...
}

ETag 和其他语义错误处理

Dapr 运行时对某些状态存储操作导致的某些错误条件有额外的处理。状态存储可以通过从其操作逻辑返回特定错误来指示此类条件：

错误	适用操作	描述
`NewETagError(state.ETagInvalid, ...)`	Delete、Set、Bulk Delete、Bulk Set	当 ETag 无效时
`NewETagError(state.ETagMismatch, ...)`	Delete、Set、Bulk Delete、Bulk Set	当 ETag 与预期值不匹配时
`NewBulkDeleteRowMismatchError(...)`	Bulk Delete	当受影响的行数与预期行数不匹配时

后续步骤

可插拔组件 Go SDK 的高级技术
了解有关实现的更多信息：
- 绑定
- 发布订阅

2.4 - Dapr 可插拔组件 Go SDK 的高级用法

如何使用 Dapr 可插拔组件 Go SDK 的高级技术

虽然大多数人通常不需要，但这些指南展示了配置 Go 可插拔组件的高级方法。

组件生命周期

可插拔组件通过传递一个"工厂方法"来注册，该方法会为与该 socket 关联的该类型的每个已配置 Dapr 组件调用。该方法返回与该 Dapr 组件关联的实例（无论是否共享）。这允许多个相同类型的 Dapr 组件使用不同的元数据集进行配置，当组件操作需要相互隔离时等。

注册多个服务

每次调用 Register() 会将一个 socket 绑定到一个注册的可插拔组件。每个 socket 可以注册每种组件类型中的一个（输入/输出绑定、发布订阅和状态存储）。

func main() {
	dapr.Register("service-a", dapr.WithStateStore(func() state.Store {
		return &components.MyDatabaseStoreComponent{}
	}))

	dapr.Register("service-a", dapr.WithOutputBinding(func() bindings.OutputBinding {
		return &components.MyDatabaseOutputBindingComponent{}
	}))

	dapr.Register("service-b", dapr.WithStateStore(func() state.Store {
		return &components.MyDatabaseStoreComponent{}
	}))

	dapr.MustRun()
}

在上面的示例中，一个状态存储和输出绑定注册到 socket service-a，而另一个状态存储注册到 socket service-b。

配置多个组件

配置 Dapr 使用托管组件与配置任何单个组件相同 — 组件 YAML 引用关联的 socket。例如，要为上面注册的两个组件（到 socket service-a 和 service-b）配置 Dapr 状态存储，您需要创建两个配置文件，每个文件引用其各自的 socket。

#
# 此组件使用与 socket `service-a` 关联的状态存储
#
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: state-store-a
spec:
  type: state.service-a
  version: v1
  metadata: []

#
# 此组件使用与 socket `service-b` 关联的状态存储
#
apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: state-store-b
spec:
  type: state.service-b
  version: v1
  metadata: []

后续步骤

了解有关实现的更多信息：

可插拔组件 SDK

可插拔组件 SDK

1 - Dapr 可插拔组件 .NET SDK 入门

前置条件

注意

创建项目

添加 NuGet 包

创建应用和服务

注意

实现和注册组件

本地测试组件

创建容器

使用 .NET SDK

注意

注意

使用 Dockerfile

注意

演示

后续步骤

1.1 - 实现 .NET 输入/输出绑定组件

添加绑定命名空间

输入绑定：实现 IInputBinding

输出绑定：实现 IOutputBinding

输入和输出绑定组件

注册绑定组件

注意

后续步骤

1.2 - 实现 .NET 发布订阅组件

添加发布订阅命名空间

实现 IPubSub

注册发布订阅组件

后续步骤

1.3 - 实现 .NET 状态存储组件

添加状态存储命名空间

实现 IStateStore

注册状态存储组件

批量状态存储

注意

事务状态存储

可查询状态存储

ETag 和其他语义错误处理

后续步骤

1.4 - Dapr 可插拔组件 .NET SDK 的高级用法

1.4.1 - .NET Dapr 可插拔组件中的多个服务

注册多个服务

配置多个组件

后续步骤

1.4.2 - .NET Dapr 可插拔组件的应用环境

依赖注入

警告

日志

注意

配置值

后续步骤

1.4.3 - .NET Dapr 可插拔组件的生命周期

单例组件

多实例组件

后续步骤

2 - Dapr 可插拔组件 Go SDK 入门

前置条件

注意

应用程序创建

导入 Dapr 软件包

创建 main 包

实现并注册组件

注意

本地测试组件

创建 Dapr 组件套接字目录

启动可插拔组件

配置 Dapr 以使用可插拔组件

启动 Dapr

创建容器

注意

后续步骤

2.1 - 实现 Go 输入/输出绑定组件

导入绑定包

输入绑定：实现 InputBinding 接口

输出绑定：实现 OutputBinding 接口

输入和输出绑定组件

注册绑定组件

输入绑定：实现 `IInputBinding`

输出绑定：实现 `IOutputBinding`

实现 `IPubSub`

实现 `IStateStore`

输入绑定：实现 `InputBinding` 接口

输出绑定：实现 `OutputBinding` 接口

实现 `PubSub` 接口

实现 `Store` 接口