azure-servicebus-ts

Azure Service Bus SDK for TypeScript

使用队列、主题和订阅实现企业级消息传递。

安装

npm install @azure/service-bus @azure/identity

环境变量

SERVICEBUS_NAMESPACE=<namespace>.servicebus.windows.net
SERVICEBUS_QUEUE_NAME=my-queue
SERVICEBUS_TOPIC_NAME=my-topic
SERVICEBUS_SUBSCRIPTION_NAME=my-subscription

身份验证

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

const fullyQualifiedNamespace = process.env.SERVICEBUS_NAMESPACE!;
const client = new ServiceBusClient(fullyQualifiedNamespace, new DefaultAzureCredential());

交付边界

示例会修改队列，且各个独立片段会重复使用名称。发送、安排和处理消息需要对确切的命名空间／实体具有授权。完成消息前，请确保业务处理具有幂等性；发生故障后，Peek-Lock 模式可能会重新投递消息。避免记录私有消息正文；必要时续订锁，并在关闭时关闭共享客户端。

核心工作流

向队列发送消息

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

// Single message
await sender.sendMessages({
  body: { orderId: "12345", amount: 99.99 },
  contentType: "application/json",
});

// Batch messages
const batch = await sender.createMessageBatch();
if (!batch.tryAddMessage({ body: "Message 1" }) ||
    !batch.tryAddMessage({ body: "Message 2" })) {
  throw new Error("Message batch exceeds size limit; split and retry deliberately");
}
await sender.sendMessages(batch);

await sender.close();

从队列接收消息

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

// Receive batch
const messages = await receiver.receiveMessages(10, { maxWaitTimeInMs: 5000 });
for (const message of messages) {
  console.log(`Received: ${message.body}`);
  await receiver.completeMessage(message);
}

await receiver.close();

订阅消息（事件驱动）

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

const subscription = receiver.subscribe({
  processMessage: async (message) => {
    console.log(`Processing: ${message.body}`);
    // Message auto-completed on success
  },
  processError: async (args) => {
    console.error(`Error: ${args.error}`);
  },
});

// Stop after some time
setTimeout(async () => {
  await subscription.close();
  await receiver.close();
}, 60000);

主题和订阅

// Send to topic
const topicSender = client.createSender("my-topic");
await topicSender.sendMessages({
  body: { event: "order.created", data: { orderId: "123" } },
  applicationProperties: { eventType: "order.created" },
});

// Receive from subscription
const subscriptionReceiver = client.createReceiver("my-topic", "my-subscription");
const messages = await subscriptionReceiver.receiveMessages(10);

消息会话

// Send session message
const sender = client.createSender("session-queue");
await sender.sendMessages({
  body: { step: 1, data: "First step" },
  sessionId: "workflow-123",
});

// Receive session messages
const sessionReceiver = await client.acceptSession("session-queue", "workflow-123");
const messages = await sessionReceiver.receiveMessages(10);

// Get/set session state
const state = await sessionReceiver.getSessionState();
await sessionReceiver.setSessionState(Buffer.from(JSON.stringify({ progress: 50 })));

await sessionReceiver.close();

死信处理

// Move to dead-letter
await receiver.deadLetterMessage(message, {
  deadLetterReason: "Validation failed",
  deadLetterErrorDescription: "Missing required field: orderId",
});

// Process dead-letter queue
const dlqReceiver = client.createReceiver("my-queue", { subQueueType: "deadLetter" });
const dlqMessages = await dlqReceiver.receiveMessages(10);
for (const msg of dlqMessages) {
  console.log(`DLQ Reason: ${msg.deadLetterReason}`);
  // Inspect and durably record the authorized recovery outcome first.
  // Complete only after successful recovery; logging alone must not delete DLQ evidence.
  // await dlqReceiver.completeMessage(msg);
}

计划消息

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

// Schedule for future delivery
const scheduledTime = new Date(Date.now() + 60000); // 1 minute from now
const sequenceNumber = await sender.scheduleMessages(
  { body: "Delayed message" },
  scheduledTime
);

// Cancel scheduled message
await sender.cancelScheduledMessages(sequenceNumber);

消息延迟

// Defer message for later
await receiver.deferMessage(message);

// Receive deferred message by sequence number
const deferredMessage = await receiver.receiveDeferredMessages(message.sequenceNumber!);
await receiver.completeMessage(deferredMessage[0]);

查看消息（非破坏性）

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

// Peek without removing
const peekedMessages = await receiver.peekMessages(10);
for (const msg of peekedMessages) {
  console.log(`Peeked: ${msg.body}`);
}

主要类型

import {
  ServiceBusClient,
  ServiceBusSender,
  ServiceBusReceiver,
  ServiceBusSessionReceiver,
  ServiceBusMessage,
  ServiceBusReceivedMessage,
  ProcessMessageCallback,
  ProcessErrorCallback,
} from "@azure/service-bus";

接收模式

// Peek-Lock (default) - message locked until completed/abandoned
const receiver = client.createReceiver("my-queue", { receiveMode: "peekLock" });
await receiver.completeMessage(message);   // Remove from queue
await receiver.abandonMessage(message);    // Return to queue
await receiver.deferMessage(message);      // Defer for later
await receiver.deadLetterMessage(message); // Move to DLQ

// Receive-and-Delete - message removed immediately
const receiver = client.createReceiver("my-queue", { receiveMode: "receiveAndDelete" });

最佳实践

1. 使用 Entra ID 身份验证 - 避免在生产环境中使用连接字符串
2. 重用客户端 - 只创建一次 ServiceBusClient，并在发送方／接收方之间共享
3. 关闭资源 - 完成后务必关闭发送方／接收方
4. 处理错误 - 为订阅接收方实现 processError 回调
5. 使用会话保证顺序 - 当组内消息顺序很重要时
6. 配置死信 - 务必处理 DLQ 消息
7. 批量发送 - 多条消息时使用 createMessageBatch()

参考文档

如需了解详细模式，请参阅：

• 队列与主题模式 - 队列／主题模式、会话、接收模式、消息处理
• 错误处理与可靠性 - ServiceBusError 代码、DLQ 处理、锁续订、平稳关闭

适用场景

当任务与概述中描述的工作流或操作相符时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请暂停并请求澄清。