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

Return to the regular view of this page.

Dapr 可插拔组件 Go SDK 入门

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

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

2: 实现 Go 发布订阅组件

3: 实现 Go 状态存储组件

4: 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 的高级技巧
了解有关实现的更多信息：

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 - 实现 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 的高级技巧
了解有关实现的更多信息：
- 绑定
- 状态

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 的高级技术
了解有关实现的更多信息：
- 绑定
- 发布订阅

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: []

后续步骤

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

Dapr 可插拔组件 Go SDK 入门

前置条件

注意

应用程序创建

导入 Dapr 软件包

创建 main 包

实现并注册组件

注意

本地测试组件

创建 Dapr 组件套接字目录

启动可插拔组件

配置 Dapr 以使用可插拔组件

启动 Dapr

创建容器

注意

后续步骤

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

导入绑定包

输入绑定：实现 InputBinding 接口

输出绑定：实现 OutputBinding 接口

输入和输出绑定组件

注册绑定组件

后续步骤

2 - 实现 Go 发布订阅组件

导入发布订阅包

实现 PubSub 接口

注册发布订阅组件

后续步骤

3 - 实现 Go 状态存储组件

导入状态存储包

实现 Store 接口

注册状态存储组件

批量状态存储

事务性状态存储

可查询状态存储

ETag 和其他语义错误处理

后续步骤

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

组件生命周期

注册多个服务

配置多个组件

后续步骤

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

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

实现 `PubSub` 接口

实现 `Store` 接口