你当前正在访问 Microsoft Azure Global Edition 技术文档网站。如果需要访问由世纪互联运营的 Microsoft Azure 中国技术文档网站，请访问 https://docs.azure.cn。

Azure 服务总线 client library for JavaScript - version 7.10.0

Azure 服务总线是 Microsoft 提供的一种高度可靠的云消息传送服务。

使用应用程序中的客户端库@azure/service-bus

将消息发送到 Azure 服务总线队列或主题
从 Azure 服务总线队列或订阅接收消息
在 Azure 服务总线命名空间中创建/获取/删除/更新/列出队列/主题/订阅/规则。

版本 7 的资源 @azure/service-bus ：

关键链接：

注意：如果您使用的是 1.1.10 或更低版本，并且想要迁移到此包的最新版本，请查看我们的迁移指南，从服务总线 V1 迁移到服务总线 V7

入门

安装软件包

使用 npm 安装 Azure 服务总线客户端库的最新版本。

npm install @azure/service-bus

目前支持的环境

Node.js的 LTS 版本

Prerequisites

配置 TypeScript

TypeScript 用户需要安装 Node 类型定义：

npm install @types/node

还需要在 tsconfig.json中启用 compilerOptions.allowSyntheticDefaultImports。请注意，如果已启用 compilerOptions.esModuleInterop，则默认启用 allowSyntheticDefaultImports。有关详细信息，请参阅 TypeScript 的编译器选项手册。

JavaScript 捆绑包

若要在浏览器中使用此客户端库，首先需要使用捆绑程序。有关如何执行此操作的详细信息，请参阅我们的捆绑文档。

除了其中介绍的内容之外，此库还需要以下 NodeJS 核心内置模块的其他填充，以便在浏览器中正常工作：

buffer
os
path
process

与 Webpack 捆绑

如果使用 Webpack v5，可以安装以下开发依赖项

npm install --save-dev os-browserify path-browserify

然后将以下内容添加到 webpack.config.js

 const path = require("path");
+const webpack = require("webpack");

 module.exports = {
   entry: "./src/index.ts",
@@ -12,8 +13,21 @@ module.exports = {
       },
     ],
   },
+  plugins: [
+    new webpack.ProvidePlugin({
+      process: "process/browser",
+    }),
+    new webpack.ProvidePlugin({
+      Buffer: ["buffer", "Buffer"],
+    }),
+  ],
   resolve: {
     extensions: [".ts", ".js"],
+    fallback: {
+      buffer: require.resolve("buffer/"),
+      os: require.resolve("os-browserify"),
+      path: require.resolve("path-browserify"),
+    },
   },

捆绑与汇总

如果使用汇总捆绑程序，请安装以下开发依赖项

npm install --save-dev @rollup/plugin-commonjs @rollup/plugin-inject @rollup/plugin-node-resolve

然后在 rollup.config.js 中包含以下内容

+import nodeResolve from "@rollup/plugin-node-resolve";
+import cjs from "@rollup/plugin-commonjs";
+import shim from "rollup-plugin-shim";
+import inject from "@rollup/plugin-inject";

export default {
  // other configs
  plugins: [
+    shim({
+      fs: `export default {}`,
+      net: `export default {}`,
+      tls: `export default {}`,
+      path: `export default {}`,
+      dns: `export function resolve() { }`,
+    }),
+    nodeResolve({
+      mainFields: ["module", "browser"],
+      preferBuiltins: false,
+    }),
+    cjs(),
+    inject({
+      modules: {
+        Buffer: ["buffer", "Buffer"],
+        process: "process",
+      },
+      exclude: ["./**/package.json"],
+    }),
  ]
};

有关使用填充的详细信息，请参阅你最喜欢的捆绑包的文档。

React 本机支持

与浏览器类似，React Native 不支持此 SDK 库使用的某些 JavaScript API，因此您需要为它们提供 polyfill。有关详细信息，请参阅 expo 的 Messaging React Native 示例。

对客户端进行身份验证

与服务总线的交互从 ServiceBusClient 类的实例开始。可以使用连接字符串或使用 Azure Active Directory 凭据向服务总线进行身份验证。

使用连接字符串

此方法将连接字符串引入服务总线实例。可以从 Azure 门户获取连接字符串。

import { ServiceBusClient } from "@azure/service-bus";

const serviceBusClient = new ServiceBusClient("<connectionString>");

有关此构造函数的更多信息，请参阅 API 文档。

使用 Azure Active Directory 凭据

使用 Azure Active Directory 进行身份验证使用 Azure 标识库。

下面的示例使用 DefaultAzureCredential，这是库中许多可用凭据提供程序 @azure/identity 之一。

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

注意：如果您使用自己的接口实现 TokenCredential 来对抗 AAD，请将 service-bus 的“scopes”设置为以下内容以获取相应的令牌：

["https://servicebus.azure.net//user_impersonation"];

有关此构造函数的更多信息，请参阅 API 文档

重要概念

初始化后， ServiceBusClient您可以在服务总线命名空间中与以下资源进行交互：

队列：允许发送和接收消息。通常用于点对点通信。
主题：与队列相反，主题更适合发布/订阅场景。主题可以发送到主题，但需要一个订阅，其中可以有多个并行订阅才能使用。
Subscriptions：从 Topic 消费的机制。每个订阅都是独立的，并接收发送到主题的每条消息的副本。 Rules 和 Filters 可用于定制特定订阅接收的消息。

有关这些资源的详细信息，请参阅什么是 Azure 服务总线？。

要与这些资源交互，应熟悉以下 SDK 概念：

使用 ServiceBusSender 创建的 ServiceBusClient.createSender()将消息发送到队列或主题。
使用 ServiceBusReceiver 使用创建的从 ServiceBusClient.createReceiver()队列或订阅接收消息。
使用ServiceBusSessionReceiver使用或 ServiceBusClient.acceptSession()创建的从ServiceBusClient.acceptNextSession()启用会话的队列或订阅接收消息。

请注意，Queues、Topics 和 Subscriptions 应在使用此库之前创建。

Examples

以下部分提供了涵盖使用 Azure 服务总线的一些常见任务的代码片段

发送消息
接收消息
结算消息
死信队列
使用 Sessions 发送消息
从 Session 接收消息
列表消息会话
管理服务总线命名空间的资源
其他示例

发送消息

创建类的ServiceBusClient实例后，可以使用 ServiceBusSender 方法获取可用于发送消息的实例。

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

const sender = serviceBusClient.createSender("my-queue");

const messages = [
  { body: "Albert Einstein" },
  { body: "Werner Heisenberg" },
  { body: "Marie Curie" },
  { body: "Steven Hawking" },
  { body: "Isaac Newton" },
  { body: "Niels Bohr" },
  { body: "Michael Faraday" },
  { body: "Galileo Galilei" },
  { body: "Johannes Kepler" },
  { body: "Nikolaus Kopernikus" },
];

// sending a single message
await sender.sendMessages(messages[0]);

// sending multiple messages in a single call
// this will fail if the messages cannot fit in a batch
await sender.sendMessages(messages);

// Sends multiple messages using one or more ServiceBusMessageBatch objects as required
let batch = await sender.createMessageBatch();

for (let i = 0; i < messages.length; i++) {
  const message = messages[i];
  if (!batch.tryAddMessage(message)) {
    // Send the current batch as it is full and create a new one
    await sender.sendMessages(batch);
    batch = await sender.createMessageBatch();

    if (!batch.tryAddMessage(messages[i])) {
      throw new Error("Message too big to fit in a batch");
    }
  }
}
// Send the batch
await sender.sendMessages(batch);

接收消息

创建类的ServiceBusClient实例后，可以使用 ServiceBusReceiver 方法获取。

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

const receiver = serviceBusClient.createReceiver("my-queue");

receiveMode有两种可用。

“peekLock” - 在 peekLock 模式下，接收方在队列上指定的持续时间内对消息具有锁定。
“receiveAndDelete” - 在 receiveAndDelete 模式下，收到消息时，将从服务总线中删除消息。

如果选项中未提供 receiveMode，则默认为 “peekLock” 模式。您还可以在 “peekLock” 模式下结算收到的消息。

您可以通过以下 3 种方式之一使用此接收器来接收消息：

获取消息数组

使用 receiveMessages 函数，该函数返回一个解析为消息数组的 Promise。

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

const receiver = serviceBusClient.createReceiver("my-queue");

const myMessages = await receiver.receiveMessages(10);

使用 subscribe 方法设置消息处理程序，并使其在需要时运行。

完成后，致电 receiver.close() 停止接收更多消息。

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

const receiver = serviceBusClient.createReceiver("my-queue");

const myMessageHandler = async (message) => {
  // your code here
  console.log(`message.body: ${message.body}`);
};
const myErrorHandler = async (args) => {
  console.log(
    `Error occurred with ${args.entityPath} within ${args.fullyQualifiedNamespace}: `,
    args.error,
  );
};
receiver.subscribe({
  processMessage: myMessageHandler,
  processError: myErrorHandler,
});

使用 async iterator

使用 getMessageIterator 获取消息的异步迭代器

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

const receiver = serviceBusClient.createReceiver("my-queue");

for await (const message of receiver.getMessageIterator()) {
  // your code here
}

结算消息

收到消息后，您可以根据您希望如何结算消息来调用completeMessage()abandonMessage()deferMessage()、、或 deadLetterMessage() 接收方。

要了解更多信息，请阅读结算收到的消息

死信队列

死信队列是一个 子队列。每个队列或订阅都有自己的死信队列。死信队列存储已明确死信（via receiver.deadLetterMessage()）的消息或已超过其最大送达计数的消息。

为死信子队列创建接收器类似于为订阅或队列创建接收器：

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

// To receive from a queue's dead letter sub-queue
const deadLetterReceiverForQueue = serviceBusClient.createReceiver("queue", {
  subQueueType: "deadLetter",
});

// To receive from a subscription's dead letter sub-queue
const deadLetterReceiverForSubscription = serviceBusClient.createReceiver("topic", "subscription", {
  subQueueType: "deadLetter",
});

// Dead letter receivers work like any other receiver connected to a queue
// ex:
const messages = await deadLetterReceiverForQueue.receiveMessages(5);

for (const message of messages) {
  console.log(`Dead lettered message: ${message.body}`);
}

更彻底地演示死信队列的完整示例：

使用 Sessions 发送消息

使用会话需要您创建启用会话的 Queue 或 Subscription。您可以在此处阅读有关如何在门户中配置此功能的更多信息。

为了向会话发送消息，请使用 ServiceBusClient 创建发件人 createSender.

发送消息时，请在消息中设置 sessionId 属性，以确保您的消息到达正确的会话。

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

const sender = serviceBusClient.createSender("my-session-queue");
await sender.sendMessages({
  body: "my-message-body",
  sessionId: "my-session",
});

您可以在此处阅读有关会话如何运作的更多信息。

从 Session 接收消息

使用会话需要您创建启用会话的 Queue 或 Subscription。您可以在此处阅读有关如何在门户中配置此功能的更多信息。

与未启用会话的 Queues 或 Subscriptions 不同，任何时候都只有一个接收器可以从会话中读取数据。这是通过锁定会话来强制实施的，该会话由服务总线处理。从概念上讲，这类似于使用 peekLock mode 时消息锁定的工作原理 - 当消息（或会话）被锁定时，您的接收者拥有对它的独占访问权限。

为了打开和锁定会话，请使用的 ServiceBusClient 实例来创建 SessionReceiver。

有两种方法可以选择要打开的会话：

sessionId指定，用于锁定命名会话。

const receiver = await serviceBusClient.acceptSession("my-session-queue", "my-session");

请勿指定会话 ID。在这种情况下，服务总线将查找下一个尚未锁定的可用会话。
```
const receiver = await serviceBusClient.acceptNextSession("my-session-queue");
```
您可以通过上的属性sessionIdSessionReceiver找到会话的名称。如果选项中未提供 receiveMode，则默认为 “peekLock” 模式。您还可以在 “peekLock” 模式下结算收到的消息。

创建接收者后，您可以选择 3 种方式来接收消息：

获取消息数组
使用消息处理程序订阅
使用 async iterator

您可以在此处阅读有关会话如何运作的更多信息。

列表消息会话

要发现队列或订阅中哪些会话有活跃消息或会话状态，请使用 listMessageSessions()：

import { DefaultAzureCredential } from "@azure/identity";
import { ServiceBusClient } from "@azure/service-bus";

const fullyQualifiedNamespace = "<name-of-service-bus-namespace>.servicebus.windows.net";
const credential = new DefaultAzureCredential();
const serviceBusClient = new ServiceBusClient(fullyQualifiedNamespace, credential);

// List all sessions with active messages or session state in a queue
for await (const sessionId of serviceBusClient.listMessageSessions("my-session-queue")) {
  console.log("Session ID:", sessionId);
}

// List sessions in a subscription
for await (const sessionId of serviceBusClient.listMessageSessions("my-topic", "my-subscription")) {
  console.log("Session ID:", sessionId);
}

// List only sessions whose stored session state was set or updated in the last seven days
const sessionStateUpdatedAfter = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000);
for await (const sessionId of serviceBusClient.listMessageSessions("my-session-queue", {
  sessionStateUpdatedAfter,
})) {
  console.log("Recently updated session ID:", sessionId);
}

管理服务总线命名空间的资源

ServiceBusAdministrationClient 允许您对实体（队列、主题和订阅）和订阅的规则执行 CRUD作来管理命名空间。

支持使用服务总线连接字符串以及类似于 @azure/identityServiceBusClient的 AAD 凭据进行身份验证。

注意：服务总线尚不支持为命名空间设置 CORS 规则，因此 ServiceBusAdministrationClient 如果不禁用 Web 安全性，则无法在浏览器中工作。有关更多信息，请参阅此处。

import { ServiceBusAdministrationClient } from "@azure/service-bus";

const queueName = "my-session-queue";

// Get the connection string from the portal
// OR
// use the token credential overload, provide the host name of your Service Bus instance and the AAD credentials from the @azure/identity library
const serviceBusAdministrationClient = new ServiceBusAdministrationClient("<connectionString>");

// Similarly, you can create topics and subscriptions as well.
const createQueueResponse = await serviceBusAdministrationClient.createQueue(queueName);
console.log("Created queue with name - ", createQueueResponse.name);

const queueRuntimeProperties =
  await serviceBusAdministrationClient.getQueueRuntimeProperties(queueName);
console.log(`Number of messages in the queue = ${queueRuntimeProperties.totalMessageCount}`);

// Topic runtime properties additionally report the total number of SQL and correlation filters
// across all of the topic's subscriptions. These counts are served by the 2024-05 service API
// version and later; on an older version they are `undefined`.
const topicName = "my-topic";
const subscriptionName = "my-subscription";
await serviceBusAdministrationClient.createTopic(topicName);
// A new subscription carries a default rule with a SQL TrueFilter. Adding a correlation rule
// gives the topic one of each, so the counts below aggregate across the subscription's rules.
await serviceBusAdministrationClient.createSubscription(topicName, subscriptionName);
await serviceBusAdministrationClient.createRule(
  topicName,
  subscriptionName,
  "my-correlation-rule",
  { correlationId: "my-correlation-id" },
);
const topicRuntimeProperties =
  await serviceBusAdministrationClient.getTopicRuntimeProperties(topicName);
console.log(`SQL filter count = ${topicRuntimeProperties.sqlFilterCount}`);
console.log(`Correlation filter count = ${topicRuntimeProperties.correlationFilterCount}`);

await serviceBusAdministrationClient.deleteTopic(topicName);
await serviceBusAdministrationClient.deleteQueue(queueName);

参考示例 - administrationClient.ts

Troubleshooting

以下是开始诊断问题的一些初始步骤。有关更多信息，请参阅服务总线故障排除指南。

AMQP 依赖项

服务总线库依赖于 rhea-promise 库来管理连接、通过 AMQP 协议发送和接收消息。

Logging

可以使用此库设置以下环境变量来获取调试日志。

从服务总线 SDK 获取调试日志

export DEBUG=azure*

从服务总线 SDK 和协议级库获取调试日志。

export DEBUG=azure*,rhea*

如果您对 查看消息转换（ 占用大量控制台/磁盘空间）不感兴趣，则可以按如下方式设置 DEBUG 环境变量：

export DEBUG=azure*,rhea*,-rhea:raw,-rhea:message,-azure:core-amqp:datatransformer

如果您只对错误感兴趣，则可以按如下方式设置 DEBUG 环境变量：

export DEBUG=azure:service-bus:error,azure:core-amqp:error,rhea-promise:error,rhea:events,rhea:frames,rhea:io,rhea:flow

登录到文件

如上所示设置 DEBUG 环境变量
按如下方式运行测试脚本：

测试脚本中的日志记录语句转到 out.log，sdk 中的日志记录语句将转到 debug.log。
```
node your-test-script.js > out.log 2>debug.log
```
通过将 stderr 重定向到 stdout（&1），然后将 stdout 重定向到文件，将测试脚本和 sdk 中的日志记录语句转到同一文件 out.log：
```
node your-test-script.js >out.log 2>&1
```
测试脚本中的日志记录语句，sdk 将转到同一文件 out.log。
```
  node your-test-script.js &> out.log
```

后续步骤

请查看示例目录，了解有关如何使用此库向服务总线队列、主题和订阅发送消息和从服务总线队列、主题和订阅发送消息的详细示例。

Contributing

如果你想为本库贡献内容，请阅读贡献指南，了解更多关于如何构建和测试代码的信息。

反馈

此页面是否有帮助？

上次更新时间为 2026-10-06