@azure/storage-queue (TypeScript/JavaScript)

SDK pour les opérations de stockage de files d'attente Azure — envoyer, recevoir, consulter et gérer les messages dans les files d'attente.

Installation

npm install @azure/storage-queue @azure/identity

Version actuelle : 12.x
Node.js : >= 18.0.0

Variables d'environnement

AZURE_STORAGE_ACCOUNT_NAME=<account-name>
AZURE_STORAGE_ACCOUNT_KEY=<account-key>
# OU chaîne de connexion
AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=...
AZURE_TOKEN_CREDENTIALS=prod # Requis uniquement si DefaultAzureCredential est utilisé en production

Authentification

Identifiants de jeton Microsoft Entra (Recommandé)

import { QueueServiceClient } from "@azure/storage-queue";
import { DefaultAzureCredential, ManagedIdentityCredential } from "@azure/identity";

// Dev local : DefaultAzureCredential. Production : définir AZURE_TOKEN_CREDENTIALS=prod ou AZURE_TOKEN_CREDENTIALS=<specific_credential>
const credential = new DefaultAzureCredential({requiredEnvVars: ["AZURE_TOKEN_CREDENTIALS"]});
// Ou utiliser un identifiant spécifique directement en production :
// Voir https://learn.microsoft.com/javascript/api/overview/azure/identity-readme?view=azure-node-latest#credential-classes
// const credential = new ManagedIdentityCredential();

const accountName = process.env.AZURE_STORAGE_ACCOUNT_NAME!;
const client = new QueueServiceClient(
  `https://${accountName}.queue.core.windows.net`,
  credential
);

Chaîne de connexion

import { QueueServiceClient } from "@azure/storage-queue";

const client = QueueServiceClient.fromConnectionString(
  process.env.AZURE_STORAGE_CONNECTION_STRING!
);

StorageSharedKeyCredential (Node.js uniquement)

import { QueueServiceClient, StorageSharedKeyCredential } from "@azure/storage-queue";

const accountName = process.env.AZURE_STORAGE_ACCOUNT_NAME!;
const accountKey = process.env.AZURE_STORAGE_ACCOUNT_KEY!;

const sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);
const client = new QueueServiceClient(
  `https://${accountName}.queue.core.windows.net`,
  sharedKeyCredential
);

Jeton SAS

import { QueueServiceClient } from "@azure/storage-queue";

const accountName = process.env.AZURE_STORAGE_ACCOUNT_NAME!;
const sasToken = process.env.AZURE_STORAGE_SAS_TOKEN!;

const client = new QueueServiceClient(
  `https://${accountName}.queue.core.windows.net${sasToken}`
);

Hiérarchie des clients

QueueServiceClient (niveau compte)
└── QueueClient (niveau file d'attente)
    └── Messages (envoyer, recevoir, consulter, supprimer)

Opérations de file d'attente

Créer une file d'attente

const queueClient = client.getQueueClient("my-queue");
await queueClient.create();

// Ou créer si elle n'existe pas
await queueClient.createIfNotExists();

Lister les files d'attente

for await (const queue of client.listQueues()) {
  console.log(queue.name);
}

// Avec filtre de préfixe
for await (const queue of client.listQueues({ prefix: "task-" })) {
  console.log(queue.name);
}

Supprimer une file d'attente

await queueClient.delete();

// Ou supprimer si elle existe
await queueClient.deleteIfExists();

Obtenir les propriétés de la file d'attente

const properties = await queueClient.getProperties();
console.log("Approximate message count:", properties.approximateMessagesCount);
console.log("Metadata:", properties.metadata);

Définir les métadonnées de la file d'attente

await queueClient.setMetadata({
  department: "engineering",
  priority: "high",
});

Opérations sur les messages

Envoyer un message

const queueClient = client.getQueueClient("my-queue");

// Message simple
await queueClient.sendMessage("Hello, World!");

// Avec options
await queueClient.sendMessage("Delayed message", {
  visibilityTimeout: 60, // Masqué pendant 60 secondes
  messageTimeToLive: 3600, // Expire dans 1 heure
});

// Message JSON (doit être une chaîne)
const task = { type: "process", data: { id: 123 } };
await queueClient.sendMessage(JSON.stringify(task));

Recevoir des messages

// Recevoir jusqu'à 32 messages (par défaut : 1)
const response = await queueClient.receiveMessages({
  numberOfMessages: 10,
  visibilityTimeout: 30, // 30 secondes pour traiter
});

for (const message of response.receivedMessageItems) {
  console.log("Message ID:", message.messageId);
  console.log("Content:", message.messageText);
  console.log("Dequeue Count:", message.dequeueCount);
  console.log("Pop Receipt:", message.popReceipt);

  // Traiter le message...

  // Supprimer après traitement
  await queueClient.deleteMessage(message.messageId, message.popReceipt);
}

Consulter les messages

Consulter sans supprimer de la file d'attente (sans délai de visibilité).

const response = await queueClient.peekMessages({
  numberOfMessages: 5,
});

for (const message of response.peekedMessageItems) {
  console.log("Message ID:", message.messageId);
  console.log("Content:", message.messageText);
  // Remarque : Pas de popReceipt - impossible de supprimer les messages consultés
}

Mettre à jour un message

Étendre le délai de visibilité ou mettre à jour le contenu.

// Recevoir un message
const response = await queueClient.receiveMessages();
const message = response.receivedMessageItems[0];

if (message) {
  // Mettre à jour le contenu et étendre la visibilité
  const updateResponse = await queueClient.updateMessage(
    message.messageId,
    message.popReceipt,
    "Updated content",
    60 // Nouveau délai de visibilité en secondes
  );

  // Utiliser le nouveau popReceipt pour les opérations suivantes
  console.log("New pop receipt:", updateResponse.popReceipt);
}

Supprimer un message

// Après réception
const response = await queueClient.receiveMessages();
const message = response.receivedMessageItems[0];

if (message) {
  await queueClient.deleteMessage(message.messageId, message.popReceipt);
}

Effacer tous les messages

await queueClient.clearMessages();

Modèles de traitement des messages

Modèle de travailleur de base

async function processQueue(queueClient: QueueClient): Promise<void> {
  while (true) {
    const response = await queueClient.receiveMessages({
      numberOfMessages: 10,
      visibilityTimeout: 30,
    });

    if (response.receivedMessageItems.length === 0) {
      // Aucun message, attendre avant de réinterroger
      await sleep(5000);
      continue;
    }

    for (const message of response.receivedMessageItems) {
      try {
        await processMessage(message.messageText);
        await queueClient.deleteMessage(message.messageId, message.popReceipt);
      } catch (error) {
        console.error(`Failed to process message ${message.messageId}:`, error);
        // Le message redevient visible après le délai d'attente
      }
    }
  }
}

async function processMessage(content: string): Promise<void> {
  const task = JSON.parse(content);
  // Traiter la tâche...
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

Gestion des messages empoisonnés

const MAX_DEQUEUE_COUNT = 5;

async function processWithPoisonHandling(
  queueClient: QueueClient,
  poisonQueueClient: QueueClient
): Promise<void> {
  const response = await queueClient.receiveMessages({
    numberOfMessages: 10,
    visibilityTimeout: 30,
  });

  for (const message of response.receivedMessageItems) {
    if (message.dequeueCount > MAX_DEQUEUE_COUNT) {
      // Déplacer vers la file d'attente de poison
      await poisonQueueClient.sendMessage(message.messageText);
      await queueClient.deleteMessage(message.messageId, message.popReceipt);
      console.log(`Moved message ${message.messageId} to poison queue`);
      continue;
    }

    try {
      await processMessage(message.messageText);
      await queueClient.deleteMessage(message.messageId, message.popReceipt);
    } catch (error) {
      console.error(`Processing failed (attempt ${message.dequeueCount}):`, error);
    }
  }
}

Traitement par lot avec extension de visibilité

async function processBatchWithExtension(queueClient: QueueClient): Promise<void> {
  const response = await queueClient.receiveMessages({
    numberOfMessages: 1,
    visibilityTimeout: 60,
  });

  const message = response.receivedMessageItems[0];
  if (!message) return;

  let popReceipt = message.popReceipt;

  // Démarrer le minuteur d'extension de visibilité
  const extensionInterval = setInterval(async () => {
    try {
      const updateResponse = await queueClient.updateMessage(
        message.messageId,
        popReceipt,
        message.messageText,
        60 // Étendre de 60 secondes supplémentaires
      );
      popReceipt = updateResponse.popReceipt;
    } catch (error) {
      console.error("Failed to extend visibility:", error);
    }
  }, 45000); // Étendre tous les 45 secondes

  try {
    await longRunningProcess(message.messageText);
    await queueClient.deleteMessage(message.messageId, popReceipt);
  } finally {
    clearInterval(extensionInterval);
  }
}

Codage des messages

Par défaut, les messages sont codés en Base64. Vous pouvez personnaliser cela :

import { QueueClient } from "@azure/storage-queue";

// Encodeur/décodeur personnalisé pour le texte brut
const queueClient = new QueueClient(
  `https://${accountName}.queue.core.windows.net/my-queue`,
  credential,
  {
    messageEncoding: "text", // "base64" (par défaut) ou "text"
  }
);

// Ou avec un encodeur personnalisé
const customQueueClient = new QueueClient(
  `https://${accountName}.queue.core.windows.net/my-queue`,
  credential,
  {
    messageEncoding: {
      encode: (message: string) => Buffer.from(message).toString("base64"),
      decode: (message: string) => Buffer.from(message, "base64").toString(),
    },
  }
);

Génération de jetons SAS (Node.js uniquement)

Générer un SAS de file d'attente

import {
  QueueSASPermissions,
  generateQueueSASQueryParameters,
  StorageSharedKeyCredential,
} from "@azure/storage-queue";

const sharedKeyCredential = new StorageSharedKeyCredential(accountName, accountKey);

const sasToken = generateQueueSASQueryParameters(
  {
    queueName: "my-queue",
    permissions: QueueSASPermissions.parse("raup"), // lire, ajouter, mettre à jour, traiter
    startsOn: new Date(),
    expiresOn: new Date(Date.now() + 3600 * 1000), // 1 heure
  },
  sharedKeyCredential
).toString();

const sasUrl = `https://${accountName}.queue.core.windows.net/my-queue?${sasToken}`;

Générer un SAS de compte

import {
  AccountSASPermissions,
  AccountSASResourceTypes,
  AccountSASServices,
  generateAccountSASQueryParameters,
} from "@azure/storage-queue";

const sasToken = generateAccountSASQueryParameters(
  {
    services: AccountSASServices.parse("q").toString(), // file d'attente
    resourceTypes: AccountSASResourceTypes.parse("sco").toString(),
    permissions: AccountSASPermissions.parse("rwdlacupi"),
    expiresOn: new Date(Date.now() + 24 * 3600 * 1000),
  },
  sharedKeyCredential
).toString();

Gestion des erreurs

import { RestError } from "@azure/storage-queue";

try {
  await queueClient.sendMessage("test");
} catch (error) {
  if (error instanceof RestError) {
    switch (error.statusCode) {
      case 404:
        console.log("Queue not found");
        break;
      case 400:
        console.log("Bad request - message too large or invalid");
        break;
      case 403:
        console.log("Access denied");
        break;
      case 409:
        console.log("Queue already exists or being deleted");
        break;
      default:
        console.error(`Storage error ${error.statusCode}: ${error.message}`);
    }
  }
  throw error;
}

Référence des types TypeScript

import {
  // Clients
  QueueServiceClient,
  QueueClient,

  // Authentification
  StorageSharedKeyCredential,
  AnonymousCredential,

  // SAS
  QueueSASPermissions,
  AccountSASPermissions,
  AccountSASServices,
  AccountSASResourceTypes,
  generateQueueSASQueryParameters,
  generateAccountSASQueryParameters,

  // Messages
  DequeuedMessageItem,
  PeekedMessageItem,
  QueueSendMessageResponse,
  QueueReceiveMessageResponse,
  QueueUpdateMessageResponse,

  // File d'attente
  QueueItem,
  QueueGetPropertiesResponse,

  // Erreurs
  RestError,
} from "@azure/storage-queue";

Limites des messages

Bonnes pratiques

1. Utiliser DefaultAzureCredential pour le dev local ; utiliser ManagedIdentityCredential ou WorkloadIdentityCredential pour la production
2. Toujours supprimer après traitement — Prévenir le traitement en doublon
3. Gérer les messages empoisonnés — Déplacer les messages qui ont échoué vers une file d'attente de lettres mortes
4. Utiliser un délai de visibilité approprié — Définir en fonction du temps de traitement attendu
5. Étendre la visibilité pour les tâches longues — Mettre à jour le message pour prévenir le délai d'attente
6. Utiliser JSON pour les données structurées — Sérialiser les objets en chaînes JSON
7. Vérifier dequeueCount — Détecter les messages qui échouent régulièrement
8. Utiliser la réception par lot — Recevoir plusieurs messages pour plus d'efficacité

Différences entre les plates-formes