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

Return to the regular view of this page.

JavaScript SDK

用于开发Dapr应用的JavaScript SDK

1: JavaScript 客户端 SDK

2: JavaScript 服务器 SDK

3: JavaScript SDK for Actors

4: JavaScript SDK中的日志记录

5: JavaScript 示例

6: 如何：在 JavaScript SDK 中编写和管理 Dapr 工作流

这是一个用于在JavaScript和TypeScript中构建Dapr应用的开发库。该库对Dapr的常用API进行了抽象，如服务调用、状态管理、发布订阅、密钥管理等，并提供了一个简单直观的API接口来帮助构建应用。

安装

要开始使用JavaScript SDK，请从NPM安装Dapr JavaScript SDK：

npm install --save @dapr/dapr

结构

Dapr JavaScript SDK包含两个主要组件：

DaprServer：用于管理Dapr sidecar与应用之间的通信。
DaprClient：用于管理应用与Dapr sidecar之间的通信。

这些通信可以配置为使用gRPC或HTTP协议。

入门

为了帮助您快速上手，请查看以下资源：

客户端

创建一个JavaScript客户端，与Dapr sidecar和其他Dapr应用进行交互（例如，发布事件，支持输出绑定等）。

服务器

创建一个JavaScript服务器，让Dapr sidecar与您的应用进行交互（例如，订阅事件，支持输入绑定等）。

虚拟演员

创建具有状态、提醒/计时器和方法的虚拟演员。

日志

配置和自定义SDK的日志功能。

示例

获取JavaScript SDK的源代码并尝试一些示例以快速入门。

1 - JavaScript 客户端 SDK

用于开发 Dapr 应用的 JavaScript 客户端 SDK

介绍

Dapr 客户端使您能够与 Dapr sidecar 进行通信，并访问其面向客户端的功能，如发布事件、调用输出绑定、状态管理、密钥管理等。

前置条件

安装和导入 Dapr 的 JS SDK

使用 npm 安装 SDK：

npm i @dapr/dapr --save

导入库：

import { DaprClient, DaprServer, HttpMethod, CommunicationProtocolEnum } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // 示例服务器的 Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 示例服务器的应用主机
const serverPort = "50051"; // 示例服务器的应用端口

// HTTP 示例
const client = new DaprClient({ daprHost, daprPort });

// GRPC 示例
const client = new DaprClient({ daprHost, daprPort, communicationProtocol: CommunicationProtocolEnum.GRPC });

运行

要运行示例，您可以使用两种不同的协议与 Dapr sidecar 交互：HTTP（默认）或 gRPC。

使用 HTTP（默认）

import { DaprClient } from "@dapr/dapr";
const client = new DaprClient({ daprHost, daprPort });

# 使用 dapr run
dapr run --app-id example-sdk --app-protocol http -- npm run start

# 或者，使用 npm 脚本
npm run start:dapr-http

使用 gRPC

由于 HTTP 是默认协议，您需要调整通信协议以使用 gRPC。您可以通过向客户端或服务器构造函数传递一个额外的参数来实现这一点。

import { DaprClient, CommunicationProtocol } from "@dapr/dapr";
const client = new DaprClient({ daprHost, daprPort, communicationProtocol: CommunicationProtocol.GRPC });

# 使用 dapr run
dapr run --app-id example-sdk --app-protocol grpc -- npm run start

# 或者，使用 npm 脚本
npm run start:dapr-grpc

环境变量

Dapr sidecar 端点

您可以使用 DAPR_HTTP_ENDPOINT 和 DAPR_GRPC_ENDPOINT 环境变量分别设置 Dapr sidecar 的 HTTP 和 gRPC 端点。当这些变量被设置时，daprHost 和 daprPort 不需要在构造函数的选项参数中设置，客户端将自动从提供的端点中解析它们。

import { DaprClient, CommunicationProtocol } from "@dapr/dapr";

// 使用 HTTP，当 DAPR_HTTP_ENDPOINT 被设置时
const client = new DaprClient();

// 使用 gRPC，当 DAPR_GRPC_ENDPOINT 被设置时
const client = new DaprClient({ communicationProtocol: CommunicationProtocol.GRPC });

如果环境变量被设置，但 daprHost 和 daprPort 值被传递给构造函数，后者将优先于环境变量。

Dapr API 令牌

您可以使用 DAPR_API_TOKEN 环境变量设置 Dapr API 令牌。当此变量被设置时，daprApiToken 不需要在构造函数的选项参数中设置，客户端将自动获取它。

通用

增加主体大小

您可以通过使用 DaprClient 的选项增加应用程序与 sidecar 通信时使用的主体大小。

import { DaprClient, CommunicationProtocol } from "@dapr/dapr";

// 允许使用 10Mb 的主体大小
// 默认是 4Mb
const client = new DaprClient({
  daprHost,
  daprPort,
  communicationProtocol: CommunicationProtocol.HTTP,
  maxBodySizeMb: 10,
});

代理请求

通过代理请求，我们可以利用 Dapr 的 sidecar 架构带来的独特功能，如服务发现、日志记录等，使我们能够立即“升级”我们的 gRPC 服务。在社区电话 41 中演示了 gRPC 代理的这一特性。

创建代理

要执行 gRPC 代理，只需通过调用 client.proxy.create() 方法创建一个代理：

// 一如既往，创建一个到我们 Dapr sidecar 的客户端
// 这个客户端负责确保 sidecar 已启动，我们可以通信，...
const clientSidecar = new DaprClient({ daprHost, daprPort, communicationProtocol: CommunicationProtocol.GRPC });

// 创建一个允许我们使用 gRPC 代码的代理
const clientProxy = await clientSidecar.proxy.create<GreeterClient>(GreeterClient);

我们现在可以调用在我们的 GreeterClient 接口中定义的方法（在这种情况下是来自 Hello World 示例）

技术细节

gRPC 服务在 Dapr 中启动。我们通过 --app-port 告诉 Dapr 这个 gRPC 服务器运行在哪个端口，并通过 --app-id <APP_ID_HERE> 给它一个唯一的 Dapr 应用 ID
我们现在可以通过一个将连接到 sidecar 的客户端调用 Dapr sidecar
在调用 Dapr sidecar 时，我们提供一个名为 dapr-app-id 的元数据键，其值为在 Dapr 中启动的 gRPC 服务器（例如，在我们的示例中为 server）
Dapr 现在将调用转发到配置的 gRPC 服务器

构建块

JavaScript 客户端 SDK 允许您与所有 Dapr 构建块进行接口交互，重点是客户端到 sidecar 的功能。

调用 API

调用服务

import { DaprClient, HttpMethod } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const serviceAppId = "my-app-id";
  const serviceMethod = "say-hello";

  // POST 请求
  const response = await client.invoker.invoke(serviceAppId, serviceMethod, HttpMethod.POST, { hello: "world" });

  // 带有头部的 POST 请求
  const response = await client.invoker.invoke(
    serviceAppId,
    serviceMethod,
    HttpMethod.POST,
    { hello: "world" },
    { headers: { "X-User-ID": "123" } },
  );

  // GET 请求
  const response = await client.invoker.invoke(serviceAppId, serviceMethod, HttpMethod.GET);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关服务调用的完整指南，请访问如何：调用服务。

状态管理 API

保存、获取和删除应用状态

import { DaprClient } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const serviceStoreName = "my-state-store-name";

  // 保存状态
  const response = await client.state.save(
    serviceStoreName,
    [
      {
        key: "first-key-name",
        value: "hello",
        metadata: {
          foo: "bar",
        },
      },
      {
        key: "second-key-name",
        value: "world",
      },
    ],
    {
      metadata: {
        ttlInSeconds: "3", // 这应该覆盖状态项中的 ttl
      },
    },
  );

  // 获取状态
  const response = await client.state.get(serviceStoreName, "first-key-name");

  // 获取批量状态
  const response = await client.state.getBulk(serviceStoreName, ["first-key-name", "second-key-name"]);

  // 状态事务
  await client.state.transaction(serviceStoreName, [
    {
      operation: "upsert",
      request: {
        key: "first-key-name",
        value: "new-data",
      },
    },
    {
      operation: "delete",
      request: {
        key: "second-key-name",
      },
    },
  ]);

  // 删除状态
  const response = await client.state.delete(serviceStoreName, "first-key-name");
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关状态操作的完整列表，请访问如何：获取和保存状态。

查询状态 API

import { DaprClient } from "@dapr/dapr";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const res = await client.state.query("state-mongodb", {
    filter: {
      OR: [
        {
          EQ: { "person.org": "Dev Ops" },
        },
        {
          AND: [
            {
              EQ: { "person.org": "Finance" },
            },
            {
              IN: { state: ["CA", "WA"] },
            },
          ],
        },
      ],
    },
    sort: [
      {
        key: "state",
        order: "DESC",
      },
    ],
    page: {
      limit: 10,
    },
  });

  console.log(res);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

PubSub API

发布消息

import { DaprClient } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const pubSubName = "my-pubsub-name";
  const topic = "topic-a";

  // 以 text/plain 格式发布消息到主题
  // 注意，内容类型是从消息类型推断的，除非明确指定
  const response = await client.pubsub.publish(pubSubName, topic, "hello, world!");
  // 如果发布失败，响应包含错误
  console.log(response);

  // 以 application/json 格式发布消息到主题
  await client.pubsub.publish(pubSubName, topic, { hello: "world" });

  // 将 JSON 消息作为纯文本发布
  const options = { contentType: "text/plain" };
  await client.pubsub.publish(pubSubName, topic, { hello: "world" }, options);

  // 以 application/cloudevents+json 格式发布消息到主题
  // 您还可以使用 cloudevent SDK 创建云事件 https://github.com/cloudevents/sdk-javascript
  const cloudEvent = {
    specversion: "1.0",
    source: "/some/source",
    type: "example",
    id: "1234",
  };
  await client.pubsub.publish(pubSubName, topic, cloudEvent);

  // 将 cloudevent 作为原始负载发布
  const options = { metadata: { rawPayload: true } };
  await client.pubsub.publish(pubSubName, topic, "hello, world!", options);

  // 以 text/plain 格式批量发布多个消息到主题
  await client.pubsub.publishBulk(pubSubName, topic, ["message 1", "message 2", "message 3"]);

  // 以 application/json 格式批量发布多个消息到主题
  await client.pubsub.publishBulk(pubSubName, topic, [
    { hello: "message 1" },
    { hello: "message 2" },
    { hello: "message 3" },
  ]);

  // 使用显式批量发布消息批量发布多个消息
  const bulkPublishMessages = [
    {
      entryID: "entry-1",
      contentType: "application/json",
      event: { hello: "foo message 1" },
    },
    {
      entryID: "entry-2",
      contentType: "application/cloudevents+json",
      event: { ...cloudEvent, data: "foo message 2", datacontenttype: "text/plain" },
    },
    {
      entryID: "entry-3",
      contentType: "text/plain",
      event: "foo message 3",
    },
  ];
  await client.pubsub.publishBulk(pubSubName, topic, bulkPublishMessages);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

Bindings API

调用输出绑定

输出绑定

import { DaprClient } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const bindingName = "my-binding-name";
  const bindingOperation = "create";
  const message = { hello: "world" };

  const response = await client.binding.send(bindingName, bindingOperation, message);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关输出绑定的完整指南，请访问如何：使用绑定。

Secret API

检索 secrets

import { DaprClient } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const secretStoreName = "my-secret-store";
  const secretKey = "secret-key";

  // 从 secret 存储中检索单个 secret
  const response = await client.secret.get(secretStoreName, secretKey);

  // 从 secret 存储中检索所有 secrets
  const response = await client.secret.getBulk(secretStoreName);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关 secrets 的完整指南，请访问如何：检索 secrets。

Configuration API

获取配置键

import { DaprClient } from "@dapr/dapr";

const daprHost = "127.0.0.1";

async function start() {
  const client = new DaprClient({
    daprHost,
    daprPort: process.env.DAPR_GRPC_PORT,
    communicationProtocol: CommunicationProtocolEnum.GRPC,
  });

  const config = await client.configuration.get("config-store", ["key1", "key2"]);
  console.log(config);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

示例输出：

{
   items: {
     key1: { key: 'key1', value: 'foo', version: '', metadata: {} },
     key2: { key: 'key2', value: 'bar2', version: '', metadata: {} }
   }
}

订阅配置更新

import { DaprClient } from "@dapr/dapr";

const daprHost = "127.0.0.1";

async function start() {
  const client = new DaprClient({
    daprHost,
    daprPort: process.env.DAPR_GRPC_PORT,
    communicationProtocol: CommunicationProtocolEnum.GRPC,
  });

  // 订阅配置存储更改的键 "key1" 和 "key2"
  const stream = await client.configuration.subscribeWithKeys("config-store", ["key1", "key2"], async (data) => {
    console.log("订阅接收到来自配置存储的更新：", data);
  });

  // 等待 60 秒并取消订阅。
  await new Promise((resolve) => setTimeout(resolve, 60000));
  stream.stop();
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

示例输出：

订阅接收到来自配置存储的更新： {
  items: { key2: { key: 'key2', value: 'bar', version: '', metadata: {} } }
}
订阅接收到来自配置存储的更新： {
  items: { key1: { key: 'key1', value: 'foobar', version: '', metadata: {} } }
}

Cryptography API

JavaScript SDK 中的 gRPC 客户端仅支持 cryptography API。

import { createReadStream, createWriteStream } from "node:fs";
import { readFile, writeFile } from "node:fs/promises";
import { pipeline } from "node:stream/promises";

import { DaprClient, CommunicationProtocolEnum } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "50050"; // 示例服务器的 Dapr sidecar 端口

async function start() {
  const client = new DaprClient({
    daprHost,
    daprPort,
    communicationProtocol: CommunicationProtocolEnum.GRPC,
  });

  // 使用流加密和解密消息
  await encryptDecryptStream(client);

  // 从缓冲区加密和解密消息
  await encryptDecryptBuffer(client);
}

async function encryptDecryptStream(client: DaprClient) {
  // 首先，加密消息
  console.log("== 使用流加密消息");
  console.log("将 plaintext.txt 加密为 ciphertext.out");

  await pipeline(
    createReadStream("plaintext.txt"),
    await client.crypto.encrypt({
      componentName: "crypto-local",
      keyName: "symmetric256",
      keyWrapAlgorithm: "A256KW",
    }),
    createWriteStream("ciphertext.out"),
  );

  // 解密消息
  console.log("== 使用流解密消息");
  console.log("将 ciphertext.out 解密为 plaintext.out");
  await pipeline(
    createReadStream("ciphertext.out"),
    await client.crypto.decrypt({
      componentName: "crypto-local",
    }),
    createWriteStream("plaintext.out"),
  );
}

async function encryptDecryptBuffer(client: DaprClient) {
  // 读取 "plaintext.txt" 以便我们有一些内容
  const plaintext = await readFile("plaintext.txt");

  // 首先，加密消息
  console.log("== 使用缓冲区加密消息");

  const ciphertext = await client.crypto.encrypt(plaintext, {
    componentName: "crypto-local",
    keyName: "my-rsa-key",
    keyWrapAlgorithm: "RSA",
  });

  await writeFile("test.out", ciphertext);

  // 解密消息
  console.log("== 使用缓冲区解密消息");
  const decrypted = await client.crypto.decrypt(ciphertext, {
    componentName: "crypto-local",
  });

  // 内容应该相等
  if (plaintext.compare(decrypted) !== 0) {
    throw new Error("解密的消息与原始消息不匹配");
  }
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关 cryptography 的完整指南，请访问如何：Cryptography。

分布式锁 API

尝试锁定和解锁 API

import { CommunicationProtocolEnum, DaprClient } from "@dapr/dapr";
import { LockStatus } from "@dapr/dapr/types/lock/UnlockResponse";

const daprHost = "127.0.0.1";
const daprPortDefault = "3500";

async function start() {
  const client = new DaprClient({ daprHost, daprPort });

  const storeName = "redislock";
  const resourceId = "resourceId";
  const lockOwner = "owner1";
  let expiryInSeconds = 1000;

  console.log(`在 ${storeName}, ${resourceId} 上以所有者：${lockOwner} 获取锁`);
  const lockResponse = await client.lock.lock(storeName, resourceId, lockOwner, expiryInSeconds);
  console.log(lockResponse);

  console.log(`在 ${storeName}, ${resourceId} 上以所有者：${lockOwner} 解锁`);
  const unlockResponse = await client.lock.unlock(storeName, resourceId, lockOwner);
  console.log("解锁 API 响应：" + getResponseStatus(unlockResponse.status));
}

function getResponseStatus(status: LockStatus) {
  switch (status) {
    case LockStatus.Success:
      return "成功";
    case LockStatus.LockDoesNotExist:
      return "锁不存在";
    case LockStatus.LockBelongsToOthers:
      return "锁属于他人";
    default:
      return "内部错误";
  }
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关分布式锁的完整指南，请访问如何：使用分布式锁。

Workflow API

Workflow 管理

import { DaprClient } from "@dapr/dapr";

async function start() {
  const client = new DaprClient();

  // 启动一个新的 workflow 实例
  const instanceId = await client.workflow.start("OrderProcessingWorkflow", {
    Name: "Paperclips",
    TotalCost: 99.95,
    Quantity: 4,
  });
  console.log(`启动了 workflow 实例 ${instanceId}`);

  // 获取一个 workflow 实例
  const workflow = await client.workflow.get(instanceId);
  console.log(
    `Workflow ${workflow.workflowName}, 创建于 ${workflow.createdAt.toUTCString()}, 状态为 ${
      workflow.runtimeStatus
    }`,
  );
  console.log(`附加属性：${JSON.stringify(workflow.properties)}`);

  // 暂停一个 workflow 实例
  await client.workflow.pause(instanceId);
  console.log(`暂停了 workflow 实例 ${instanceId}`);

  // 恢复一个 workflow 实例
  await client.workflow.resume(instanceId);
  console.log(`恢复了 workflow 实例 ${instanceId}`);

  // 终止一个 workflow 实例
  await client.workflow.terminate(instanceId);
  console.log(`终止了 workflow 实例 ${instanceId}`);

  // 清除一个 workflow 实例
  await client.workflow.purge(instanceId);
  console.log(`清除了 workflow 实例 ${instanceId}`);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

2 - JavaScript 服务器 SDK

用于开发 Dapr 应用的 JavaScript 服务器 SDK

介绍

Dapr 服务器使您能够接收来自 Dapr sidecar 的通信，并访问其面向服务器的功能，例如：事件订阅、接收输入绑定等。

准备条件

安装和导入 Dapr 的 JS SDK

使用 npm 安装 SDK：

npm i @dapr/dapr --save

导入库：

import { DaprServer, CommunicationProtocolEnum } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

// HTTP 示例
const server = new DaprServer({
  serverHost,
  serverPort,
  communicationProtocol: CommunicationProtocolEnum.HTTP, // DaprClient 使用与 DaprServer 相同的通信协议，除非另有说明
  clientOptions: {
    daprHost,
    daprPort,
  },
});

// GRPC 示例
const server = new DaprServer({
  serverHost,
  serverPort,
  communicationProtocol: CommunicationProtocolEnum.GRPC,
  clientOptions: {
    daprHost,
    daprPort,
  },
});

运行

要运行示例，您可以使用两种不同的协议与 Dapr sidecar 交互：HTTP（默认）或 gRPC。

使用 HTTP（内置 express 网络服务器）

import { DaprServer } from "@dapr/dapr";

const server = new DaprServer({
  serverHost: appHost,
  serverPort: appPort,
  clientOptions: {
    daprHost,
    daprPort,
  },
});
// 在服务器启动前初始化订阅，Dapr sidecar 依赖于这些
await server.start();

# 使用 dapr run
dapr run --app-id example-sdk --app-port 50051 --app-protocol http -- npm run start

# 或者，使用 npm 脚本
npm run start:dapr-http

ℹ️ 注意： 这里需要 app-port，因为这是我们的服务器需要绑定的地方。Dapr 将检查应用程序是否绑定到此端口，然后完成启动。

使用 HTTP（自带 express 网络服务器）

除了使用内置的网络服务器进行 Dapr sidecar 到应用程序的通信，您还可以自带实例。这在构建 REST API 后端并希望直接集成 Dapr 时非常有用。

注意，这目前仅适用于 express。

💡 注意：使用自定义网络服务器时，SDK 将配置服务器属性，如最大主体大小，并向其添加新路由。这些路由是独特的，以避免与您的应用程序发生任何冲突，但不能保证不发生冲突。

import { DaprServer, CommunicationProtocolEnum } from "@dapr/dapr";
import express from "express";

const myApp = express();

myApp.get("/my-custom-endpoint", (req, res) => {
  res.send({ msg: "My own express app!" });
});

const daprServer = new DaprServer({
      serverHost: "127.0.0.1", // 应用主机
      serverPort: "50002", // 应用端口
      serverHttp: myApp,
      clientOptions: {
        daprHost,
        daprPort
      }
    });

// 在服务器启动前初始化订阅，Dapr sidecar 使用它。
// 这也将初始化应用服务器本身（无需调用 `app.listen`）。
await daprServer.start();

配置完上述内容后，您可以像往常一样调用您的自定义端点：

const res = await fetch(`http://127.0.0.1:50002/my-custom-endpoint`);
const json = await res.json();

使用 gRPC

由于 HTTP 是默认的，您需要调整通信协议以使用 gRPC。您可以通过向客户端或服务器构造函数传递额外的参数来实现这一点。

import { DaprServer, CommunicationProtocol } from "@dapr/dapr";

const server = new DaprServer({
  serverHost: appHost,
  serverPort: appPort,
  communicationProtocol: CommunicationProtocolEnum.GRPC,
  clientOptions: {
    daprHost,
    daprPort,
  },
});
// 在服务器启动前初始化订阅，Dapr sidecar 依赖于这些
await server.start();

# 使用 dapr run
dapr run --app-id example-sdk --app-port 50051 --app-protocol grpc -- npm run start

# 或者，使用 npm 脚本
npm run start:dapr-grpc

ℹ️ 注意： 这里需要 app-port，因为这是我们的服务器需要绑定的地方。Dapr 将检查应用程序是否绑定到此端口，然后完成启动。

构建块

JavaScript 服务器 SDK 允许您与所有 Dapr 构建块进行接口交互，重点是 sidecar 到应用程序的功能。

调用 API

监听调用

import { DaprServer, DaprInvokerCallbackContent } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const callbackFunction = (data: DaprInvokerCallbackContent) => {
    console.log("Received body: ", data.body);
    console.log("Received metadata: ", data.metadata);
    console.log("Received query: ", data.query);
    console.log("Received headers: ", data.headers); // 仅在 HTTP 中可用
  };

  await server.invoker.listen("hello-world", callbackFunction, { method: HttpMethod.GET });

  // 您现在可以使用您的应用 ID 和方法 "hello-world" 调用服务

  await server.start();
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关服务调用的完整指南，请访问如何：调用服务。

PubSub API

订阅消息

可以通过多种方式订阅消息，以提供接收主题消息的灵活性：

通过 subscribe 方法直接订阅
通过 subscribeWithOptions 方法直接订阅并带有选项
通过 susbcribeOnEvent 方法之后订阅

每次事件到达时，我们将其主体作为 data 传递，并将头信息作为 headers 传递，其中可以包含事件发布者的属性（例如，来自 IoT Hub 的设备 ID）

Dapr 要求在启动时设置订阅，但在 JS SDK 中，我们允许之后添加事件处理程序，为您提供编程的灵活性。

下面提供了一个示例

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const pubSubName = "my-pubsub-name";
  const topic = "topic-a";

  // 为主题配置订阅者
  // 方法 1：通过 `subscribe` 方法直接订阅
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) =>
    console.log(`Received Data: ${JSON.stringify(data)} with headers: ${JSON.stringify(headers)}`),
  );

  // 方法 2：通过 `subscribeWithOptions` 方法直接订阅并带有选项
  await server.pubsub.subscribeWithOptions(pubSubName, topic, {
    callback: async (data: any, headers: object) =>
      console.log(`Received Data: ${JSON.stringify(data)} with headers: ${JSON.stringify(headers)}`),
  });

  // 方法 3：通过 `susbcribeOnEvent` 方法之后订阅
  // 注意：我们使用默认值，因为如果没有传递路由（空选项），我们将使用 "default" 作为路由名称
  await server.pubsub.subscribeWithOptions("pubsub-redis", "topic-options-1", {});
  server.pubsub.subscribeToRoute("pubsub-redis", "topic-options-1", "default", async (data: any, headers: object) => {
    console.log(`Received Data: ${JSON.stringify(data)} with headers: ${JSON.stringify(headers)}`);
  });

  // 启动服务器
  await server.start();
}

有关状态操作的完整列表，请访问如何：发布和订阅。

使用 SUCCESS/RETRY/DROP 状态订阅

Dapr 支持重试逻辑的状态码，以指定消息处理后应执行的操作。

⚠️ JS SDK 允许在同一主题上有多个回调，我们处理状态优先级为 RETRY > DROP > SUCCESS，默认为 SUCCESS

⚠️ 确保在应用程序中配置弹性以处理 RETRY 消息

在 JS SDK 中，我们通过 DaprPubSubStatusEnum 枚举支持这些消息。为了确保 Dapr 将重试，我们还配置了一个弹性策略。

components/resiliency.yaml

apiVersion: dapr.io/v1alpha1
kind: Resiliency
metadata:
  name: myresiliency
spec:
  policies:
    retries:
      # 全局重试策略用于入站组件操作
      DefaultComponentInboundRetryPolicy:
        policy: constant
        duration: 500ms
        maxRetries: 10
  targets:
    components:
      messagebus:
        inbound:
          retry: DefaultComponentInboundRetryPolicy

src/index.ts

import { DaprServer, DaprPubSubStatusEnum } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const pubSubName = "my-pubsub-name";
  const topic = "topic-a";

  // 成功处理消息
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
    return DaprPubSubStatusEnum.SUCCESS;
  });

  // 重试消息
  // 注意：此示例将继续重试传递消息
  // 注意 2：每个组件可以有自己的重试配置
  //   例如，https://docs.dapr.io/reference/components-reference/supported-pubsub/setup-redis-pubsub/
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
    return DaprPubSubStatusEnum.RETRY;
  });

  // 丢弃消息
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
    return DaprPubSubStatusEnum.DROP;
  });

  // 启动服务器
  await server.start();
}

基于规则订阅消息

Dapr 支持路由消息到不同的处理程序（路由）基于规则。

例如，您正在编写一个需要根据消息的 “type” 处理消息的应用程序，使用 Dapr，您可以将它们发送到不同的路由 handlerType1 和 handlerType2，默认路由为 handlerDefault

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const pubSubName = "my-pubsub-name";
  const topic = "topic-a";

  // 为主题配置订阅者并设置规则
  // 注意：默认路由和匹配模式是可选的
  await server.pubsub.subscribe("pubsub-redis", "topic-1", {
    default: "/default",
    rules: [
      {
        match: `event.type == "my-type-1"`,
        path: "/type-1",
      },
      {
        match: `event.type == "my-type-2"`,
        path: "/type-2",
      },
    ],
  });

  // 为每个路由添加处理程序
  server.pubsub.subscribeToRoute("pubsub-redis", "topic-1", "default", async (data) => {
    console.log(`Handling Default`);
  });
  server.pubsub.subscribeToRoute("pubsub-redis", "topic-1", "type-1", async (data) => {
    console.log(`Handling Type 1`);
  });
  server.pubsub.subscribeToRoute("pubsub-redis", "topic-1", "type-2", async (data) => {
    console.log(`Handling Type 2`);
  });

  // 启动服务器
  await server.start();
}

使用通配符订阅

支持流行的通配符 * 和 +（请确保验证 pubsub 组件是否支持）并可以按如下方式订阅：

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const pubSubName = "my-pubsub-name";

  // * 通配符
  await server.pubsub.subscribe(pubSubName, "/events/*", async (data: any, headers: object) =>
    console.log(`Received Data: ${JSON.stringify(data)}`),
  );

  // + 通配符
  await server.pubsub.subscribe(pubSubName, "/events/+/temperature", async (data: any, headers: object) =>
    console.log(`Received Data: ${JSON.stringify(data)}`),
  );

  // 启动服务器
  await server.start();
}

批量订阅消息

支持批量订阅，并可通过以下 API 获得：

通过 subscribeBulk 方法进行批量订阅：maxMessagesCount 和 maxAwaitDurationMs 是可选的；如果未提供，将使用相关组件的默认值。

在监听消息时，应用程序以批量方式从 Dapr 接收消息。然而，与常规订阅一样，回调函数一次接收一条消息，用户可以选择返回 DaprPubSubStatusEnum 值以确认成功、重试或丢弃消息。默认行为是返回成功响应。

请参阅此文档以获取更多详细信息。

import { DaprServer } from "@dapr/dapr";

const pubSubName = "orderPubSub";
const topic = "topicbulk";

const daprHost = process.env.DAPR_HOST || "127.0.0.1";
const daprHttpPort = process.env.DAPR_HTTP_PORT || "3502";
const serverHost = process.env.SERVER_HOST || "127.0.0.1";
const serverPort = process.env.APP_PORT || 5001;

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort: daprHttpPort,
    },
  });

  // 使用默认配置向主题发布多条消息。
  await client.pubsub.subscribeBulk(pubSubName, topic, (data) =>
    console.log("Subscriber received: " + JSON.stringify(data)),
  );

  // 使用特定的 maxMessagesCount 和 maxAwaitDurationMs 向主题发布多条消息。
  await client.pubsub.subscribeBulk(
    pubSubName,
    topic,
    (data) => {
      console.log("Subscriber received: " + JSON.stringify(data));
      return DaprPubSubStatusEnum.SUCCESS; // 如果应用程序没有返回任何内容，默认是 SUCCESS。应用程序还可以根据传入的消息返回 RETRY 或 DROP。
    },
    {
      maxMessagesCount: 100,
      maxAwaitDurationMs: 40,
    },
  );
}

死信主题

Dapr 支持死信主题。这意味着当消息处理失败时，它会被发送到死信队列。例如，当消息在 /my-queue 上处理失败时，它将被发送到 /my-queue-failed。例如，当消息在 /my-queue 上处理失败时，它将被发送到 /my-queue-failed。

您可以使用 subscribeWithOptions 方法的以下选项：

deadletterTopic：指定死信主题名称（注意：如果未提供，我们将创建一个名为 deadletter 的主题）
deadletterCallback：作为死信处理程序触发的方法

在 JS SDK 中实现死信支持可以通过以下方式：

作为选项传递 deadletterCallback
通过 subscribeToRoute 手动订阅路由

下面提供了一个示例

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr sidecar 主机
const daprPort = "3500"; // Dapr sidecar 端口
const serverHost = "127.0.0.1"; // 应用主机
const serverPort = "50051"; // 应用端口

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const pubSubName = "my-pubsub-name";

  // 方法 1（通过 subscribeWithOptions 直接订阅）
  await server.pubsub.subscribeWithOptions("pubsub-redis", "topic-options-5", {
    callback: async (data: any) => {
      throw new Error("Triggering Deadletter");
    },
    deadLetterCallback: async (data: any) => {
      console.log("Handling Deadletter message");
    },
  });

  // 方法 2（之后订阅）
  await server.pubsub.subscribeWithOptions("pubsub-redis", "topic-options-1", {
    deadletterTopic: "my-deadletter-topic",
  });
  server.pubsub.subscribeToRoute("pubsub-redis", "topic-options-1", "default", async () => {
    throw new Error("Triggering Deadletter");
  });
  server.pubsub.subscribeToRoute("pubsub-redis", "topic-options-1", "my-deadletter-topic", async () => {
    console.log("Handling Deadletter message");
  });

  // 启动服务器
  await server.start();
}

Bindings API

接收输入绑定

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";
const serverHost = "127.0.0.1";
const serverPort = "5051";

async function start() {
  const server = new DaprServer({
    serverHost,
    serverPort,
    clientOptions: {
      daprHost,
      daprPort,
    },
  });

  const bindingName = "my-binding-name";

  const response = await server.binding.receive(bindingName, async (data: any) =>
    console.log(`Got Data: ${JSON.stringify(data)}`),
  );

  await server.start();
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

有关输出绑定的完整指南，请访问如何：使用绑定。

Configuration API

💡 配置 API 目前仅通过 gRPC 可用

获取配置值

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";
const serverHost = "127.0.0.1";
const serverPort = "5051";

async function start() {
  const client = new DaprClient({
    daprHost,
    daprPort,
    communicationProtocol: CommunicationProtocolEnum.GRPC,
  });
  const config = await client.configuration.get("config-redis", ["myconfigkey1", "myconfigkey2"]);
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

订阅键更改

import { DaprServer } from "@dapr/dapr";

const daprHost = "127.0.0.1";
const daprPort = "3500";
const serverHost = "127.0.0.1";
const serverPort = "5051";

async function start() {
  const client = new DaprClient({
    daprHost,
    daprPort,
    communicationProtocol: CommunicationProtocolEnum.GRPC,
  });
  const stream = await client.configuration.subscribeWithKeys("config-redis", ["myconfigkey1", "myconfigkey2"], () => {
    // 收到键更新
  });

  // 当您准备好停止监听时，调用以下命令
  await stream.close();
}

start().catch((e) => {
  console.error(e);
  process.exit(1);
});

3 - JavaScript SDK for Actors

如何使用 Dapr JavaScript SDK 快速上手 actor

Dapr actors 包允许您通过 JavaScript 应用程序与 Dapr 虚拟 actor 交互。以下示例展示了如何使用 JavaScript SDK 与虚拟 actor 进行交互。

有关 Dapr actor 的详细介绍，请访问 actor 概述页面。

前置条件

场景

以下代码示例大致描述了一个停车场车位监控系统的场景，可以在 Mark Russinovich 的这个视频中看到。

一个停车场由数百个停车位组成，每个停车位都配有一个传感器，该传感器向集中监控系统提供更新。停车位传感器（即我们的 actor）用于检测停车位是否被占用或可用。

要运行此示例，请克隆源代码，源代码位于 JavaScript SDK 示例目录中。

Actor 接口

actor 接口定义了 actor 实现和调用 actor 的客户端之间共享的契约。在下面的示例中，我们为停车场传感器创建了一个接口。每个传感器都有两个方法：carEnter 和 carLeave，它们定义了停车位的状态：

export default interface ParkingSensorInterface {
  carEnter(): Promise<void>;
  carLeave(): Promise<void>;
}

Actor 实现

actor 实现通过扩展基类型 AbstractActor 并实现 actor 接口（在此示例中为 ParkingSensorInterface）来定义一个类。

以下代码描述了一个 actor 实现以及一些辅助方法。

import { AbstractActor } from "@dapr/dapr";
import ParkingSensorInterface from "./ParkingSensorInterface";

export default class ParkingSensorImpl extends AbstractActor implements ParkingSensorInterface {
  async carEnter(): Promise<void> {
    // 实现更新停车位被占用的状态。
  }

  async carLeave(): Promise<void> {
    // 实现更新停车位可用的状态。
  }

  private async getInfo(): Promise<object> {
    // 实现从停车位传感器请求更新。
  }

  /**
   * @override
   */
  async onActivate(): Promise<void> {
    // 由 AbstractActor 调用的初始化逻辑。
  }
}

配置 Actor 运行时

要配置 actor 运行时，请使用 DaprClientOptions。各种参数及其默认值记录在如何：在 Dapr 中使用虚拟 actor中。

注意，超时和间隔应格式化为 time.ParseDuration 字符串，这是一种用于表示时间段的格式。

import { CommunicationProtocolEnum, DaprClient, DaprServer } from "@dapr/dapr";

// 使用 DaprClientOptions 配置 actor 运行时。
const clientOptions = {
  daprHost: daprHost,
  daprPort: daprPort,
  communicationProtocol: CommunicationProtocolEnum.HTTP,
  actor: {
    actorIdleTimeout: "1h",
    actorScanInterval: "30s",
    drainOngoingCallTimeout: "1m",
    drainRebalancedActors: true,
    reentrancy: {
      enabled: true,
      maxStackDepth: 32,
    },
    remindersStoragePartitions: 0,
  },
};

// 在创建 DaprServer 和 DaprClient 时使用这些选项。

// 注意，DaprServer 内部创建了一个 DaprClient，需要使用 clientOptions 进行配置。
const server = new DaprServer({ serverHost, serverPort, clientOptions });

const client = new DaprClient(clientOptions);

注册 Actor

使用 DaprServer 包初始化并注册您的 actor：

import { DaprServer } from "@dapr/dapr";
import ParkingSensorImpl from "./ParkingSensorImpl";

const daprHost = "127.0.0.1";
const daprPort = "50000";
const serverHost = "127.0.0.1";
const serverPort = "50001";

const server = new DaprServer({
  serverHost,
  serverPort,
  clientOptions: {
    daprHost,
    daprPort,
  },
});

await server.actor.init(); // 让服务器知道我们需要 actor
server.actor.registerActor(ParkingSensorImpl); // 注册 actor
await server.start(); // 启动服务器

// 要获取已注册的 actor，可以调用 `getRegisteredActors`：
const resRegisteredActors = await server.actor.getRegisteredActors();
console.log(`Registered Actors: ${JSON.stringify(resRegisteredActors)}`);

调用 Actor 方法

在注册 actor 之后，使用 ActorProxyBuilder 创建一个实现 ParkingSensorInterface 的代理对象。您可以通过直接调用代理对象上的方法来调用 actor 方法。在内部，它会转换为对 actor API 的网络调用并获取结果。

import { ActorId, DaprClient } from "@dapr/dapr";
import ParkingSensorImpl from "./ParkingSensorImpl";
import ParkingSensorInterface from "./ParkingSensorInterface";

const daprHost = "127.0.0.1";
const daprPort = "50000";

const client = new DaprClient({ daprHost, daprPort });

// 创建一个新的 actor 构建器。它可以用于创建多种类型的 actor。
const builder = new ActorProxyBuilder<ParkingSensorInterface>(ParkingSensorImpl, client);

// 创建一个新的 actor 实例。
const actor = builder.build(new ActorId("my-actor"));
// 或者，使用随机 ID
// const actor = builder.build(ActorId.createRandomId());

// 调用方法。
await actor.carEnter();

使用 actor 的状态

import { AbstractActor } from "@dapr/dapr";
import ActorStateInterface from "./ActorStateInterface";

export default class ActorStateExample extends AbstractActor implements ActorStateInterface {
  async setState(key: string, value: any): Promise<void> {
    await this.getStateManager().setState(key, value);
    await this.getStateManager().saveState();
  }

  async removeState(key: string): Promise<void> {
    await this.getStateManager().removeState(key);
    await this.getStateManager().saveState();
  }

  // 使用特定类型获取状态
  async getState<T>(key: string): Promise<T | null> {
    return await this.getStateManager<T>().getState(key);
  }

  // 不指定类型获取状态为 `any`
  async getState(key: string): Promise<any> {
    return await this.getStateManager().getState(key);
  }
}

Actor 定时器和提醒

JS SDK 支持 actor 通过注册定时器或提醒来在自身上安排周期性工作。定时器和提醒之间的主要区别在于，Dapr actor 运行时在停用后不保留有关定时器的任何信息，但使用 Dapr actor 状态提供程序持久化提醒信息。

这种区别允许用户在轻量级但无状态的定时器与更耗资源但有状态的提醒之间进行权衡。

定时器和提醒的调度接口是相同的。有关调度配置的更深入了解，请参阅 actor 定时器和提醒文档。

Actor 定时器

// ...

const actor = builder.build(new ActorId("my-actor"));

// 注册一个定时器
await actor.registerActorTimer(
  "timer-id", // 定时器的唯一名称。
  "cb-method", // 定时器触发时要执行的回调方法。
  Temporal.Duration.from({ seconds: 2 }), // DueTime
  Temporal.Duration.from({ seconds: 1 }), // Period
  Temporal.Duration.from({ seconds: 1 }), // TTL
  50, // 要发送到定时器回调的状态。
);

// 删除定时器
await actor.unregisterActorTimer("timer-id");

Actor 提醒

// ...

const actor = builder.build(new ActorId("my-actor"));

// 注册一个提醒，它有一个默认回调：`receiveReminder`
await actor.registerActorReminder(
  "reminder-id", // 提醒的唯一名称。
  Temporal.Duration.from({ seconds: 2 }), // DueTime
  Temporal.Duration.from({ seconds: 1 }), // Period
  Temporal.Duration.from({ seconds: 1 }), // TTL
  100, // 要发送到提醒回调的状态。
);

// 删除提醒
await actor.unregisterActorReminder("reminder-id");

要处理回调，您需要在 actor 中重写默认的 receiveReminder 实现。例如，从我们原来的 actor 实现中：

export default class ParkingSensorImpl extends AbstractActor implements ParkingSensorInterface {
  // ...

  /**
   * @override
   */
  async receiveReminder(state: any): Promise<void> {
    // 在这里处理
  }

  // ...
}

有关 actor 的完整指南，请访问如何：在 Dapr 中使用虚拟 actor。

4 - JavaScript SDK中的日志记录

配置JavaScript SDK中的日志记录

介绍

JavaScript SDK自带一个内置的Console日志记录器。SDK会生成各种内部日志，帮助用户理解事件流程并排查问题。用户可以自定义日志的详细程度，并提供自己的日志记录器实现。

配置日志级别

日志记录有五个级别，按重要性从高到低排列 - error、warn、info、verbose和debug。设置日志级别意味着日志记录器将记录所有该级别及更高重要性的日志。例如，设置为verbose级别意味着SDK不会记录debug级别的日志。默认的日志级别是info。

Dapr Client

import { CommunicationProtocolEnum, DaprClient, LogLevel } from "@dapr/dapr";

// 创建一个日志级别设置为verbose的客户端实例。
const client = new DaprClient({
  daprHost,
  daprPort,
  communicationProtocol: CommunicationProtocolEnum.HTTP,
  logger: { level: LogLevel.Verbose },
});

有关如何使用Client的更多详细信息，请参见JavaScript Client。

DaprServer

import { CommunicationProtocolEnum, DaprServer, LogLevel } from "@dapr/dapr";

// 创建一个日志级别设置为error的服务器实例。
const server = new DaprServer({
  serverHost,
  serverPort,
  clientOptions: {
    daprHost,
    daprPort,
    logger: { level: LogLevel.Error },
  },
});

有关如何使用Server的更多详细信息，请参见JavaScript Server。

自定义LoggerService

JavaScript SDK使用内置的Console进行日志记录。要使用自定义日志记录器，如Winston或Pino，可以实现LoggerService接口。

基于Winston的日志记录：

创建LoggerService的新实现。

import { LoggerService } from "@dapr/dapr";
import * as winston from "winston";

export class WinstonLoggerService implements LoggerService {
  private logger;

  constructor() {
    this.logger = winston.createLogger({
      transports: [new winston.transports.Console(), new winston.transports.File({ filename: "combined.log" })],
    });
  }

  error(message: any, ...optionalParams: any[]): void {
    this.logger.error(message, ...optionalParams);
  }
  warn(message: any, ...optionalParams: any[]): void {
    this.logger.warn(message, ...optionalParams);
  }
  info(message: any, ...optionalParams: any[]): void {
    this.logger.info(message, ...optionalParams);
  }
  verbose(message: any, ...optionalParams: any[]): void {
    this.logger.verbose(message, ...optionalParams);
  }
  debug(message: any, ...optionalParams: any[]): void {
    this.logger.debug(message, ...optionalParams);
  }
}

将新的实现传递给SDK。

import { CommunicationProtocolEnum, DaprClient, LogLevel } from "@dapr/dapr";
import { WinstonLoggerService } from "./WinstonLoggerService";

const winstonLoggerService = new WinstonLoggerService();

// 创建一个日志级别设置为verbose且日志服务为winston的客户端实例。
const client = new DaprClient({
  daprHost,
  daprPort,
  communicationProtocol: CommunicationProtocolEnum.HTTP,
  logger: { level: LogLevel.Verbose, service: winstonLoggerService },
});

5 - JavaScript 示例

通过一些示例来学习如何使用 Dapr JavaScript SDK！

快速开始

状态管理: 了解 Dapr 的状态管理功能
发布与订阅: 创建您自己的发布/订阅系统
密钥管理: 学习如何管理密钥
服务调用: 掌握服务调用的技巧

想要分享您的文章？告诉我们！我们会将其添加到下面的列表中。

6 - 如何：在 JavaScript SDK 中编写和管理 Dapr 工作流

如何使用 Dapr JavaScript SDK 快速启动和运行工作流

我们将创建一个 Dapr 工作流并通过控制台调用它。在这个示例中，您将：

使用 JavaScript 工作流 worker 来执行工作流实例
利用 JavaScript 工作流客户端和 API 调用来启动和终止工作流实例

此示例在自托管模式下运行，使用 dapr init 的默认配置。

先决条件

Dapr CLI 和已初始化的环境。
Node.js 和 npm
Docker Desktop
确保您使用的是最新版本的 proto 绑定

设置环境

克隆 JavaScript SDK 仓库并进入其中。

git clone https://github.com/dapr/js-sdk
cd js-sdk

从 JavaScript SDK 根目录，导航到 Dapr 工作流示例。

cd examples/workflow/authoring

运行以下命令以安装运行此工作流示例所需的 Dapr JavaScript SDK 依赖。

npm install

运行 `activity-sequence.ts`

activity-sequence 文件在 Dapr 工作流运行时中注册了一个工作流和一个活动。工作流是按顺序执行的一系列活动。我们使用 DaprWorkflowClient 来调度一个新的工作流实例并等待其完成。

const daprHost = "localhost";
const daprPort = "50001";
const workflowClient = new DaprWorkflowClient({
  daprHost,
  daprPort,
});
const workflowRuntime = new WorkflowRuntime({
  daprHost,
  daprPort,
});

const hello = async (_: WorkflowActivityContext, name: string) => {
  return `Hello ${name}!`;
};

const sequence: TWorkflow = async function* (ctx: WorkflowContext): any {
  const cities: string[] = [];

  const result1 = yield ctx.callActivity(hello, "Tokyo");
  cities.push(result1);
  const result2 = yield ctx.callActivity(hello, "Seattle");
  cities.push(result2);
  const result3 = yield ctx.callActivity(hello, "London");
  cities.push(result3);

  return cities;
};

workflowRuntime.registerWorkflow(sequence).registerActivity(hello);

// 将 worker 启动包装在 try-catch 块中以处理启动期间的任何错误
try {
  await workflowRuntime.start();
  console.log("工作流运行时启动成功");
} catch (error) {
  console.error("启动工作流运行时出错：", error);
}

// 调度一个新的编排
try {
  const id = await workflowClient.scheduleNewWorkflow(sequence);
  console.log(`编排已调度，ID：${id}`);

  // 等待编排完成
  const state = await workflowClient.waitForWorkflowCompletion(id, undefined, 30);

  console.log(`编排完成！结果：${state?.serializedOutput}`);
} catch (error) {
  console.error("调度或等待编排时出错：", error);
}

在上面的代码中：

workflowRuntime.registerWorkflow(sequence) 将 sequence 注册为 Dapr 工作流运行时中的一个工作流。
await workflowRuntime.start(); 构建并启动 Dapr 工作流运行时中的引擎。
await workflowClient.scheduleNewWorkflow(sequence) 在 Dapr 工作流运行时中调度一个新的工作流实例。
await workflowClient.waitForWorkflowCompletion(id, undefined, 30) 等待工作流实例完成。

在终端中，执行以下命令以启动 activity-sequence.ts：

npm run start:dapr:activity-sequence

预期输出

你已启动并运行！Dapr 和您的应用程序日志将出现在这里。

...

== APP == 编排已调度，ID：dc040bea-6436-4051-9166-c9294f9d2201
== APP == 等待 30 秒以完成实例 dc040bea-6436-4051-9166-c9294f9d2201...
== APP == 收到实例 id 为 'dc040bea-6436-4051-9166-c9294f9d2201' 的 "Orchestrator Request" 工作项
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 使用 0 个历史事件重建本地状态...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 处理 2 个新历史事件：[ORCHESTRATORSTARTED=1, EXECUTIONSTARTED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 等待 1 个任务和 0 个事件完成...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 返回 1 个动作
== APP == 收到 "Activity Request" 工作项
== APP == 活动 hello 完成，输出 "Hello Tokyo!" (14 个字符)
== APP == 收到实例 id 为 'dc040bea-6436-4051-9166-c9294f9d2201' 的 "Orchestrator Request" 工作项
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 使用 3 个历史事件重建本地状态...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 处理 2 个新历史事件：[ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 等待 1 个任务和 0 个事件完成...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 返回 1 个动作
== APP == 收到 "Activity Request" 工作项
== APP == 活动 hello 完成，输出 "Hello Seattle!" (16 个字符)
== APP == 收到实例 id 为 'dc040bea-6436-4051-9166-c9294f9d2201' 的 "Orchestrator Request" 工作项
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 使用 6 个历史事件重建本地状态...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 处理 2 个新历史事件：[ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 等待 1 个任务和 0 个事件完成...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 返回 1 个动作
== APP == 收到 "Activity Request" 工作项
== APP == 活动 hello 完成，输出 "Hello London!" (15 个字符)
== APP == 收到实例 id 为 'dc040bea-6436-4051-9166-c9294f9d2201' 的 "Orchestrator Request" 工作项
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 使用 9 个历史事件重建本地状态...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 处理 2 个新历史事件：[ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 编排完成，状态为 COMPLETED
== APP == dc040bea-6436-4051-9166-c9294f9d2201: 返回 1 个动作
INFO[0006] dc040bea-6436-4051-9166-c9294f9d2201: 'sequence' 完成，状态为 COMPLETED。 app_id=activity-sequence-workflow instance=kaibocai-devbox scope=wfengine.backend type=log ver=1.12.3
== APP == 实例 dc040bea-6436-4051-9166-c9294f9d2201 完成
== APP == 编排完成！结果：["Hello Tokyo!","Hello Seattle!","Hello London!"]

JavaScript SDK

安装

结构

入门

客户端

服务器

虚拟演员

日志

示例

1 - JavaScript 客户端 SDK

介绍

前置条件

安装和导入 Dapr 的 JS SDK

运行

使用 HTTP（默认）

使用 gRPC

环境变量

Dapr sidecar 端点

Dapr API 令牌

通用

增加主体大小

代理请求

创建代理

技术细节

构建块

调用 API

调用服务

状态管理 API

保存、获取和删除应用状态

查询状态 API

PubSub API

发布消息

Bindings API

调用输出绑定

Secret API

检索 secrets

Configuration API

获取配置键

订阅配置更新

Cryptography API

分布式锁 API

尝试锁定和解锁 API

Workflow API

Workflow 管理

相关链接

2 - JavaScript 服务器 SDK

介绍

准备条件

安装和导入 Dapr 的 JS SDK

运行

使用 HTTP（内置 express 网络服务器）

使用 HTTP（自带 express 网络服务器）

使用 gRPC

构建块

调用 API

监听调用

PubSub API

订阅消息

使用 SUCCESS/RETRY/DROP 状态订阅

基于规则订阅消息

使用通配符订阅

批量订阅消息

死信主题

Bindings API

接收输入绑定

Configuration API

获取配置值

订阅键更改

相关链接

3 - JavaScript SDK for Actors

前置条件

场景

Actor 接口

Actor 实现

配置 Actor 运行时

注册 Actor

调用 Actor 方法

使用 actor 的状态

Actor 定时器和提醒

Actor 定时器

运行 `activity-sequence.ts`