Azure Service Bus トピックへのメッセージ送信とトピックのサブスクリプションからのメッセージ受信 (Python)

[アーティクル]
02/06/2024

このチュートリアルでは、次の手順を実行します。

Azure Portal を使用して Service Bus 名前空間を作成する。
Azure Portal を使用して Service Bus トピックを作成する。
そのトピックに対する Service Bus サブスクリプションを Azure Portal で作成する。
次の目的で azure-servicebus パッケージを使用する Python アプリケーションを作成する。
- メッセージのセットをトピックに送信する。
- サブスクリプションからそれらのメッセージを受信する。

注意

このクイックスタートでは、メッセージのバッチを Service Bus トピックに送信し、それらのメッセージをトピックのサブスクリプションから受信するという単純なシナリオの詳細な手順を説明します。 Azure Service Bus の事前構築済みの Python サンプルが、GitHub の Azure SDK for Python リポジトリに用意されています。

前提条件

Azure サブスクリプション。
Python 3.8 以降

Note

このチュートリアルでは、コピーして Python を使用して実行できるサンプルを扱います。 Python アプリケーションを作成する手順については、Python アプリケーションを作成して Azure Web サイトにデプロイする方法に関する記事を参照してください。このチュートリアルで使用するパッケージのインストールの詳細については、Python インストールガイドを参照してください。

Azure Portal での名前空間の作成

Azure の Service Bus メッセージングエンティティを使用するには、Azure 全体で一意となる名前を備えた名前空間を最初に作成しておく必要があります。名前空間により、ご利用のアプリケーション内に Service Bus リソース (キュー、トピックなど) 用のスコープコンテナーが提供されます。

名前空間を作成するには:

Azure portal にサインインします。
[すべてのサービス] ページに移動します。
左側のナビゲーションバーで、カテゴリの一覧から [統合] を選択し、[Service Bus] 上にマウスポインターを置き、[Service Bus] タイルの [+] ボタンを選択します。
[名前空間の作成] ページの [基本] タブで、こちらの手順を実行します。
1. [サブスクリプション] で、名前空間を作成する Azure サブスクリプションを選択します。
2. [リソースグループ] で、名前空間を追加する既存のリソースグループを選択するか、新しいリソースグループを作成します。
3. 名前空間の名前を入力します。名前空間名は次の名前付け規則に従う必要があります。
  - この名前は Azure 全体で一意である必要があります。その名前が使用できるかどうかがすぐに自動で確認されます。
  - 名前の長さは 6 ～ 50 文字である。
  - この名前には、文字、数字、ハイフン "-" のみを含めることができます。
  - 名前の先頭は文字、末尾は文字または数字にする必要があります。
  - 名前の末尾は “-sb“ または “-mgmt“ にはできません。
4. [場所] で、名前空間をホストするリージョンを選択します。
5. [価格レベル] で、名前空間の価格レベル (Basic、Standard、Premium) を選択します。このクイックスタートでは、 [Standard] を選択します。
  
  重要
  
  トピックとサブスクリプションを使用する場合は、Standard または Premium を選択してください。 Basic 価格レベルでは、トピックとサブスクリプションはサポートされていません。
  
  [Premium] 価格レベルを選択した場合は、メッセージングユニットの数を指定します。 Premium レベルでは、各ワークロードが分離した状態で実行されるように、CPU とメモリのレベルでリソースが分離されます。このリソースのコンテナーをメッセージングユニットと呼びます。 Premium 名前空間には、少なくとも 1 つのメッセージングユニットがあります。 Service Bus の Premium 名前空間ごとに、1 個、2 個、4 個、8 個、または 16 個のメッセージングユニットを選択できます。詳細については、Service Bus の Premium メッセージングに関するページをご覧ください。
6. ページ下部にある [確認と作成] を選択します。
7. [確認および作成] ページで、設定を確認し、 [作成] を選択します。
リソースのデプロイが成功したら、デプロイページで [リソースに移動] を選択します。
Service Bus 名前空間のホームページが表示されます。

Azure Portal を使用したトピックの作成

[Service Bus 名前空間] ページで、左側のメニューの [トピック] を選択します。
ツールバーの [+ トピック] を選択します。
トピックの名前を入力します。他のオプションは既定値のままにしてください。
[作成] を選択します。

トピックに対するサブスクリプションの作成

前のセクションで作成したトピックを選択します。
[Service Bus トピック] ページで、ツールバーの [+ サブスクリプション] を選択します。
[サブスクリプションの作成] ページで、次の手順に従います。
1. サブスクリプションの名前として「S1」と入力します。
2. [最大配信数] に「3」と入力します。
3. 次に、 [作成] を選択してサブスクリプションを作成します。

Azure に対してアプリを認証する

このクイックスタートでは、Azure Service Bus に接続する 2 つの方法である、パスワードレスと接続文字列について説明します。

最初のオプションでは、Microsoft Entra ID とロールベースのアクセス制御 (RBAC) でセキュリティプリンシパルを使用して Service Bus 名前空間に接続する方法を示します。コードや構成ファイル、または Azure Key Vault などのセキュリティで保護されたストレージに、ハードコーディングされた接続文字列を含める心配はありません。

2 番目のオプションでは、接続文字列を使用して Service Bus 名前空間に接続する方法を示します。 Azure を初めて使用する場合は、接続文字列オプションの方が理解しやすいかもしれません。実際のアプリケーションと運用環境では、パスワードレスオプションを使用することをお勧めします。詳細については、「認証と承認」を参照してください。パスワードレス認証の詳細については、概要ページを参照してください。

パスワードレス (推奨)
接続文字列

Microsoft Entra ユーザーにロールを割り当てる

ローカルでの開発時には、Azure Service Bus に接続するユーザーアカウントに正しいアクセス許可があることを確認してください。メッセージを送受信するには、Azure Service Bus データ所有者ロールが必要です。このロールを自分に割り当てるには、ユーザーアクセス管理者ロール、または Microsoft.Authorization/roleAssignments/write アクションを含む別のロールが必要です。 Azure portal、Azure CLI、または Azure PowerShell を使用して、ユーザーに Azure RBAC ロールを割り当てることができます。ロールの割り当てに使用できるスコープの詳細は、スコープの概要ページを参照してください。

次の例では、ユーザーアカウントに、Azure Service Bus Data Owner ロールを割り当てます。これにより、Azure Service Bus リソースへのフルアクセスが提供されます。実際のシナリオでは、より安全な運用環境を実現するため、最小限の特権の原則に従って、必要な最小限のアクセス許可のみをユーザーに付与します。

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 操作に必要な権限に関するページを参照してください。

Microsoft Entra ユーザーを Azure Service Bus 所有者ロールに追加する

Microsoft Entra ユーザー名を、Service Bus 名前空間レベルの Azure Service Bus データ所有者ロールに追加します。これにより、ユーザーアカウントのコンテキストで実行されているアプリがキューまたはトピックにメッセージを送信し、キューまたはトピックのサブスクリプションからメッセージを受信できるようになります。

重要

ほとんどの場合、ロールの割り当てが Azure に反映されるまでの時間は 1、2 分です。まれに、最大 8 分かかる場合があります。初めてコードを実行したときに認証エラーを受け取る場合は、しばらく待ってから再試行してください。

Azure portal で Service Bus 名前空間ページが開いていない場合は、メイン検索バーまたは左側のナビゲーションを使用して Service Bus 名前空間を見つけます。
概要ページで、左側のメニューから [アクセス制御 (IAM)] を選択します。
[アクセス制御 (IAM)] ページで、[ロールの割り当て] タブを選びます。
上部のメニューから [+ 追加] を選択し、次に結果のドロップダウンメニューから [ロールの割り当ての追加] を選択します。
検索ボックスを使って、結果を目的のロールに絞り込みます。この例では、Azure Service Bus Data Owner を検索して一致する結果を選択します。 [次へ] を選びます。
[アクセスの割り当て先] で、[ユーザー、グループ、またはサービスプリンシパル] を選び、[+ メンバーの選択] を選びます。
ダイアログで、自分の Microsoft Entra ユーザー名 (通常は user@domain メールアドレス) を検索し、ダイアログの下部にある [選択] を選びます。
[レビューと割り当て] を選んで最終ページに移動し、もう一度 [レビューと割り当て] を行ってプロセスを完了します。

パスワードレス認証と自分の Azure アカウントを使用してこのクイックスタートに従うには:

Azure CLI をインストールします。
ターミナルまたはコマンドプロンプトで、az login を使って Azure アカウントでサインインします。
チュートリアルで後ほどリソースに適切なロールを追加するときに、同じアカウントを使用します。
同じターミナルまたはコマンドプロンプトで、チュートリアルコードを実行します。

重要

必ず az login を使用してサインインしてください。パスワードレスコードのクラス DefaultAzureCredential は、Azure CLI 資格情報を使用して Microsoft Entra ID で認証します。

パスワードレスコードを使用するには、以下を指定する必要があります。

完全修飾サービスバス名前空間 (例: <service-bus-namespace>.servicebus.windows.net)
トピック名
subscription name

pip を使用してパッケージをインストールする

パスワードレス (推奨)
接続文字列

この Service Bus チュートリアルに必要な Python パッケージをインストールするには、パスに Python が含まれているコマンドプロンプトを開きます。サンプルを配置するフォルダーにディレクトリを変更します。

パッケージをインストールします。

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

この Service Bus チュートリアルに必要な Python パッケージをインストールするには、パスに Python が含まれているコマンドプロンプトを開きます。サンプルを配置するフォルダーにディレクトリを変更します。
パッケージをインストールします。
```
pip install azure-servicebus
```

メッセージをトピックに送信する

次のサンプルコードは、メッセージのバッチを Service Bus トピックに送信する方法を示しています。詳細については、コードのコメントを参照してください。

Visual Studio Code などのお気に入りのエディターを開き、send.py ファイルを作成し、それに次のコードを追加します。

パスワードレス (推奨)
接続文字列

次の import ステートメントを追加します。

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

定数を追加し、資格情報を定義します。
```
FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
TOPIC_NAME = "TOPIC_NAME"

credential = DefaultAzureCredential()
```
重要
- FULLY_QUALIFIED_NAMESPACE を Service Bus 名前空間の完全修飾名前空間に置き換えます。
- TOPIC_NAME を対象のトピックの名前に置き換えます。
前述のコードでは、Azure Identity クライアントライブラリの DefaultAzureCredential クラスを使用しました。開発中にアプリがローカルで実行されると、DefaultAzureCredential は、Azure CLI にログインしたアカウントを使用して、Azure に対する検出と認証を自動的に行います。アプリが Azure にデプロイされると、DefaultAzureCredential は、コードを変更することなく、マネージド ID を介して Microsoft Entra ID に対してアプリを認証できます。
単一のメッセージを送信するためのメソッドを追加します。
```
async def send_single_message(sender):
    # Create a Service Bus message
    message = ServiceBusMessage("Single Message")
    # send the message to the topic
    await sender.send_messages(message)
    print("Sent a single message")
```
sender は、作成したトピックのクライアントとして機能するオブジェクトです。それを後で作成し、引数としてこの関数に送信することになります。

一連のメッセージを送信するためのメソッドを追加します。

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

メッセージのバッチを送信するためのメソッドを追加します。

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 topic
        await sender.send_messages(batch_message)
    print("Sent a batch of 10 messages")

Service Bus クライアントを作成し、メッセージを送信するためのトピックの sender オブジェクトを作成します。

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 Topic Sender object to send messages to the topic
        sender = servicebus_client.get_topic_sender(topic_name=TOPIC_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()

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

次の import ステートメントを追加します。

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

次の定数を追加します。
```
NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STRING"
TOPIC_NAME = "TOPIC_NAME"
```
重要
- NAMESPACE_CONNECTION_STRING を名前空間の接続文字列に置き換えます。
- TOPIC_NAME を対象のトピックの名前に置き換えます。
単一のメッセージを送信するためのメソッドを追加します。
```
async def send_single_message(sender):
    # Create a Service Bus message
    message = ServiceBusMessage("Single Message")
    # send the message to the topic
    await sender.send_messages(message)
    print("Sent a single message")
```
sender は、作成したトピックのクライアントとして機能するオブジェクトです。それを後で作成し、引数としてこの関数に送信することになります。

一連のメッセージを送信するためのメソッドを追加します。

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

メッセージのバッチを送信するためのメソッドを追加します。

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 topic
        await sender.send_messages(batch_message)
    print("Sent a batch of 10 messages")

Service Bus クライアントを作成し、メッセージを送信するためのトピックの sender オブジェクトを作成します。

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 Topic Sender object to send messages to the topic
        sender = servicebus_client.get_topic_sender(topic_name=TOPIC_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)

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

サブスクリプションからメッセージを受信する

次のサンプルコードは、サブスクリプションからメッセージを受信する方法を示しています。このコードは、新しいメッセージを受信しない状態が 5 (max_wait_time) 秒続くまで、新しいメッセージを受信し続けます。

Visual Studio Code などのお気に入りのエディターを開き、revc.py ファイルを作成し、それに次のコードを追加します。

パスワードレス (推奨)
接続文字列

送信サンプルと同様に、import ステートメントを追加し、独自の値に置き換える定数を定義し、資格情報を定義します。

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

FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
SUBSCRIPTION_NAME = "SUBSCRIPTION_NAME"
TOPIC_NAME = "TOPIC_NAME"

credential = DefaultAzureCredential()

Service Bus クライアントを作成してから、メッセージを受信するためのサブスクリプションの receiver オブジェクトを作成します。

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:

        async with servicebus_client:
            # get the Subscription Receiver object for the subscription
            receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, 
            subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
            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 subscription
                    await receiver.complete_message(msg)
        # Close credential when no longer needed.
        await credential.close()

run メソッドを呼び出します。
```
asyncio.run(run())
```

送信サンプルと同様に、import ステートメントを追加し、独自の値に置き換える定数を定義します。

import asyncio
from azure.servicebus.aio import ServiceBusClient

NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STRING"
SUBSCRIPTION_NAME = "SUBSCRIPTION_NAME"
TOPIC_NAME = "TOPIC_NAME"

Service Bus クライアントを作成してから、メッセージを受信するためのサブスクリプションの receiver オブジェクトを作成します。

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 Subscription Receiver object for the subscription
            receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_NAME, 
            subscription_name=SUBSCRIPTION_NAME, max_wait_time=5)
            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 subscription
                    await receiver.complete_message(msg)

run メソッドを呼び出します。
```
asyncio.run(run())
```

アプリを実行する

パスに Python が含まれているコマンドプロンプトを開き、コードを実行して、トピックのサブスクリプションに対するメッセージを送受信します。

python send.py; python recv.py

次の出力が表示されます。

Sent a single message
Sent a list of 5 messages
Sent a batch of 10 messages
Done sending messages
-----------------------
Received: Single Message
Received: Message in list
Received: Message in list
Received: Message in list
Received: Message in list
Received: Message in list
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch
Received: Message inside a ServiceBusMessageBatch

Azure portal で、使用する Service Bus 名前空間に移動します。 [概要] ページで、受信メッセージ数と送信メッセージ数が 16 になっていることを確認します。これらのカウントが表示されない場合は、数分待ってから、ページを最新の情報に更新してください。

受信メッセージ数と送信メッセージ数

下のペインでトピックを選択し、トピックの [Service Bus トピック] ページを表示します。このページの [メッセージ] グラフに 3 つの受信メッセージと 3 つの送信メッセージが確認できると思います。

受信メッセージと送信メッセージ

このページでいずれかのサブスクリプションを選択すると、その [Service Bus Subscription](Service Bus サブスクリプション) ページが表示されます。このページで、アクティブなメッセージ数や配信不能メッセージ数を確認できます。この例では、すべてのメッセージが受信済みであるため、アクティブなメッセージ数はゼロです。

アクティブなメッセージ数

受信コードをコメントアウトした場合、アクティブなメッセージ数が "16" と表示されます。

アクティブなメッセージ数 - 受信しなかった場合

次の手順

次のドキュメントおよびサンプルを参照してください。

Python 用の Azure Service Bus クライアントライブラリ
サンプル。
- sync_samples フォルダーには、Service Bus を同期的な方法で扱う方法を紹介したサンプルが格納されています。このクイックスタートで使用したのは、こちらの方法です。
- async_samples フォルダーには、Service Bus を非同期的な方法で扱う方法を紹介したサンプルが格納されています。
azure-servicebus のリファレンスドキュメント

次の方法で共有