Jak używać usługi Azure Queue Storage z poziomu języka Python

Artykuł
01/19/2023

Omówienie

W tym artykule przedstawiono typowe scenariusze korzystania z usługi Azure Queue Storage. Omówione scenariusze obejmują wstawianie, podgląd, pobieranie i usuwanie komunikatów w kolejce. Omówiony jest również kod tworzenia i usuwania kolejek.

Przykłady w tym artykule zostały napisane w języku Python i używają biblioteki klienta usługi Azure Queue Storage dla języka Python. Aby uzyskać więcej informacji na temat kolejek, zobacz sekcję Następne kroki .

Co to jest usługa Queue Storage?

Azure Queue Storage to usługa do przechowywania dużej liczby komunikatów, do której można uzyskać dostęp z dowolnego miejsca na świecie za pośrednictwem uwierzytelnionego połączenia za pomocą protokołu HTTP lub HTTPS. Pojedynczy komunikat z kolejki nie może przekraczać 64 KB, a kolejka może zawierać miliony komunikatów — maksymalnie liczbę nieprzekraczającą całkowitego limitu pojemności konta magazynu. Usługa Queue Storage jest często używana do tworzenia listy prac w celu przetwarzania asynchronicznego.

Pojęcia dotyczące usługi kolejki

Usługa Azure Queue service zawiera następujące składniki:

Składniki usługi Azure Queue Service

Konto magazynu: cały dostęp do usługi Azure Storage odbywa się przez konto magazynu. Aby uzyskać więcej informacji na temat kont magazynu, zobacz Omówienie konta magazynu.
Kolejka: kolejka zawiera zestaw komunikatów. Wszystkie komunikaty muszą być w kolejce. Pamiętaj, że nazwa kolejki może zawierać tylko małe litery. Informacje dotyczące nazewnictwa kolejek można znaleźć w temacie Naming Queues and Metadata (Nazewnictwo kolejek i metadanych).
Komunikat: komunikat w dowolnym formacie, o maksymalnym rozmiarze 64 KB. Maksymalny czas pozostawania komunikatu w kolejce wynosi 7 dni. W przypadku wersji 2017-07-29 lub nowszej maksymalny czas wygaśnięcia może być dowolną liczbą dodatnią lub -1 wskazującą, że komunikat nie wygaśnie. Jeśli ten parametr zostanie pominięty, domyślny czas wygaśnięcia wynosi siedem dni.
Format adresu URL: Kolejki są adresowalne przy użyciu następującego formatu adresu URL: http://<storage account>.queue.core.windows.net/<queue>

Następujący adres URL dotyczy kolejki w schemacie:

http://myaccount.queue.core.windows.net/incoming-orders

Tworzenie konta usługi Azure Storage

Najprościej jest utworzyć pierwsze konto usługi Azure Storage przy użyciu witryny Azure Portal. Więcej informacji można znaleźć w temacie Tworzenie konta magazynu.

Można również utworzyć konto usługi Azure Storage przy użyciu programu Azure PowerShell, interfejsu wiersza polecenia Azure lub dostawcy zasobów usługi Azure Storage dla platformy .NET.

Jeśli w tej chwili nie chcesz tworzyć konta magazynu na platformie Azure, możesz również użyć emulatora magazynu Azurite, aby uruchomić i przetestować kod w środowisku lokalnym. Aby uzyskać więcej informacji, zobacz Use the Azurite emulator for local Azure Storage development (Używanie emulatora Azurite do lokalnego programowania w usłudze Azure Storage).

Pobieranie i instalowanie zestawu SDK usługi Azure Storage dla języka Python

Zestaw SDK usługi Azure Storage dla języka Python wymaga języka Python w wersji 2.7, 3.3 lub nowszej.

Instalowanie za pośrednictwem interfejsu PyPI

Aby zainstalować za pośrednictwem indeksu pakietów języka Python (PyPI), wpisz:

pip install azure-storage-queue

Uwaga

Jeśli uaktualniasz zestaw Azure Storage SDK dla języka Python w wersji 0.36 lub starszej, odinstaluj starszy zestaw SDK przed pip uninstall azure-storage zainstalowaniem najnowszego pakietu.

Aby zapoznać się z alternatywnymi metodami instalacji, zobacz Zestaw Azure SDK dla języka Python.

Kopiowanie poświadczeń z witryny Azure Portal

Gdy przykładowa aplikacja wysyła żądanie do usługi Azure Storage, musi być autoryzowana. Aby autoryzować żądanie, dodaj poświadczenia konta magazynu do aplikacji jako parametry połączenia. Aby wyświetlić poświadczenia konta magazynu, wykonaj następujące kroki:

Zaloguj się w witrynie Azure Portal.
Odszukaj konto magazynu.
W okienku menu konta magazynu w obszarze Zabezpieczenia i sieć wybierz pozycję Klucze dostępu. W tym miejscu możesz wyświetlić klucze dostępu do konta i pełne parametry połączenia dla każdego klucza.
W okienku Klucze dostępu wybierz pozycję Pokaż klucze.
W sekcji key1 znajdź wartość Parametry połączenia . Wybierz ikonę Kopiuj do schowka , aby skopiować parametry połączenia. W następnej sekcji dodasz wartość parametrów połączenia do zmiennej środowiskowej.

Konfigurowanie parametrów połączenia magazynu

Po skopiowaniu parametrów połączenia zapisz ją w nowej zmiennej środowiskowej na komputerze lokalnym z uruchomioną aplikacją. Aby ustawić zmienną środowiskową, otwórz okno konsoli i postępuj zgodnie z instrukcjami dla systemu operacyjnego. Zastąp <yourconnectionstring> wartość rzeczywistymi parametrami połączenia.

Windows
Linux i macOS

setx AZURE_STORAGE_CONNECTION_STRING "<yourconnectionstring>"

export AZURE_STORAGE_CONNECTION_STRING="<yourconnectionstring>"

Po dodaniu zmiennej środowiskowej w systemie Windows należy uruchomić nowe wystąpienie okna poleceń.

Ponowne uruchamianie programów

Po dodaniu zmiennej środowiskowej uruchom ponownie wszystkie uruchomione programy, które będą musiały odczytać zmienną środowiskową. Na przykład uruchom ponownie środowisko projektowe lub edytor przed kontynuowaniem.

Konfigurowanie aplikacji w celu uzyskania dostępu do usługi Queue Storage

Obiekt QueueClient umożliwia pracę z kolejką. Dodaj następujący kod w górnej części dowolnego pliku języka Python, w którym chcesz programowo uzyskać dostęp do kolejki platformy Azure:

from azure.storage.queue import (
        QueueClient,
        BinaryBase64EncodePolicy,
        BinaryBase64DecodePolicy
)

import os, uuid

Pakiet os zapewnia obsługę pobierania zmiennej środowiskowej. Pakiet uuid zapewnia obsługę generowania unikatowego identyfikatora nazwy kolejki.

Tworzenie kolejki

Parametry połączenia są pobierane z ustawionej wcześniej zmiennej środowiskowej AZURE_STORAGE_CONNECTION_STRING .

Poniższy kod tworzy QueueClient obiekt przy użyciu parametrów połączenia magazynu.

# Retrieve the connection string from an environment
# variable named AZURE_STORAGE_CONNECTION_STRING
connect_str = os.getenv("AZURE_STORAGE_CONNECTION_STRING")

# Create a unique name for the queue
q_name = "queue-" + str(uuid.uuid4())

# Instantiate a QueueClient object which will
# be used to create and manipulate the queue
print("Creating queue: " + q_name)
queue_client = QueueClient.from_connection_string(connect_str, q_name)

# Create the queue
queue_client.create_queue()

Komunikaty kolejki platformy Azure są przechowywane jako tekst. Jeśli chcesz przechowywać dane binarne, skonfiguruj funkcje kodowania i dekodowania Base64 przed umieszczeniem komunikatu w kolejce.

Skonfiguruj funkcje kodowania i dekodowania base64 podczas tworzenia obiektu klienta.

# Setup Base64 encoding and decoding functions
base64_queue_client = QueueClient.from_connection_string(
                            conn_str=connect_str, queue_name=q_name,
                            message_encode_policy = BinaryBase64EncodePolicy(),
                            message_decode_policy = BinaryBase64DecodePolicy()
                        )

Wstawianie komunikatu do kolejki

Aby wstawić komunikat do kolejki, użyj send_message metody .

message = u"Hello World"
print("Adding message: " + message)
queue_client.send_message(message)

Podgląd wiadomości

Komunikaty można wyświetlać bez usuwania ich z kolejki przez wywołanie peek_messages metody . Domyślnie ta metoda zagląda do pojedynczego komunikatu.

# Peek at the first message
messages = queue_client.peek_messages()

for peeked_message in messages:
    print("Peeked message: " + peeked_message.content)

Zmiana zawartości komunikatu w kolejce

Możesz zmienić zawartość komunikatu w kolejce. Jeśli komunikat reprezentuje zadanie, możesz użyć tej funkcji, aby zaktualizować stan zadania.

Poniższy kod używa update_message metody w celu zaktualizowania komunikatu. Limit czasu widoczności jest ustawiony na 0, co oznacza, że komunikat jest wyświetlany natychmiast i zawartość jest aktualizowana.

messages = queue_client.receive_messages()
list_result = next(messages)

message = queue_client.update_message(
        list_result.id, list_result.pop_receipt,
        visibility_timeout=0, content=u'Hello World Again')

print("Updated message to: " + message.content)

Pobieranie długości kolejki

Możesz uzyskać szacunkową liczbę komunikatów w kolejce.

Metoda get_queue_properties zwraca właściwości kolejki, w tym approximate_message_count.

properties = queue_client.get_queue_properties()
count = properties.approximate_message_count
print("Message count: " + str(count))

Wynik jest przybliżony tylko dlatego, że komunikaty można dodać lub usunąć po odpowiedzi usługi na żądanie.

Dequeue messages (Dequeue messages)

Usuń komunikat z kolejki w dwóch krokach. Jeśli kod nie może przetworzyć komunikatu, ten dwuetapowy proces gwarantuje, że będzie można uzyskać ten sam komunikat i spróbować ponownie. Wywołaj połączenie delete_message po pomyślnym przetworzeniu komunikatu.

Po wywołaniu receive_messages domyślnie zostanie wyświetlony następny komunikat w kolejce. Komunikat zwrócony z receive_messages staje się niewidoczny dla każdego innego kodu 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ć delete_message.

messages = queue_client.receive_messages()

for message in messages:
    print("Dequeueing message: " + message.content)
    queue_client.delete_message(message.id, message.pop_receipt)

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 metody do pobierania receive_messages komunikatów w partiach. Następnie przetwarza każdy komunikat w każdej partii przy użyciu zagnieżdżonej for pętli. Ustawia również limitu czasu niewidoczności na pięć minut dla każdego komunikatu.

messages = queue_client.receive_messages(messages_per_page=5, visibility_timeout=5*60)

for msg_batch in messages.by_page():
   for msg in msg_batch:
      print("Batch dequeue message: " + msg.content)
      queue_client.delete_message(msg)

Usuwanie kolejki

Aby usunąć kolejkę i wszystkie zawarte w nim komunikaty, wywołaj metodę delete_queue .

print("Deleting queue: " + queue_client.queue_name)
queue_client.delete_queue()

Porada

Wypróbuj program Microsoft Azure Storage Explorer

Microsoft Azure Storage Explorer jest bezpłatną aplikacją autonomiczną oferowaną przez firmę Microsoft, która umożliwia wizualną pracę z danymi w usłudze Azure Storage w systemach Windows, macOS i Linux.

Następne kroki

Teraz, gdy znasz już podstawy usługi Queue Storage, skorzystaj z poniższych linków, aby dowiedzieć się więcej.

Aby zapoznać się z powiązanymi przykładami kodu korzystającymi z przestarzałych zestawów SDK języka Python w wersji 2, zobacz Przykłady kodu korzystające z języka Python w wersji 2.

Udostępnij za pośrednictwem