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 软件包

一个用于在 JavaScript 和 TypeScript 中构建 Dapr 应用程序的客户端库。该客户端抽象了服务调用、状态管理、发布订阅、密钥管理等公开 Dapr API，并提供简单直观的 API 用于构建应用程序。

安装

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

npm install --save @dapr/dapr

结构

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

  • DaprServer：管理所有 Dapr 边车与应用程序之间的通信。
  • DaprClient：管理所有应用程序与 Dapr 边车之间的通信。

上述通信可配置为使用 gRPC 或 HTTP 协议。

Dapr ServerDapr Client

快速入门

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

客户端

创建 JavaScript 客户端并与 Dapr 边车及其他 Dapr 应用程序交互（例如，发布事件、输出绑定支持等）。

服务器

创建 JavaScript 服务器并让 Dapr 边车与您的应用程序交互（例如，订阅事件、输入绑定支持等）。

Actors

创建具有状态、提醒/定时器和方法的虚拟 Actor。


日志

配置和自定义 SDK 日志。

示例

克隆 JavaScript SDK 源代码并试用一些示例以快速入门。

1 - JavaScript 客户端 SDK

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

简介

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

前置条件

安装和导入 Dapr JS SDK

  1. 使用 npm 安装 SDK：
npm i @dapr/dapr --save
  1. 导入库：
import { DaprClient, DaprServer, HttpMethod, CommunicationProtocolEnum } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr 边车主机
const daprPort = "3500"; // 此示例服务器的 Dapr 边车端口
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 边车交互：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 script
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 script
npm run start:dapr-grpc

环境变量

Dapr 边车端点

您可以使用 DAPR_HTTP_ENDPOINT 和 DAPR_GRPC_ENDPOINT 环境变量来分别设置 Dapr 边车的 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 的选项来增加应用程序用于与边车通信的主体大小。

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

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

代理请求

通过代理请求，我们可以利用 Dapr 通过其边车架构带来的独特能力，如服务发现、日志记录等，使我们能够立即"升级"我们的 gRPC 服务。gRPC 代理的此功能在 社区会议 41 中进行了演示。

创建代理

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

// 像往常一样，为我们的 dapr 边车创建一个客户端
// 此客户端负责确保边车已启动、我们可以通信等
const clientSidecar = new DaprClient({ daprHost, daprPort, communicationProtocol: CommunicationProtocol.GRPC });

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

现在我们可以按照 GreeterClient 接口中定义的方法调用（在本例中，该接口来自 Hello World 示例）

幕后（技术工作原理）

架构

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

构建块

JavaScript 客户端 SDK 允许您与所有 Dapr 构建块 交互，重点关注客户端到边车的功能。

调用 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);
});

发布订阅 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!");
  // 如果发布失败，response 包含错误
  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);

  // 以原始负载形式发布云事件
  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);
});

绑定 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);
});

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

密钥 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 secretStoreName = "my-secret-store";
  const secretKey = "secret-key";

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

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

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

有关密钥的完整指南，请访问 操作指南：检索密钥。

配置 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: {} } }
}

加密 API

JavaScript SDK 中仅在 gRPC 客户端上支持加密 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 边车端口

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);
});

有关加密的完整指南，请访问 操作指南：加密。

分布式锁 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 "Success";
    case LockStatus.LockDoesNotExist:
      return "LockDoesNotExist";
    case LockStatus.LockBelongsToOthers:
      return "LockBelongsToOthers";
    default:
      return "InternalError";
  }
}

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

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

工作流 API

工作流管理

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

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

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

  // 获取工作流实例
  const workflow = await client.workflow.get(instanceId);
  console.log(
    `工作流 ${workflow.workflowName}，创建于 ${workflow.createdAt.toUTCString()}，状态为 ${
      workflow.runtimeStatus
    }`,
  );
  console.log(`其他属性：${JSON.stringify(workflow.properties)}`);

  // 暂停工作流实例
  await client.workflow.pause(instanceId);
  console.log(`已暂停工作流实例 ${instanceId}`);

  // 恢复工作流实例
  await client.workflow.resume(instanceId);
  console.log(`已恢复工作流实例 ${instanceId}`);

  // 终止工作流实例
  await client.workflow.terminate(instanceId);
  console.log(`已终止工作流实例 ${instanceId}`);

  // 清除工作流实例
  await client.workflow.purge(instanceId);
  console.log(`已清除工作流实例 ${instanceId}`);
}

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

相关链接

2 - JavaScript Server SDK

用于开发 Dapr 应用程序的 JavaScript Server SDK

简介

Dapr Server 将允许您接收来自 Dapr 边车的通信，并访问其面向服务器的功能，例如：订阅事件、接收输入绑定等。

前置条件

安装和导入 Dapr 的 JS SDK

  1. 使用 npm 安装 SDK：
npm i @dapr/dapr --save
  1. 导入库：
import { DaprServer, CommunicationProtocolEnum } from "@dapr/dapr";

const daprHost = "127.0.0.1"; // Dapr Sidecar Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server

// HTTP Example
const server = new DaprServer({
  serverHost,
  serverPort,
  communicationProtocol: CommunicationProtocolEnum.HTTP, // DaprClient to use same communication protocol as DaprServer, in case DaprClient protocol not mentioned explicitly
  clientOptions: {
    daprHost,
    daprPort,
  },
});

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

运行

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

使用 HTTP（内置 express webserver）

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

const server = new DaprServer({
  serverHost: appHost,
  serverPort: appPort,
  clientOptions: {
    daprHost,
    daprPort,
  },
});
// initialize subscribtions, ... before server start
// the dapr sidecar relies on these
await server.start();

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

# or, using npm script
npm run start:dapr-http

ℹ️ Note: The app-port is required here, as this is where our server will need to bind to. Dapr will check for the application to bind to this port, before finishing start-up.

使用 HTTP（自备 express webserver）

您不一定要使用 Dapr sidecar 的内置 web 服务器与应用程序通信，您也可以自带实例。这在构建 REST API 后端并希望直接集成 Dapr 的场景中非常有用。

注意，目前仅适用于 express。

💡 Note: when using a custom web-server, the SDK will configure server properties like max body size, and add new routes to it. The routes are unique on their own to avoid any collisions with your application, but it’s not guaranteed to not collide.

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", // App Host
      serverPort: "50002", // App Port
      serverHttp: myApp,
      clientOptions: {
        daprHost
        daprPort
      }
    });

// Initialize subscriptions before the server starts, the Dapr sidecar uses it.
// This will also initialize the app server itself (removing the need for `app.listen` to be called).
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,
  },
});
// initialize subscribtions, ... before server start
// the dapr sidecar relies on these
await server.start();

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

# or, using npm script
npm run start:dapr-grpc

ℹ️ Note: The app-port is required here, as this is where our server will need to bind to. Dapr will check for the application to bind to this port, before finishing start-up.

构建块

JavaScript Server SDK 允许您与所有 Dapr 构建块 进行交互，主要专注于 Sidecar 到应用程序的功能。

Invocation API

Listen to an Invocation

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

const daprHost = "127.0.0.1"; // Dapr Sidecar Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server "

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); // only available in HTTP
  };

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

  // You can now invoke the service with your app id and method "hello-world"

  await server.start();
}

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

For a full guide on service invocation visit How-To: Invoke a service.

PubSub API

Subscribe to messages

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

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

每次事件到达时，我们将其正文作为 data 传递，将标头作为 headers 传递，这些标头可能包含事件发布者的属性（例如，来自 IoT Hub 的设备 ID）

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

下面提供了一个示例

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

const daprHost = "127.0.0.1"; // Dapr Sidecar Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server "

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

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

  // Configure Subscriber for a Topic
  // Method 1: Direct subscription through the `subscribe` method
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) =>
    console.log(`Received Data: ${JSON.stringify(data)} with headers: ${JSON.stringify(headers)}`),
  );

  // Method 2: Direct susbcription with options through the `subscribeWithOptions` method
  await server.pubsub.subscribeWithOptions(pubSubName, topic, {
    callback: async (data: any, headers: object) =>
      console.log(`Received Data: ${JSON.stringify(data)} with headers: ${JSON.stringify(headers)}`),
  });

  // Method 3: Subscription afterwards through the `susbcribeOnEvent` method
  // Note: we use default, since if no route was passed (empty options) we utilize "default" as the route name
  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)}`);
  });

  // Start the server
  await server.start();
}

For a full list of state operations visit How-To: Publish & subscribe.

Subscribe with SUCCESS/RETRY/DROP status

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:
      # Global Retry Policy for Inbound Component operations
      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 Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server "

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

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

  // Process a message successfully
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
    return DaprPubSubStatusEnum.SUCCESS;
  });

  // Retry a message
  // Note: this example will keep on retrying to deliver the message
  // Note 2: each component can have their own retry configuration
  //   e.g., 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;
  });

  // Drop a message
  await server.pubsub.subscribe(pubSubName, topic, async (data: any, headers: object) => {
    return DaprPubSubStatusEnum.DROP;
  });

  // Start the server
  await server.start();
}

Subscribe to messages rule based

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

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

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

const daprHost = "127.0.0.1"; // Dapr Sidecar Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server "

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

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

  // Configure Subscriber for a Topic with rule set
  // Note: the default route and match patterns are optional
  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",
      },
    ],
  });

  // Add handlers for each route
  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`);
  });

  // Start the server
  await server.start();
}

Susbcribe with Wildcards

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

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

const daprHost = "127.0.0.1"; // Dapr Sidecar Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server "

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

  const pubSubName = "my-pubsub-name";

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

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

  // Start the server
  await server.start();
}

Bulk Subscribe to messages

批量订阅受支持，可通过以下 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,
    },
  });

  // Publish multiple messages to a topic with default config.
  await client.pubsub.subscribeBulk(pubSubName, topic, (data) =>
    console.log("Subscriber received: " + JSON.stringify(data)),
  );

  // Publish multiple messages to a topic with specific maxMessagesCount and maxAwaitDurationMs.
  await client.pubsub.subscribeBulk(
    pubSubName,
    topic,
    (data) => {
      console.log("Subscriber received: " + JSON.stringify(data));
      return DaprPubSubStatusEnum.SUCCESS; // If App doesn't return anything, the default is SUCCESS. App can also return RETRY or DROP based on the incoming message.
    },
    {
      maxMessagesCount: 100,
      maxAwaitDurationMs: 40,
    },
  );
}

Dead Letter Topics

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 Host
const daprPort = "3500"; // Dapr Sidecar Port of this Example Server
const serverHost = "127.0.0.1"; // App Host of this Example Server
const serverPort = "50051"; // App Port of this Example Server "

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

  const pubSubName = "my-pubsub-name";

  // Method 1 (direct subscribing through 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");
    },
  });

  // Method 2 (subscribe afterwards)
  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");
  });

  // Start server
  await server.start();
}

Bindings API

Receive an Input Binding

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);
});

For a full guide on output bindings visit How-To: Use bindings.

Configuration API

💡 The configuration API is currently only available through gRPC

Getting a configuration value

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);
});

Subscribing to Key Changes

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"], () => {
    // Received a key update
  });

  // When you are ready to stop listening, call the following
  await stream.close();
}

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

相关链接

3 - JavaScript SDK for Actors

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

Dapr Actor 包允许你从 JavaScript 应用程序与 Dapr 虚拟 Actor 交互。下面的示例演示了如何使用 JavaScript SDK 与虚拟 Actor 进行交互。

有关 Dapr Actor 的更深入概述，请访问 Actor 概述页面。

前置条件

场景

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

一个停车场包含数百个停车位，每个停车位都有一个传感器，向中央监控系统提供更新。停车位传感器（我们的 Actor）检测停车位是被占用还是可用。

要立即亲自运行此示例，请克隆源代码，可以在 JavaScript SDK 示例目录中找到。

Actor 接口

Actor 接口定义了 Actor 实现和调用 Actor 的客户端之间共享的合约。在下面的示例中，我们为停车场传感器创建了一个接口。每个传感器有 2 个方法：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 的 Proxy 对象。你可以通过直接调用 Proxy 对象上的方法来调用 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();
  }

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

  // 不使用类型作为 `any` 的 getState
  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 会发出各种内部日志，以帮助用户了解事件链并排查问题。SDK 的使用者可以自定义日志的详细程度，也可以提供自己的日志记录器实现。

配置日志级别

日志共有五个级别，按重要性降序排列 - error、warn、info、verbose 和 debug。将日志设置到某个级别意味着日志记录器将发出所有重要性不低于该级别的日志。例如，设置为 verbose 级别意味着 SDK 将不会发出 debug 级别的日志。默认日志级别为 info。

Dapr Client

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

// create a client instance with log level set to 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";

// create a server instance with log level set to 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();

// create a client instance with log level set to verbose and logger service as winston.
const client = new DaprClient({
  daprHost,
  daprPort,
  communicationProtocol: CommunicationProtocolEnum.HTTP,
  logger: { level: LogLevel.Verbose, service: winstonLoggerService },
});

5 - JavaScript 示例

通过示例快速上手 Dapr JavaScript SDK！

快速入门

文章

想要添加您的文章？告诉我们！，以便我们将其添加到下方

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

如何使用 Dapr JavaScript SDK 快速上手工作流

让我们创建一个 Dapr 工作流并通过控制台调用它。借助提供的工作流示例，你将：

本示例使用了在自托管模式下通过 dapr init 获得的默认配置。

前提条件

设置环境

克隆 JavaScript SDK 仓库并进入该目录。

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

从 JavaScript SDK 根目录进入 Dapr Workflow 示例。

cd examples/workflow/authoring

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

npm install

运行 activity-sequence.ts

activity-sequence 文件向 Dapr Workflow 运行时注册了一个工作流和一个 activity。该工作流是按顺序执行的多个 activity 组成的序列。我们使用 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("Workflow runtime started successfully");
} catch (error) {
  console.error("Error starting workflow runtime:", error);
}

// 调度新的编排
try {
  const id = await workflowClient.scheduleNewWorkflow(sequence);
  console.log(`Orchestration scheduled with ID: ${id}`);

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

  console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
} catch (error) {
  console.error("Error scheduling or waiting for orchestration:", error);
}

在以上代码中：

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

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

npm run start:dapr:activity-sequence

预期输出

You're up and running! Both Dapr and your app logs will appear here.

...

== APP == Orchestration scheduled with ID: dc040bea-6436-4051-9166-c9294f9d2201
== APP == Waiting 30 seconds for instance dc040bea-6436-4051-9166-c9294f9d2201 to complete...
== APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 0 history event...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, EXECUTIONSTARTED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Waiting for 1 task(s) and 0 event(s) to complete...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
== APP == Received "Activity Request" work item
== APP == Activity hello completed with output "Hello Tokyo!" (14 chars)
== APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 3 history event...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Waiting for 1 task(s) and 0 event(s) to complete...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
== APP == Received "Activity Request" work item
== APP == Activity hello completed with output "Hello Seattle!" (16 chars)
== APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 6 history event...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Waiting for 1 task(s) and 0 event(s) to complete...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
== APP == Received "Activity Request" work item
== APP == Activity hello completed with output "Hello London!" (15 chars)
== APP == Received "Orchestrator Request" work item with instance id 'dc040bea-6436-4051-9166-c9294f9d2201'
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Rebuilding local state with 9 history event...
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Processing 2 new history event(s): [ORCHESTRATORSTARTED=1, TASKCOMPLETED=1]
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Orchestration completed with status COMPLETED
== APP == dc040bea-6436-4051-9166-c9294f9d2201: Returning 1 action(s)
INFO[0006] dc040bea-6436-4051-9166-c9294f9d2201: 'sequence' completed with a COMPLETED status.  app_id=activity-sequence-workflow instance=kaibocai-devbox scope=wfengine.backend type=log ver=1.12.3
== APP == Instance dc040bea-6436-4051-9166-c9294f9d2201 completed
== APP == Orchestration completed! Result: ["Hello Tokyo!","Hello Seattle!","Hello London!"]

后续步骤