Rozpoczynanie pracy z usługą Azure Queue Storage przy użyciu języka F#

Artykuł
03/05/2024

Usługa Azure Queue Storage zapewnia obsługę komunikatów w chmurze między składnikami aplikacji. W przypadku projektowania aplikacji pod kątem skalowania składniki aplikacji są często rozłączane, dzięki czemu mogą być skalowane niezależnie. Usługa Queue Storage zapewnia asynchroniczne przesyłanie komunikatów na potrzeby komunikacji między składnikami aplikacji niezależnie od tego, czy działają w chmurze, na komputerze, serwerze lokalnym czy urządzeniu przenośnym. Magazyn kolejek obsługuje również zarządzanie asynchronicznymi zadaniami oraz przepływy pracy procesu kompilacji.

Informacje o tym samouczku

W tym samouczku pokazano, jak napisać kod języka F# dla niektórych typowych zadań przy użyciu usługi Azure Queue Storage. Omówione zadania obejmują tworzenie i usuwanie kolejek oraz dodawanie, odczytywanie i usuwanie komunikatów w kolejce.

Aby zapoznać się z koncepcyjnym omówieniem magazynu kolejek, zobacz przewodnik platformy .NET dotyczący magazynu kolejek.

Wymagania wstępne

Aby użyć tego przewodnika, musisz najpierw utworzyć konto usługi Azure Storage. Będziesz również potrzebować klucza dostępu do magazynu dla tego konta.

Tworzenie skryptu języka F# i uruchamianie interaktywnego języka F#

Przykłady w tym artykule mogą być używane w aplikacji języka F# lub skryptu języka F#. Aby utworzyć skrypt języka F#, utwórz plik z .fsx rozszerzeniem, na przykład queues.fsx, w środowisku deweloperów języka F#.

Jak wykonywać skrypty

Program F# Interactive, dotnet fsi, można uruchomić interaktywnie lub można go uruchomić z poziomu wiersza polecenia, aby uruchomić skrypt. Składnia wiersza polecenia to

> dotnet fsi [options] [ script-file [arguments] ]

Dodawanie pakietów w skry skrycie

Następnie użyj polecenia #rnuget:package name , aby zainstalować Azure.Storage.Queues pakiet i open przestrzenie nazw. Na przykład

> #r "nuget: Azure.Storage.Queues"
open Azure.Storage.Queues

Dodawanie deklaracji przestrzeni nazw

Dodaj następujące instrukcje open na początku pliku queues.fsx:

open Azure.Storage.Queues // Namespace for Queue storage types
open System
open System.Text

Uzyskiwanie parametrów połączenia

Na potrzeby tego samouczka będziesz potrzebować parametry połączenia usługi Azure Storage. Aby uzyskać więcej informacji na temat parametry połączenia, zobacz Konfigurowanie ciągów Połączenie ion magazynu.

W tym samouczku wprowadzisz parametry połączenia w skrycie, w następujący sposób:

let storageConnString = "..." // fill this in from your storage account

Tworzenie klienta usługi kolejki

Klasa QueueClient umożliwia pobieranie kolejek przechowywanych w usłudze Queue Storage. Oto jeden ze sposobów tworzenia klienta:

let queueClient = QueueClient(storageConnString, "myqueue")

Teraz możesz przystąpić do pisania kodu, który będzie odczytywać dane z usługi Queue Storage i zapisywać je w nim.

Utwórz kolejkę

W tym przykładzie pokazano, jak utworzyć kolejkę, jeśli jeszcze nie istnieje:

queueClient.CreateIfNotExists()

Wstawianie komunikatu do kolejki

Aby wstawić komunikat do istniejącej kolejki, najpierw utwórz nowy komunikat. Następnie wywołaj metodę SendMessage . Komunikat można utworzyć na podstawie ciągu (w formacie UTF-8) lub byte tablicy w następujący sposób:

queueClient.SendMessage("Hello, World") // Insert a String message into a queue
queueClient.SendMessage(BinaryData.FromBytes(Encoding.UTF8.GetBytes("Hello, World"))) // Insert a BinaryData message into a queue

Podgląd kolejnego komunikatu

Możesz zajrzeć do komunikatu z przodu kolejki, bez usuwania go z kolejki, wywołując metodę PeekMessage .

let peekedMessage = queueClient.PeekMessage()
let messageContents = peekedMessage.Value.Body.ToString()

Pobierz następny komunikat na potrzeby przetwarzania

Komunikat można pobrać z przodu kolejki do przetworzenia, wywołując metodę ReceiveMessage .

let updateMessage = queueClient.ReceiveMessage().Value

Później wskazujesz pomyślne przetwarzanie komunikatu przy użyciu polecenia DeleteMessage.

Zmiana zawartości komunikatu w kolejce

Zawartość pobranego komunikatu można zmienić w kolejce. Jeśli komunikat reprezentuje zadanie robocze, możesz użyć tej funkcji, aby zaktualizować stan zadania. Poniższy kod aktualizuje komunikat kolejki o nową zawartość i ustawia rozszerzenie limitu czasu widoczności o kolejne 60 sekund. Operacja ta zapisuje stan pracy powiązanej z komunikatem i daje klientowi kolejną minutę na kontynuowanie pracy nad komunikatem. Możesz użyć tej metody do śledzenia wieloetapowych przepływów pracy związanych z komunikatami kolejek, bez konieczności rozpoczynania od nowa, gdy dany etap nie powiedzie się ze względu na awarię sprzętu lub oprogramowania. Zazwyczaj należy zachować liczbę ponownych prób, a jeśli komunikat zostanie ponowiony więcej niż kilka razy, należy go usunąć. Jest to zabezpieczenie przed komunikatami, które wyzwalają błąd aplikacji zawsze, gdy są przetwarzane.

queueClient.UpdateMessage(
    updateMessage.MessageId,
    updateMessage.PopReceipt,
    "Updated contents.",
    TimeSpan.FromSeconds(60.0))

Usunięcie następnego komunikatu z kolejki

Twój kod usuwa komunikat z kolejki w dwóch etapach. Po wywołaniu ReceiveMessagemetody zostanie wyświetlony następny komunikat w kolejce. Zwrócony komunikat staje ReceiveMessage się niewidoczny dla innych kodów odczytujących komunikaty z tej kolejki. Domyślnie komunikat pozostanie niewidoczny przez 30 sekund. Aby zakończyć usuwanie komunikatu z kolejki, należy również wywołać metodę DeleteMessage. Ten dwuetapowy proces usuwania komunikatów gwarantuje, że jeśli kod nie będzie w stanie przetworzyć komunikatu z powodu awarii sprzętu lub oprogramowania, inne wystąpienie kodu będzie w stanie uzyskać ten sam komunikat i ponowić próbę. Twoje wywołania DeleteMessage kodu bezpośrednio po przetworzeniu wiadomości. Wszystkie metody kolejkowania, które pokazaliśmy do tej pory, mają Async alternatywy.

let deleteMessage = queueClient.ReceiveMessage().Value
queueClient.DeleteMessage(deleteMessage.MessageId, deleteMessage.PopReceipt)

Używanie przepływów pracy asynchronicznych z typowymi interfejsami API usługi Queue Storage

W tym przykładzie pokazano, jak używać asynchronicznego przepływu pracy z typowymi interfejsami API usługi Queue Storage.

async {
    let! exists = queueClient.CreateIfNotExistsAsync() |> Async.AwaitTask

    let! delAsyncMessage = queueClient.ReceiveMessageAsync() |> Async.AwaitTask

    // ... process the message here ...

    // Now indicate successful processing:
    queueClient.DeleteMessageAsync(delAsyncMessage.Value.MessageId, delAsyncMessage.Value.PopReceipt) |> Async.AwaitTask
}

Dodatkowe opcje usuwania komunikatów kolejkowania

Istnieją dwa sposoby dostosowania pobierania komunikatów z kolejki. Po pierwsze można uzyskać komunikaty zbiorczo (do 32). Po drugie można ustawić dłuższy lub krótszy limit czasu niewidoczności, dzięki czemu kod będzie mieć więcej lub mniej czasu na pełne przetworzenie każdego komunikatu. Poniższy przykład kodu używa ReceiveMessages metody do pobierania 20 komunikatów w jednym wywołaniu, a następnie przetwarzania każdego komunikatu. Ustawia również limitu czasu niewidoczności na pięć minut dla każdego komunikatu. 5 minut rozpoczyna się dla wszystkich komunikatów w tym samym czasie, więc po 5 minutach minęło od wywołania metody ReceiveMessages, wszystkie komunikaty, które nie zostały usunięte, staną się widoczne ponownie.

for dequeueMessage in queueClient.ReceiveMessages(20, Nullable(TimeSpan.FromMinutes(5.))).Value do
        // Process the message here.
        queueClient.DeleteMessage(dequeueMessage.MessageId, dequeueMessage.PopReceipt)

Pobieranie długości kolejki

Możesz uzyskać szacunkową liczbę komunikatów w kolejce. Metoda GetProperties prosi usługę Kolejki o pobranie atrybutów kolejki, w tym liczbę komunikatów. Właściwość ApproximateMessagesCount zwraca ostatnią wartość pobraną przez metodę GetProperties .

let properties = queueClient.GetProperties().Value
let count = properties.ApproximateMessagesCount

Usuwanie kolejki

Aby usunąć kolejkę i wszystkie zawarte w niej komunikaty, wywołaj metodę Delete w obiekcie kolejki.

queueClient.DeleteIfExists()

Uwaga

Jeśli migrujesz ze starych bibliotek, domyślnie są to komunikaty zakodowane w formacie Base64, ale nowe biblioteki nie są bardziej wydajne. Aby uzyskać informacje na temat sposobu konfigurowania kodowania, zobacz MessageEncoding.