Azure 服务 JavaScript 的巴士客户端库
Azure 服务巴士 是一种来自微软的高度可靠的云信息传输服务.
使用客户端库 @azure/service-bus 在您的应用程序中
- 向 Azure 服务总线队列或主题发送消息
- 从 Azure 服务总线队列或订阅接收信件
- 在 Azure 服务总线命名空间内创建/Get/Delete/Update/List 队列/Topics/订阅/规则.
资源 @azure/service-bus 版本 7 :
密钥链接 :
页:1: 如果您正在使用1.1.10或更低的版本, 并想要迁移到最新版本 这个包裹请看看我们 从服务巴士V1到服务巴士V7的迁移指南
开始
安装软件包
使用 npm 为 Azure 服务巴士客户端库安装最新版本 .
npm install @azure/service-bus
目前得到支持的环境
先决条件
配置类型脚本
TypeScript 用户需要安装节点类型定义 :
npm install @types/node
您还需要启用 compilerOptions.allowSyntheticDefaultImports 在你的Tsconfig.json中。 请注意, 如果您已启用 compilerOptions.esModuleInterop, (中文(简体) ), allowSyntheticDefaultImports 默认启用。 见 TypeScript 的编译器选项手册 更多信息.
Java 脚本套装
要在浏览器中使用此客户端库,首先需要使用捆绑器. 关于如何做到这一点的详情,请参见我们 捆绑文件。 。 。 .
除了上面描述的内容外,这个库还需要为以下节点JS核心内置模块增加多填充,以便在浏览器中正常工作: 1
bufferospathprocess
装入 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"),
+ },
},
与滚滚相撞
如果您正在使用 Rollup 捆绑器, 请安装以下依赖性
npm install --save-dev @rollup/plugin-commonjs @rollup/plugin-inject @rollup/plugin-node-resolve
然后在您的滚动中包含以下内容
+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 Introduct不支持这个SDK库使用的一些JavaScript api,所以需要为它们提供多填充. 请看 以 Expo 发布原始样本 更多细节.
认证客户端
与服务总线的互动始于实例 服务快捷键 课。 你们可以 使用连接字符串或使用 Azure Active Directory 证书认证到服务总线 .
使用连接字符串
这种方法将连接字符串带入您的服务总线实例 。 你可以来 azure 门户网站上的连接字符串.
import { ServiceBusClient } from "@azure/service-bus";
const serviceBusClient = new ServiceBusClient("<connectionString>");
有关此构建器的更多信息,请访问 API 文档。 。 。 .
使用 Azure 活动目录证书
使用 Azure 活动目录进行认证 Azure 身份库。 。 。 .
下面的例子使用 默认标记,许多
现有证书提供者 @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, 然后设置“ 镜片” , 用于服务车到以下 适当的标志 :
["https://servicebus.azure.net//user_impersonation"];
有关此构建器的更多信息,请访问 API 文档
关键概念
一旦你刚开始 ServiceBusClient中,您可以在
服务总线命名空间 :
- 队列:允许发送和接收消息. 常用于点对点通信.
- 专题: 相对于"队列","主题"更适合发布/订阅情景. 一个话题可以发送到,但需要订阅,其中可以有多个平行的订阅,从中消耗.
- 订阅: 从主题中消耗的机制。 每个订阅都是独立的,收到发送到该主题的每封信件的副本. 规则和过滤器可以用来裁剪特定订阅收到哪些信件.
关于这些资源的更多信息,请参见: Azure服务巴士是什么?。 。 。 .
为了与这些资源互动,应当熟悉以下SDK概念:
- 使用
ServiceBusSender使用ServiceBusClient.createSender()。 。 。 . - 使用
ServiceBusReceiver使用ServiceBusClient.createReceiver()。 。 。 . - 从会话允许的队列或订阅接收信件,使用
ServiceBusSessionReceiver使用ServiceBusClient.acceptSession()或者说ServiceBusClient.acceptNextSession()。 。 。 .
请注意,在使用该图书馆之前,应先创建队列、主题和订阅.
实例
以下各节提供代码片段,以使用 Azure Service Bus 来覆盖一些常见的任务
发送信件
一旦你创造了一个实例 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" - 在"偷看Lock"模式中,接收器在队列指定的时间段对消息有锁.
- "收发AndDelete"——在收发AndDelete模式中,消息会随即从服务巴士上被删除.
如果选项中不提供接收模式,则默认为"peekLock"模式. 你也可以这样 处理信件 以"peekLock"模式收到.
您可以以三种方式之一使用此接收器接收信件 :
获取一系列消息
使用 接收信数 函数返回承诺 解决一系列消息.
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);
使用信件处理器订阅
使用 订阅 设置信件处理器并拥有 它运行,只要你需要.
完事了就打电话 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,
});
使用同步执行器
使用 获取编辑器 获取消息的同步转发器
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() 以您想要如何解析消息为基础.
要多学习,请读 正在处理已收信件
死信队列
死信队列是 子队列。每个队列或订阅都有自己的死信队列。 死信队列存储
信件已明确写入 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}`);
}
完整样本更彻底地显示已死字母队列:
使用会话发送消息
使用会话需要您创建启用的会话队列或订阅. 你们可以 阅读更多关于如何在门户网站中配置此功能的内容 这里。 。 。 .
为了向会话发送消息,请使用 ServiceBusClient 使用
创建种子。 。 。 .
当发送消息时,设置 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",
});
你可以读到更多关于课程如何运作的信息 这里。 。 。 .
从会话接收信件
使用会话需要您创建启用的会话队列或订阅. 你们可以 阅读更多关于如何在门户网站中配置此功能的内容 这里。 。 。 .
与非会话启用的队列或订阅不同,只有一个接收器
可以在任何时间从会话中读取。 这是由 锁定 一届会议,
由服务巴士处理 从概念上来说,这类似于信息
使用时锁定工作 peekLock 模式 - 当一个消息(或会话)是
锁上你的接收器 独家访问它.
为了打开和锁定会话,请使用 ServiceBusClient 创建 会话接收者。 。 。 .
选择哪个会场开幕有两种方式:
-
指定 a
sessionId,它锁定一个命名的会话.const receiver = await serviceBusClient.acceptSession("my-session-queue", "my-session"); -
不指定会话 ID 。 在此情况下,服务巴士会找到下一个可用的会话 还没有锁定.
const receiver = await serviceBusClient.acceptNextSession("my-session-queue");您可以通过
sessionId财产SessionReceiver。 。 。 . 如果选项中不提供接收模式,则默认为"peekLock"模式. 你也可以这样 处理信件 以"peekLock"模式收到.
一旦创建了接收器,您可以选择三种方式接收消息:
你可以读到更多关于课程如何运作的信息 这里。 。 。 .
列表消息会话
要发现队列或订阅中哪些会话有活动消息或会话状态, 请使用 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 操作管理命名空间.
- 支持使用服务总线连接字符串以及来自 AAD 的认证
@azure/identity类似于ServiceBusClient。 。 。 .
注:服务 巴士不支持设置 CORS 规则命名空间,因此 ServiceBusAdministrationClient 在浏览器中工作时不会损坏网络安全。 关于更多信息,参见 这里。 。 。 .
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);
- 参考样本 - 行政单位
解决问题
以下是开始诊断问题的初步步骤. 详情请查阅 服务巴士解决问题指南。 。 。 .
AMQP 依赖关系
服务巴士库取决于 rhea - 保证 用于管理连接、发送和接收信件的库 AMQP 调试 协议.
日志
您可以在此库中设置以下环境变量以获取调试日志 .
- 从服务巴士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.logsdk 的日志语句debug.log。 。 。 .node your-test-script.js > out.log 2>debug.log - 从测试脚本记录语句, sdk 转到同一文件
out.log将 stderr 重定向为 stdout( & 1), 然后将 stdout 重定向为文件 :node your-test-script.js >out.log 2>&1 - 从测试脚本记录语句, sdk 转到同一文件
out.log。 。 。 .node your-test-script.js &> out.log
今后的步骤
请看一下 样本 关于如何使用此库发送和接收信件的详细示例的目录 服务巴士队列、主题和订阅。 。 。 .
捐款
如果您想为图书馆捐钱 请读一下 贡献指南 学习如何构建和测试密码.