azure-servicebus-ts

TypeScript 用 Azure Service Bus SDK

キュー、トピック、サブスクリプションを使用したエンタープライズメッセージングです。

インストール

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

配信上の境界

サンプルではキューを変更し、個別のコード断片では同じ名前を再利用します。メッセージの送信、スケジュール設定、処理完了には、対象の名前空間およびエンティティに対する権限が必要です。処理を完了する前に、ビジネス処理を冪等にしてください。ピーク・ロックでは、失敗後にメッセージが再配信される可能性があります。非公開のメッセージ本文をログに記録しないようにし、必要に応じてロックを更新し、シャットダウン時に共有クライアントを閉じてください。

基本的なワークフロー

キューへのメッセージ送信

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 の処理、ロックの更新、正常なシャットダウン

使用するタイミング

このスキルは、概要で説明されているワークフローまたはアクションを実行する場合に適用できます。

制限事項

• このスキルは、タスクが上記の説明範囲に明確に一致する場合にのみ使用してください。
• 出力を、環境固有の検証、テスト、または専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中断して確認してください。