Azure Service Bus 큐에서 메시지 보내기 및 받기(Python)

사용자 고유의 Azure 계정으로 이 빠른 시작을 사용하려면 다음이 필요합니다.

• 개발자 머신에 암호 없는 인증을 제공하는 Azure CLI를 설치합니다.
• 터미널 또는 명령 프롬프트에서 az login을 사용하여 Azure 계정으로 로그인합니다.
• 리소스에 적절한 데이터 역할을 추가할 때 동일한 계정을 사용합니다.
• 동일한 터미널 또는 명령 프롬프트에서 코드를 실행합니다.
• Service Bus 네임스페이스의 큐 이름을 적어둡니다. 코드에서 필요합니다.

아래 코드에서 사용할 다음 사항을 적어둡니다.

• Service Bus 네임스페이스 연결 문자열
• 만든 Service Bus 네임스페이스 큐

Microsoft Entra 사용자에게 역할 할당

로컬에서 개발할 때 Azure Service Bus에 연결하는 사용자 계정에 올바른 권한이 있는지 확인합니다. 메시지를 보내고 받으려면 Azure Service Bus 데이터 소유자 역할이 필요합니다. 자신에게 이 역할을 할당하려면 사용자 액세스 관리자 역할 또는 Microsoft.Authorization/roleAssignments/write 작업을 포함하는 다른 역할이 필요합니다. Azure Portal, Azure CLI 또는 Azure PowerShell을 사용하여 사용자에게 Azure RBAC 역할을 할당할 수 있습니다. 범위 개요 페이지에서 역할 할당에 사용할 수 있는 범위에 대해 자세히 알아봅니다.

다음 예에서는 Azure Service Bus 리소스에 대한 모든 권한을 제공하는 Azure Service Bus Data Owner 역할을 사용자 계정에 할당합니다. 실제 시나리오에서는 최소 권한 원칙에 따라 사용자에게 보다 안전한 프로덕션 환경에 필요한 최소한의 권한만 부여합니다.

Azure Service Bus에 대한 Azure 기본 제공 역할

Azure Service Bus의 경우 Azure Portal 및 Azure 리소스 관리 API를 통한 네임스페이스 및 관련된 모든 리소스의 관리는 이미 Azure RBAC 모델을 사용하여 보호되고 있습니다. Azure는 Service Bus 네임스페이스에 대한 액세스 권한을 부여하기 위해 아래와 같은 Azure 기본 제공 역할을 제공합니다.

• Azure Service Bus 데이터 소유자: Service Bus 네임스페이스 및 해당 엔터티(큐, 토픽, 구독 및 필터)에 대한 데이터 액세스를 사용하도록 설정합니다. 이 역할의 멤버는 큐 또는 토픽/구독에서 메시지를 보내고 받을 수 있습니다.
• Azure Service Bus 데이터 보내는 사람: 이 역할을 사용하여 Service Bus 네임스페이스 및 해당 엔터티에 대한 전송 액세스 권한을 부여합니다.
• Azure Service Bus 데이터 받는 사람: 이 역할을 사용하여 Service Bus 네임스페이스 및 해당 엔터티에 대한 수신 액세스 권한을 부여합니다.

사용자 지정 역할을 만들려면 Service Bus 작업에 필요한 권한을 참조하세요.

Azure Service Bus 소유자 역할에 Microsoft Entra 사용자 추가

Service Bus 네임스페이스 수준에서 Azure Service Bus 데이터 소유자 역할에 Microsoft Entra 사용자 이름을 추가합니다. 사용자 계정의 컨텍스트에서 실행되는 앱이 큐 또는 토픽에 메시지를 보내고 큐 또는 토픽의 구독에서 메시지를 받을 수 있습니다.

1. Azure Portal에 Service Bus 네임스페이스 페이지가 열려 있지 않은 경우 기본 검색 창이나 왼쪽 탐색 모음을 사용하여 Service Bus 네임스페이스를 찾습니다.

2. 개요 페이지의 왼쪽 메뉴에서 액세스 제어(IAM)를 선택합니다.

3. 액세스 제어(IAM) 페이지에서 역할 할당 탭을 선택합니다.

4. 위쪽 메뉴에서 + 추가를 선택한 다음, 드롭다운 메뉴에서 역할 할당 추가를 선택합니다.

   

5. 검색 상자를 사용하여 결과를 원하는 역할로 필터링합니다. 이 예에서는 Azure Service Bus Data Owner를 검색하고 일치하는 결과를 선택합니다. 다음을 선택합니다.

6. 다음에 대한 액세스 할당 아래에서 사용자, 그룹 또는 서비스 주체를 선택한 다음, + 멤버 선택을 선택합니다.

7. 대화 상자에서 Microsoft Entra 사용자 이름(일반적으로 user@domain 이메일 주소)을 검색한 다음, 대화 상자 하단에서 선택을 선택합니다.

8. 검토 + 할당을 선택하여 최종 페이지로 이동한 다음, 검토 + 할당을 다시 선택하여 프로세스를 완료합니다.

연결 문자열 가져오기

새 네임스페이스를 만들면 기본 키 및 보조 키와 각각 네임스페이스의 모든 측면에 대한 전체 제어 권한을 부여하는 기본 및 보조 연결 문자열이 포함된 초기 SAS(공유 액세스 서명) 정책이 자동으로 생성됩니다. 일반적인 보낸 사람과 받는 사람에 대해 자세히 제한된 권한이 적용된 규칙을 만드는 방법에 대한 자세한 내용은 Service Bus 인증 및 권한 부여를 참조하세요.

클라이언트는 연결 문자열을 사용하여 Service Bus 네임스페이스에 연결할 수 있습니다. 네임스페이스에 대한 기본 연결 문자열을 복사하려면 다음 단계를 수행합니다.

1. Service Bus 네임스페이스 페이지의 왼쪽 메뉴에서 공유 액세스 정책을 선택합니다.

2. 공유 액세스 정책 페이지에서 RootManageSharedAccessKey를 선택합니다.

3. 정책: RootManageSharedAccessKey 창에서 기본 연결 문자열 옆의 복사 단추를 선택하여 나중에 사용할 수 있도록 연결 문자열을 클립보드에 복사합니다. 메모장이나 기타 다른 위치에 임시로 이 값을 붙여 넣습니다.

   

   이 페이지를 사용하여 기본 키, 보조 키, 기본 연결 문자열 및 보조 연결 문자열을 복사할 수 있습니다.

1. 이 Service Bus 자습서에 대한 필수 Python 패키지를 설치하려면 경로에 Python이 있는 명령 프롬프트를 열고, 샘플을 저장하려는 폴더로 디렉터리를 변경합니다.

2. 다음 패키지를 설치합니다.

   pip install azure-servicebus
   pip install azure-identity
   pip install aiohttp
   

1. 이 Service Bus 자습서에 대한 필수 Python 패키지를 설치하려면 경로에 Python이 있는 명령 프롬프트를 열고, 샘플을 저장하려는 폴더로 디렉터리를 변경합니다.

2. 다음 패키지를 설치합니다.

   pip install azure-servicebus
   

1. Import 문을 추가합니다.

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   from azure.servicebus import ServiceBusMessage
   from azure.identity.aio import DefaultAzureCredential
   

2. 상수를 추가하고 자격 증명을 정의합니다.

   FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
   QUEUE_NAME = "QUEUE_NAME"
   
   credential = DefaultAzureCredential()
   

   Important

   • FULLY_QUALIFIED_NAMESPACE를 Service Bus 네임스페이스의 정규화된 네임스페이스로 바꿉니다.
   • QUEUE_NAME을 큐 이름으로 바꿉니다.

3. 단일 메시지를 보내는 메서드를 추가합니다.

   async def send_single_message(sender):
       # Create a Service Bus message and send it to the queue
       message = ServiceBusMessage("Single Message")
       await sender.send_messages(message)
       print("Sent a single message")
   

   보낸 사람은 만든 큐에 대한 클라이언트 역할을 하는 개체입니다. 나중에 만들어 이 함수에 인수로 보냅니다.

4. 메시지 목록을 보내는 메서드를 추가합니다.

   async def send_a_list_of_messages(sender):
       # Create a list of messages and send it to the queue
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       await sender.send_messages(messages)
       print("Sent a list of 5 messages")
   

5. 메시지 일괄 처리를 보내는 메서드를 추가합니다.

   async def send_batch_message(sender):
       # Create a batch of messages
       async with sender:
           batch_message = await sender.create_message_batch()
           for _ in range(10):
               try:
                   # Add a message to the batch
                   batch_message.add_message(ServiceBusMessage("Message inside a ServiceBusMessageBatch"))
               except ValueError:
                   # ServiceBusMessageBatch object reaches max_size.
                   # New ServiceBusMessageBatch object can be created here to send more data.
                   break
           # Send the batch of messages to the queue
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. Service Bus 클라이언트를 만든 다음, 메시지를 보낼 큐 발신자 개체를 만듭니다.

   async def run():
       # create a Service Bus client using the credential
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
           # get a Queue Sender object to send messages to the queue
           sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)
           async with sender:
               # send one message
               await send_single_message(sender)
               # send a list of messages
               await send_a_list_of_messages(sender)
               # send a batch of messages
               await send_batch_message(sender)
   
           # Close credential when no longer needed.
           await credential.close()
   

7. run 메서드를 호출하고 메시지를 인쇄합니다.

   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. Import 문을 추가합니다.

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   from azure.servicebus import ServiceBusMessage
   

2. 상수를 추가합니다.

   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

   Important

   • NAMESPACE_CONNECTION_STR을 Service Bus 네임스페이스의 연결 문자열로 바꿉니다.
   • QUEUE_NAME을 큐 이름으로 바꿉니다.

3. 단일 메시지를 보내는 메서드를 추가합니다.

   async def send_single_message(sender):
       # Create a Service Bus message and send it to the queue
       message = ServiceBusMessage("Single Message")
       await sender.send_messages(message)
       print("Sent a single message")
   

   보낸 사람은 만든 큐에 대한 클라이언트 역할을 하는 개체입니다. 나중에 만들어 이 함수에 인수로 보냅니다.

4. 메시지 목록을 보내는 메서드를 추가합니다.

   async def send_a_list_of_messages(sender):
       # Create a list of messages and send it to the queue
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       await sender.send_messages(messages)
       print("Sent a list of 5 messages")
   

5. 메시지 일괄 처리를 보내는 메서드를 추가합니다.

   async def send_batch_message(sender):
       # Create a batch of messages
       async with sender:
           batch_message = await sender.create_message_batch()
           for _ in range(10):
               try:
                   # Add a message to the batch
                   batch_message.add_message(ServiceBusMessage("Message inside a ServiceBusMessageBatch"))
               except ValueError:
                   # ServiceBusMessageBatch object reaches max_size.
                   # New ServiceBusMessageBatch object can be created here to send more data.
                   break
           # Send the batch of messages to the queue
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. Service Bus 클라이언트를 만든 다음, 메시지를 보낼 큐 발신자 개체를 만듭니다.

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient.from_connection_string(
           conn_str=NAMESPACE_CONNECTION_STR,
           logging_enable=True) as servicebus_client:
           # Get a Queue Sender object to send messages to the queue
           sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)
           async with sender:
               # Send one message
               await send_single_message(sender)
               # Send a list of messages
               await send_a_list_of_messages(sender)
               # Send a batch of messages
               await send_batch_message(sender)
   

7. run 메서드를 호출하고 메시지를 인쇄합니다.

   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. send 샘플과 마찬가지로 import 문을 추가하고, 고유한 값으로 대체해야 하는 상수를 정의하고, 자격 증명을 정의합니다.

   import asyncio
   
   from azure.servicebus.aio import ServiceBusClient
   from azure.identity.aio import DefaultAzureCredential
   
   FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
   QUEUE_NAME = "QUEUE_NAME"
   
   credential = DefaultAzureCredential()
   

2. Service Bus 클라이언트를 만든 다음, 메시지를 받을 큐 수신자 개체를 만듭니다.

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Queue Receiver object for the queue
               receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
               async with receiver:
                   received_msgs = await receiver.receive_messages(max_wait_time=5, max_message_count=20)
                   for msg in received_msgs:
                       print("Received: " + str(msg))
                       # complete the message so that the message is removed from the queue
                       await receiver.complete_message(msg)
   
           # Close credential when no longer needed.
           await credential.close()
   

3. run 메서드를 호출합니다.

   asyncio.run(run())
   

1. send 샘플과 마찬가지로 import 문을 추가하고 고유한 값으로 대체해야 하는 상수를 정의합니다.

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   
   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

2. Service Bus 클라이언트를 만든 다음, 메시지를 받을 큐 수신자 개체를 만듭니다.

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient.from_connection_string(
           conn_str=NAMESPACE_CONNECTION_STR,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Queue Receiver object for the queue
               receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
               async with receiver:
                   received_msgs = await receiver.receive_messages(max_wait_time=5, max_message_count=20)
                   for msg in received_msgs:
                       print("Received: " + str(msg))
                       # complete the message so that the message is removed from the queue
                       await receiver.complete_message(msg)
   

3. run 메서드를 호출합니다.

   asyncio.run(run())