如何：实现可插拔组件

学习如何编写和实现可插拔组件

在本指南中，您将学习实现可插拔组件的原因和方法。要了解如何配置和注册可插拔组件，请参阅如何：注册可插拔组件。

实现可插拔组件

要实现可插拔组件，需在组件中实现 gRPC 服务。实现 gRPC 服务需要三个步骤：

找到 proto 定义文件

每个支持的服务接口（如状态存储、发布订阅、绑定、密钥存储）都提供了 proto 定义。

目前支持以下组件 API：

  • 状态存储
  • 发布订阅
  • 绑定
  • 密钥存储
组件类型gRPC 定义内置参考实现文档
状态存储statestate.protoRedis概念, 如何, API 规范
发布订阅pubsubpubsub.protoRedis概念, 如何, API 规范
绑定bindingsbindings.protoKafka概念, 输入如何, 输出如何, API 规范
密钥存储secretstoressecretstore.protoHashicorp/Vault概念, 如何-secrets, API 规范

以下是可插拔组件状态存储的 gRPC 服务定义片段（[state.proto]）：

// StateStore 服务为状态存储组件提供 gRPC 接口。
service StateStore {
  // 使用给定的元数据初始化状态存储组件。
  rpc Init(InitRequest) returns (InitResponse) {}
  // 返回已实现的状态存储功能列表。
  rpc Features(FeaturesRequest) returns (FeaturesResponse) {}
  // Ping 状态存储。用于活跃性目的。
  rpc Ping(PingRequest) returns (PingResponse) {}
  
  // 从状态存储中删除指定的键。
  rpc Delete(DeleteRequest) returns (DeleteResponse) {}
  // 从给定的键获取数据。
  rpc Get(GetRequest) returns (GetResponse) {}
  // 设置指定键的值。
  rpc Set(SetRequest) returns (SetResponse) {}

  // 一次删除多个键。
  rpc BulkDelete(BulkDeleteRequest) returns (BulkDeleteResponse) {}
  // 一次检索多个键。
  rpc BulkGet(BulkGetRequest) returns (BulkGetResponse) {}
  // 一次设置多个键的值。
  rpc BulkSet(BulkSetRequest) returns (BulkSetResponse) {}
}

StateStore 服务接口总共公开了 9 个方法：

  • 2 个用于初始化和组件能力广告的方法（Init 和 Features）
  • 1 个用于健康或活跃性检查的方法（Ping）
  • 3 个用于 CRUD 的方法（Get、Set、Delete）
  • 3 个用于批量 CRUD 操作的方法（BulkGet、BulkSet、BulkDelete）

创建服务脚手架

使用 protocol buffers 和 gRPC 工具生成服务的脚手架。通过 gRPC 概念文档了解更多关于这些工具的信息。

这些工具生成针对任何 gRPC 支持的语言的代码。此代码作为您的服务器的基础，并提供：

  • 处理客户端调用的功能
  • 基础设施以：
    • 解码传入请求
    • 执行服务方法
    • 编码服务响应

生成的代码是不完整的。它缺少：

  • 您的目标服务定义的方法的具体实现（您可插拔组件的核心）。
  • 关于如何处理 Unix Socket Domain 集成的代码，这是 Dapr 特有的。
  • 处理与下游服务集成的代码。

在下一步中了解更多关于填补这些空白的信息。

定义服务

提供所需服务的具体实现。每个组件都有一个 gRPC 服务定义，用于其核心功能，与核心组件接口相同。例如：

  • 状态存储

    可插拔状态存储必须提供 StateStore 服务接口的实现。

    除了这个核心功能外，一些组件可能还会在其他可选服务下公开功能。例如，您可以通过定义 QueriableStateStore 服务和 TransactionalStateStore 服务的实现来添加额外功能。

  • 发布订阅

    可插拔发布订阅组件只有一个核心服务接口定义 pubsub.proto。它们没有可选的服务接口。

  • 绑定

    可插拔输入和输出绑定在 bindings.proto 上有一个核心服务定义。它们没有可选的服务接口。

  • 密钥存储

    可插拔密钥存储在 secretstore.proto 上有一个核心服务定义。它们没有可选的服务接口。

在使用 gRPC 和 protocol buffers 工具生成上述状态存储示例的服务脚手架代码后，您可以为 service StateStore 下定义的 9 个方法定义具体实现，以及初始化和与依赖项通信的代码。

这个具体实现和辅助代码是您可插拔组件的核心。它们定义了您的组件在处理来自 Dapr 的 gRPC 请求时的行为。

返回语义错误

返回语义错误也是可插拔组件协议的一部分。组件必须返回对用户应用程序具有语义意义的特定 gRPC 代码，这些错误用于从并发要求到仅信息的各种情况。

错误gRPC 错误代码源组件描述
ETag 不匹配codes.FailedPrecondition状态存储错误映射以满足并发要求
ETag 无效codes.InvalidArgument状态存储
批量删除行不匹配codes.Internal状态存储

在 状态管理概述中了解更多关于并发要求的信息。

以下示例演示了如何在您自己的可插拔组件中返回错误，并根据需要更改消息。


重要提示： 为了使用 .NET 进行错误映射，首先安装 Google.Api.CommonProtos NuGet 包。

ETag 不匹配

var badRequest = new BadRequest();
var des = "提供的 ETag 字段与存储中的不匹配";
badRequest.FieldViolations.Add(    
   new Google.Rpc.BadRequest.Types.FieldViolation
       {        
         Field = "etag",
         Description = des
       });

var baseStatusCode = Grpc.Core.StatusCode.FailedPrecondition;
var status = new Google.Rpc.Status{    
   Code = (int)baseStatusCode
};

status.Details.Add(Google.Protobuf.WellKnownTypes.Any.Pack(badRequest));

var metadata = new Metadata();
metadata.Add("grpc-status-details-bin", status.ToByteArray());
throw new RpcException(new Grpc.Core.Status(baseStatusCode, "fake-err-msg"), metadata);

ETag 无效

var badRequest = new BadRequest();
var des = "ETag 字段只能包含字母数字字符";
badRequest.FieldViolations.Add(
   new Google.Rpc.BadRequest.Types.FieldViolation
   {
      Field = "etag",
      Description = des
   });

var baseStatusCode = Grpc.Core.StatusCode.InvalidArgument;
var status = new Google.Rpc.Status
{
   Code = (int)baseStatusCode
};

status.Details.Add(Google.Protobuf.WellKnownTypes.Any.Pack(badRequest));

var metadata = new Metadata();
metadata.Add("grpc-status-details-bin", status.ToByteArray());
throw new RpcException(new Grpc.Core.Status(baseStatusCode, "fake-err-msg"), metadata);

批量删除行不匹配

var errorInfo = new Google.Rpc.ErrorInfo();

errorInfo.Metadata.Add("expected", "100");
errorInfo.Metadata.Add("affected", "99");

var baseStatusCode = Grpc.Core.StatusCode.Internal;
var status = new Google.Rpc.Status{
    Code = (int)baseStatusCode
};

status.Details.Add(Google.Protobuf.WellKnownTypes.Any.Pack(errorInfo));

var metadata = new Metadata();
metadata.Add("grpc-status-details-bin", status.ToByteArray());
throw new RpcException(new Grpc.Core.Status(baseStatusCode, "fake-err-msg"), metadata);

就像 Dapr Java SDK 一样，Java 可插拔组件 SDK 使用 Project Reactor，它为 Java 提供了异步 API。

错误可以通过以下方式直接返回：

  1. 在您的方法返回的 Mono 或 Flux 中调用 .error() 方法
  2. 提供适当的异常作为参数。

您也可以引发异常，只要它被捕获并反馈到您结果的 Mono 或 Flux 中。

ETag 不匹配

final Status status = Status.newBuilder()
    .setCode(io.grpc.Status.Code.FAILED_PRECONDITION.value())
    .setMessage("fake-err-msg-for-etag-mismatch")
    .addDetails(Any.pack(BadRequest.FieldViolation.newBuilder()
        .setField("etag")
        .setDescription("提供的 ETag 字段与存储中的不匹配")
        .build()))
    .build();
return Mono.error(StatusProto.toStatusException(status));

ETag 无效

final Status status = Status.newBuilder()
    .setCode(io.grpc.Status.Code.INVALID_ARGUMENT.value())
    .setMessage("fake-err-msg-for-invalid-etag")
    .addDetails(Any.pack(BadRequest.FieldViolation.newBuilder()
        .setField("etag")
        .setDescription("ETag 字段只能包含字母数字字符")
        .build()))
    .build();
return Mono.error(StatusProto.toStatusException(status));

批量删除行不匹配

final Status status = Status.newBuilder()
    .setCode(io.grpc.Status.Code.INTERNAL.value())
    .setMessage("fake-err-msg-for-bulk-delete-row-mismatch")
    .addDetails(Any.pack(ErrorInfo.newBuilder()
        .putAllMetadata(Map.ofEntries(
            Map.entry("affected", "99"),
            Map.entry("expected", "100")
        ))
        .build()))
    .build();
return Mono.error(StatusProto.toStatusException(status));

ETag 不匹配

st := status.New(codes.FailedPrecondition, "fake-err-msg")
desc := "提供的 ETag 字段与存储中的不匹配"
v := &errdetails.BadRequest_FieldViolation{
	Field:       etagField,
	Description: desc,
}
br := &errdetails.BadRequest{}
br.FieldViolations = append(br.FieldViolations, v)
st, err := st.WithDetails(br)

ETag 无效

st := status.New(codes.InvalidArgument, "fake-err-msg")
desc := "ETag 字段只能包含字母数字字符"
v := &errdetails.BadRequest_FieldViolation{
	Field:       etagField,
	Description: desc,
}
br := &errdetails.BadRequest{}
br.FieldViolations = append(br.FieldViolations, v)
st, err := st.WithDetails(br)

批量删除行不匹配

st := status.New(codes.Internal, "fake-err-msg")
br := &errdetails.ErrorInfo{}
br.Metadata = map[string]string{
	affected: "99",
	expected: "100",
}
st, err := st.WithDetails(br)

下一步