Python에서 Azure Queue Storage를 사용하는 방법

아티클
01/19/2023

개요

이 문서에서는 Azure Queue Storage 서비스를 사용하는 일반적인 시나리오를 보여 줍니다. 여기서 다루는 시나리오에는 큐 메시지의 삽입, 피킹(peeking), 가져오기 및 삭제가 포함됩니다. 큐를 만들고 삭제하는 코드도 다룹니다.

이 문서의 예제는 Python으로 작성되었으며 Python용 Azure Queue Storage 클라이언트 라이브러리를 사용합니다. 큐에 대한 자세한 내용은 다음 단계 섹션을 참조하세요.

Queue Storage란?

Azure Queue Storage는 HTTP 또는 HTTPS를 사용하여 인증된 호출을 통해 전 세계 어디에서나 액세스할 수 있는 다수의 메시지를 저장하기 위한 서비스입니다. 단일 큐 메시지의 크기는 최대 64KB일 수 있으며, 하나의 큐에 스토리지 계정의 총 용량 제한까지 수백만 개의 메시지가 포함될 수 있습니다. Queue Storage는 비동기적으로 처리할 작업의 백로그를 만드는 데 자주 사용됩니다.

큐 서비스 개념

Azure 큐 서비스에는 다음 구성 요소가 포함됩니다.

Azure 큐 서비스 구성 요소

Storage 계정: Azure Storage에 대한 모든 액세스는 Storage 계정을 통해 수행됩니다. 스토리지 계정에 대한 자세한 내용은 스토리지 계정 개요를 참조하세요.
큐: 큐에는 메시지 집합이 포함됩니다. 모든 메시지는 큐에 있어야 합니다. 큐 이름은 모두 소문자여야 합니다. 큐의 명명에 대한 자세한 내용은 큐 및 메타데이터 명명을 참조하세요.
메시지: 최대 64KB인 임의 형식의 메시지입니다. 메시지가 큐에 남아 있을 수 있는 최대 시간은 7일입니다. 2017-07-29 이상 버전에서 허용되는 최대 TTL(Time to Live)은 모든 양수 또는 메시지가 만료되지 않는 -1입니다. 이 매개 변수를 생략하면 기본 TTL(Time to Live)은 7일입니다.
URL 형식: 큐는 http://<storage account>.queue.core.windows.net/<queue> URL 형식을 사용하여 주소를 지정할 수 있습니다.

다음 URL은 다이어그램에 있는 큐의 주소를 지정합니다.

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

Azure Storage 계정 만들기

첫 번째 Azure Storage 계정을 가장 쉽게 만드는 방법은 Azure Portal을 사용하는 것입니다. 자세한 내용은 스토리지 계정 만들기를 참조하십시오.

또한 Azure PowerShell, Azure CLI 또는 .NET용 Azure Storage 리소스 공급자를 사용하여 Azure 스토리지 계정을 만들 수도 있습니다.

이 시점에서 스토리지 계정을 Azure에 만들지 않으려는 경우 Azurite 스토리지 에뮬레이터를 사용하여 로컬 환경에서 코드를 실행하고 테스트할 수도 있습니다. 자세한 내용은 로컬 Azure Storage 개발에 Azurite 에뮬레이터 사용을 참조하세요.

Python용 Azure Storage SDK 다운로드 및 설치

Python용 Azure Storage SDK에는 Python v2.7, v3.3 이상이 필요합니다.

PyPI를 통해 설치

PyPi(Python Package Index)를 통해 설치하려면 다음을 입력합니다.

pip install azure-storage-queue

참고

Python용 Azure Storage SDK v0.36 또는 이전 버전에서 업그레이드하는 경우 최신 패키지를 설치하기 전에 pip uninstall azure-storage를 사용하여 이전 SDK를 제거합니다.

대체 설치 방법은 Python용 Azure SDK를 참조하세요.

Azure Portal에서 자격 증명 복사

샘플 애플리케이션이 Azure Storage에 대한 요청을 수행하는 경우 권한이 있어야 합니다. 요청에 권한을 부여하려면 스토리지 계정 자격 증명을 연결 문자열로 애플리케이션에 추가합니다. 스토리지 계정 자격 증명을 보려면 다음 단계를 따릅니다.

Azure Portal에 로그인합니다.
스토리지 계정을 찾습니다.
스토리지 계정 메뉴 창의 보안 + 네트워킹에서 액세스 키를 선택합니다. 여기에서 계정 액세스 키 및 각 키의 전체 연결 문자열을 볼 수 있습니다.
액세스 키 창에서 키 표시를 선택합니다.
key1 섹션에서 연결 문자열 값을 찾습니다. 클립보드에 복사 아이콘을 선택하여 연결 문자열을 복사합니다. 다음 단계에서 연결 문자열 값을 환경 변수에 추가합니다.

스토리지 연결 문자열 구성

연결 문자열을 복사한 후 애플리케이션을 실행하는 로컬 컴퓨터의 새 환경 변수에 적습니다. 환경 변수를 설정하려면 콘솔 창을 열고 사용 중인 운영 체제의 지침을 따릅니다. <yourconnectionstring>을 실제 연결 문자열로 바꿉니다.

Windows
Linux 및 macOS

setx AZURE_STORAGE_CONNECTION_STRING "<yourconnectionstring>"

export AZURE_STORAGE_CONNECTION_STRING="<yourconnectionstring>"

Windows에서 환경 변수가 추가되면 명령 창의 새 인스턴스를 시작해야 합니다.

프로그램 다시 시작

환경 변수가 추가되면 이 환경 변수를 읽어야 하는 실행 중인 프로그램을 다시 시작합니다. 예를 들어 개발 환경 또는 편집기를 다시 시작한 후에 계속합니다.

Queue Storage에 액세스하도록 애플리케이션 구성

QueueClient 개체를 사용하면 큐로 작업할 수 있습니다. 프로그래밍 방식으로 Azure 큐에 액세스하려는 Python 파일의 맨 위쪽에 다음 코드를 추가합니다.

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

import os, uuid

os 패키지는 환경 변수 검색을 지원합니다. uuid 패키지는 큐 이름에 대한 고유 식별자 생성을 지원합니다.

큐 만들기

연결 문자열은 이전에 설정된 AZURE_STORAGE_CONNECTION_STRING 환경 변수에서 검색됩니다.

다음 코드는 스토리지 연결 문자열을 사용하는 QueueClient 개체를 만듭니다.

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

Azure 큐 메시지는 텍스트로 저장됩니다. 이진 데이터를 저장하려면 메시지를 큐에 넣기 전에 Base64 인코딩 및 디코딩 기능을 설정합니다.

클라이언트 개체를 만들 때 Base64 인코딩 및 디코딩 함수를 구성합니다.

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

큐에 메시지 삽입

큐에 메시지를 삽입하려면 send_message 메서드를 사용합니다.

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

메시지 보기

peek_messages 메서드를 호출하여 큐에서 제거하지 않고 메시지를 피킹할 수 있습니다. 기본적으로 이 메서드는 단일 메시지를 피킹합니다.

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

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

대기 중인 메시지의 콘텐츠 변경

큐에 있는 메시지의 콘텐츠를 변경할 수 있습니다. 메시지가 작업을 나타내는 경우 이 기능을 사용하여 작업의 상태를 업데이트할 수 있습니다.

다음 코드에서는 update_message 메서드를 사용하여 메시지를 업데이트합니다. 표시 제한 시간은 0으로 설정되어 있습니다.이는 메시지가 즉시 표시되고 콘텐츠가 업데이트됨을 의미합니다.

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)

큐 길이 가져오기

큐에 있는 메시지의 추정된 개수를 가져올 수 있습니다.

get_queue_properties 메서드는 approximate_message_count를 포함한 큐 속성을 반환합니다.

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

서비스가 요청에 응답한 후 메시지가 추가되거나 제거될 수 있으므로 이 결과는 근사치일 뿐입니다.

큐에서 메시지 제거

2단계를 거쳐 큐에서 메시지를 제거합니다. 코드에서 메시지를 처리하지 못하는 경우 이 두 단계 프로세스를 통해 동일한 메시지를 받고 다시 시도할 수 있습니다. 메시지가 성공적으로 처리된 후에 delete_message를 호출합니다.

receive_messages를 호출하면 기본적으로 큐에서 다음 메시지가 나타납니다. receive_messages에서 반환된 메시지는 이 큐의 메시지를 읽는 다른 코드에는 표시되지 않습니다. 기본적으로, 이 메시지는 30초간 표시되지 않습니다. 큐에서 메시지 제거를 완료하려면 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)

큐에서 메시지 검색을 사용자 지정할 수 있는 방법으로는 두 가지가 있습니다. 먼저, 메시지의 배치(최대 32개)를 가져올 수 있습니다. 두 번째로, 표시하지 않는 제한 시간을 더 길거나 더 짧게 설정하여 코드에서 각 메시지를 완전히 처리하는 시간을 늘리거나 줄일 수 있습니다.

다음 코드 예제는 receive_messages 메서드를 사용하여 메시지를 일괄적으로 가져옵니다. 그런 다음, 중첩된 for 루프를 사용하여 각 일괄 처리 내 각 메시지를 처리합니다. 또한 각 메시지에 대해 표시하지 않는 제한 시간을 5분으로 설정합니다.

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)

큐 삭제

큐와 큐에 표함된 모든 메시지를 삭제하려면 delete_queue 메서드를 호출합니다.

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

팁

Microsoft Azure Storage Explorer 시도

Microsoft Azure Storage Explorer는 Windows, macOS 및 Linux에서 Azure Storage 데이터로 시각적으로 작업할 수 있도록 해주는 Microsoft의 독립 실행형 무료 앱입니다.

다음 단계

이제 Queue Storage의 기본 사항을 배웠으므로 다음 링크를 따라 자세히 알아보세요.

사용되지 않는 Python 버전 2 SDK를 사용하는 관련 코드 샘플은 Python 버전 2를 사용하는 코드 샘플을 참조하세요.

다음을 통해 공유