Azure Service Bus client library for JavaScript - version 7.10.0

Azure Service Bus to wysoce niezawodna usługa obsługi wiadomości w chmurze firmy Microsoft.

Użyj biblioteki @azure/service-bus klienta w aplikacji, aby

  • Wysyłanie komunikatów do kolejki lub tematu usługi Azure Service Bus
  • Odbieranie komunikatów z kolejki lub subskrypcji usługi Azure Service Bus
  • Tworzenie/pobieranie/usuwanie/aktualizowanie/wyświetlanie listy kolejek/tematów/subskrypcji/reguł w przestrzeni nazw usługi Azure Service Bus.

Zasoby dla @azure/service-bus wersji 7:

Kluczowe linki:

UWAGA: Jeśli używasz wersji 1.1.10 lub starszej i chcesz przeprowadzić migrację do najnowszej wersji tego pakietu, zapoznaj się z naszym przewodnikiem migracji, aby przejść z Service Bus V1 do Service Bus V7

Rozpoczęcie pracy

Instalowanie pakietu

Zainstaluj najnowszą wersję biblioteki klienta Azure Service Bus przy użyciu narzędzia npm.

npm install @azure/service-bus

Obecnie obsługiwane środowiska

  • Wersje LTS systemu Node.js

Prerequisites

Konfigurowanie języka TypeScript

Użytkownicy języka TypeScript muszą mieć zainstalowane definicje typu węzła:

npm install @types/node

Należy również włączyć compilerOptions.allowSyntheticDefaultImports w tsconfig.json. Należy pamiętać, że jeśli włączono compilerOptions.esModuleInterop, allowSyntheticDefaultImports jest domyślnie włączona. Aby uzyskać więcej informacji, zobacz podręcznik opcje kompilatora języka TypeScript.

Pakiet JavaScript

Aby użyć tej biblioteki klienta w przeglądarce, najpierw należy użyć pakietu. Aby uzyskać szczegółowe informacje o tym, jak to zrobić, zapoznaj się z naszą dokumentacją dotyczącą tworzenia pakietów .

Oprócz opisanych w tym miejscu elementów ta biblioteka wymaga również dodatkowych poliwypełnień dla następujących wbudowanych modułów nodeJS, aby działały prawidłowo w przeglądarkach:

  • buffer
  • os
  • path
  • process

Tworzenie pakietów zbiorczych za pomocą pakietu WebPack

Jeśli używasz pakietu WebPack w wersji 5, możesz zainstalować następujące zależności deweloperskie

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

następnie dodaj następujące elementy do 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"),
+    },
   },

Łączenie z pakietem zbiorczym

Jeśli używasz pakietu zbiorczego, zainstaluj następujące zależności deweloperskie

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

Następnie uwzględnij następujące elementy w 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"],
+    }),
  ]
};

Aby uzyskać więcej informacji na temat używania polyfills, zapoznaj się z dokumentacją ulubionego pakietu.

Obsługa natywna platformy React

Podobnie jak przeglądarki, React Native nie obsługuje niektórych interfejsów API JavaScript używanych przez tę bibliotekę SDK, więc musisz dostarczyć dla nich polyfille. Aby uzyskać więcej informacji, zobacz przykład Messaging React Native z Expo.

Uwierzytelnianie klienta

Interakcja z usługą Service Bus rozpoczyna się od wystąpienia klasy ServiceBusClient . Uwierzytelnianie w usłudze Service Bus można przeprowadzić przy użyciu parametrów połączenia lub poświadczeń usługi Azure Active Directory.

Używanie ciągu połączenia

Ta metoda przyjmuje parametry połączenia do wystąpienia usługi Service Bus. Parametry połączenia można pobrać z witryny Azure Portal.

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

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

Więcej informacji na temat tego konstruktora dostępnych jest w dokumentacji API.

Używanie poświadczeń usługi Azure Active Directory

Uwierzytelnianie za pomocą usługi Azure Active Directory korzysta z biblioteki tożsamości platformy Azure.

W poniższym przykładzie użyto DefaultAzureCredential, jednego z wielu dostępnych dostawców poświadczeń z biblioteki @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);

UWAGA: Jeśli używasz własnej implementacji TokenCredential interfejsu względem usługi AAD, ustaw wartość "zakresy" dla usługi Service-Bus na następującą, aby uzyskać odpowiedni token:

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

Więcej informacji na temat tego konstruktora dostępnych jest w dokumentacji API

Podstawowe pojęcia

Po zainicjowaniu ServiceBusClientprogramu można wchodzić w interakcje z tymi zasobami w przestrzeni nazw usługi Service Bus:

  • Kolejki: Umożliwia wysyłanie i odbieranie wiadomości. Często używany do komunikacji punkt-punkt.
  • Tematy: W przeciwieństwie do kolejek, tematy lepiej nadają się do scenariuszy publikowania/subskrybowania. Temat może zostać wysłany do, ale wymaga subskrypcji, której może być wiele równolegle, z której można korzystać.
  • Subskrypcje: mechanizm korzystania z tematu. Każda subskrypcja jest niezależna i otrzymuje kopię każdej wiadomości wysłanej do tematu. Reguły i filtry mogą służyć do dostosowywania, które wiadomości są odbierane przez określoną subskrypcję.

Aby uzyskać więcej informacji na temat tych zasobów, zobacz Co to jest Azure Service Bus?.

Aby wchodzić w interakcje z tymi zasobami, należy zapoznać się z następującymi pojęciami dotyczącymi zestawu SDK:

Należy pamiętać, że kolejki, tematy i subskrypcje należy utworzyć przed użyciem tej biblioteki.

Examples

Poniższe sekcje zawierają fragmenty kodu, które obejmują niektóre typowe zadania przy użyciu Azure Service Bus

Wysyłanie komunikatów

Po utworzeniu instancji ServiceBusClientServiceBusSender klasy możesz ją uzyskać za pomocą metody createSender , której możesz używać do wysyłania wiadomości.

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

Odbieranie komunikatów

Po utworzeniu instancji ServiceBusClient klasy można ją pobrać za ServiceBusReceiver pomocą metody createReceiver .

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

Dostępne są dwa receiveModes.

  • "peekLock" — w trybie peekLock odbiornik ma blokadę komunikatu na czas określony w kolejce.
  • "receiveAndDelete" — w trybie receiveAndDelete komunikaty są usuwane z usługi Service Bus w miarę ich odbierania.

Jeśli receiveMode nie jest podany w opcjach, domyślnie jest to tryb "peekLock". Możesz również rozliczyć otrzymane wiadomości w trybie "peekLock".

Możesz użyć tego odbiornika na jeden z 3 sposobów odbierania wiadomości:

Pobieranie szeregu wiadomości

Użyj receiveMessages funkcji, która zwraca obietnicę, która jest rozpoznawana jako tablica komunikatów.

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

Subskrybowanie przy użyciu programu obsługi wiadomości

Użyj metody subskrypcji , aby skonfigurować programy obsługi komunikatów i uruchomić je tak długo, jak potrzebujesz.

Kiedy skończysz, zadzwoń receiver.close() , aby przestać otrzymywać więcej wiadomości.

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

Korzystanie z iteratora asynchronicznego

Użyj getMessageIterator, aby uzyskać iterator asynchroniczny dla komunikatów

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
}

Rozliczanie wiadomości

Po otrzymaniu wiadomości możesz zadzwonić do completeMessage(), abandonMessage()lub deferMessage()deadLetterMessage() do odbiorcy w zależności od tego, jak chcesz uregulować wiadomość.

Aby dowiedzieć się więcej, przeczytaj artykuł Rozliczanie otrzymanych wiadomości

Kolejki wiadomości nie do dostarczenia

Kolejka utraconych wiadomości jest kolejką podrzędną. Każda kolejka lub subskrypcja ma własną kolejkę utraconych wiadomości. Kolejki utraconych wiadomości przechowują wiadomości, które zostały jawnie zapisane jako aktywne (za pośrednictwem receiver.deadLetterMessage()) lub wiadomości, które przekroczyły maksymalną liczbę dostarczeń.

Tworzenie odbiornika dla podkolejki utraconych wiadomości jest podobne do tworzenia odbiornika dla subskrypcji lub kolejki:

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

Pełne próbki pokazujące bardziej szczegółowe kolejki utraconych wiadomości:

Wysyłanie wiadomości przy użyciu sesji

Korzystanie z sesji wymaga utworzenia kolejki lub subskrypcji z włączoną sesją. Więcej informacji na temat konfigurowania tej funkcji w portalu można znaleźć tutaj.

Aby wysyłać wiadomości do sesji, użyj przycisku ServiceBusClient , aby utworzyć nadawcę za pomocą createSender.

Podczas wysyłania wiadomości ustaw właściwość sessionId w wiadomości, aby upewnić się, że wiadomość trafi do właściwej sesji.

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

Więcej o tym, jak działają sesje, przeczytasz tutaj.

Odbieranie wiadomości z sesji

Korzystanie z sesji wymaga utworzenia kolejki lub subskrypcji z włączoną sesją. Więcej informacji na temat konfigurowania tej funkcji w portalu można znaleźć tutaj.

W przeciwieństwie do kolejek lub subskrypcji, które nie obsługują sesji, tylko jeden odbiornik może odczytywać dane z sesji w dowolnym momencie. Jest to wymuszane przez zablokowanie sesji, która jest obsługiwana przez Service Bus. Koncepcyjnie jest to podobne do tego, jak działa blokowanie wiadomości podczas korzystania z peekLock trybu - gdy wiadomość (lub sesja) jest zablokowana, odbiorca ma do niej wyłączny dostęp.

Aby otworzyć i zablokować sesję, użyj wystąpienia , ServiceBusClient aby utworzyć SessionReceiver.

Istnieją dwa sposoby wyboru, którą sesję chcesz otworzyć:

  1. Określ sessionIdparametr , który blokuje nazwaną sesję.

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

  2. Nie podawaj identyfikatora sesji. W takim przypadku usługa Service Bus znajdzie następną dostępną sesję, która nie jest jeszcze zablokowana.

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

    Nazwę sesji można znaleźć za pomocą sessionId właściwości w pliku .SessionReceiver Jeśli receiveMode nie jest podany w opcjach, domyślnie jest to tryb "peekLock". Możesz również rozliczyć otrzymane wiadomości w trybie "peekLock".

Po utworzeniu odbiornika możesz wybrać jeden z 3 sposobów odbierania wiadomości:

Więcej o tym, jak działają sesje, przeczytasz tutaj.

Wyświetl sesje wiadomości

Aby odkryć, które sesje mają aktywne komunikaty lub stan sesji w kolejce lub subskrypcji, użyj 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);
}

Zarządzanie zasobami przestrzeni nazw usługi Service Bus

ServiceBusAdministrationClient umożliwia zarządzanie przestrzenią nazw z operacjami CRUD na jednostkach (kolejkach, tematach i subskrypcjach) oraz na regułach subskrypcji.

  • Obsługuje uwierzytelnianie za pomocą parametrów połączenia usługi Service Bus, a także za pomocą poświadczeń usługi AAD z @azure/identity poziomu podobnego do ServiceBusClient.

Uwaga: Service Bus nie obsługuje jeszcze ustawiania reguł CORS dla przestrzeni nazw, dlatego ServiceBusAdministrationClient nie będzie działać w przeglądarce bez wyłączenia zabezpieczeń internetowych. Aby uzyskać więcej informacji, zapoznaj się tutaj.

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

Troubleshooting

Oto kilka początkowych kroków, aby rozpocząć diagnozowanie problemów. Aby uzyskać więcej informacji, zapoznaj się z przewodnikiem rozwiązywania problemów z usługą Service Bus.

Zależności protokołu AMQP

Biblioteka usługi Service Bus zależy od biblioteki rhea-promise do zarządzania połączeniami, wysyłania i odbierania komunikatów za pośrednictwem protokołu AMQP .

Logging

Możesz ustawić następującą zmienną środowiskową, aby pobrać dzienniki debugowania podczas korzystania z tej biblioteki.

  • Pobieranie dzienników debugowania z zestawu SDK usługi Service Bus
export DEBUG=azure*
  • Pobieranie dzienników debugowania z zestawu SDK usługi Service Bus i biblioteki na poziomie protokołu.
export DEBUG=azure*,rhea*
  • Jeśli nie jesteś zainteresowany przeglądaniem transformacji wiadomości (która zajmuje dużo miejsca w konsoli/na dysku), możesz ustawić zmienną środowiskową DEBUG w następujący sposób:
export DEBUG=azure*,rhea*,-rhea:raw,-rhea:message,-azure:core-amqp:datatransformer
  • Jeśli interesują Cię tylko błędy, możesz ustawić zmienną środowiskową DEBUG w następujący sposób:
export DEBUG=azure:service-bus:error,azure:core-amqp:error,rhea-promise:error,rhea:events,rhea:frames,rhea:io,rhea:flow

Rejestrowanie w pliku

  1. Ustaw zmienną środowiskową, jak pokazano DEBUG powyżej
  2. Uruchom skrypt testowy w następujący sposób:
  • Instrukcje rejestrowania ze skryptu testowego przejdź do out.log i instrukcji rejestrowania z zestawu SDK przejdź do debug.log. 
    node your-test-script.js > out.log 2>debug.log
  • Instrukcje rejestrowania ze skryptu testowego i zestawu SDK przechodzą do tego samego pliku out.log przez przekierowanie narzędzia stderr do pliku stdout (&1), a następnie przekierowanie elementu stdout do pliku: 
    node your-test-script.js >out.log 2>&1
  • Instrukcje rejestrowania ze skryptu testowego i zestawu SDK przejdź do tego samego pliku out.log. 
      node your-test-script.js &> out.log

Następne kroki

Zapoznaj się z katalogiem przykładów , aby uzyskać szczegółowe przykłady używania tej biblioteki do wysyłania i odbierania komunikatów do/z kolejek, tematów i subskrypcji usługi Service Bus.

Contributing

Jeśli chcesz przyczynić się do tej biblioteki, przeczytaj przewodnik wkład aby dowiedzieć się więcej o budowaniu i testowaniu kodu.

  • Ostatnia aktualizacja o