AMQ Python クライアントの使用

AMQ Python は、以下の業界標準およびネットワークプロトコルをサポートします。

AMQ Python でサポートされている設定については、Red Hat カスタマーポータルの「Red Hat AMQ 7 でサポートされる構成」を参照してください。

本セクションでは、コア API エンティティーを紹介し、それらが一緒に操作する方法を説明します。

AMQ Python は メッセージを送受信します。メッセージは、送信側と受信側を介して接続されたピア間で転送されます。送信側およびレシーバーは セッション 上で確立されます。セッションはコネクションを介して確立されます。接続は、一意に識別された 2 つのコンテナー間で確立されます。コネクションには複数のセッションを含めることができますが、多くの場合、これは必要ありません。API を使用すると、セッションが必要でない限り、セッションを無視できます。

送信ピアは、メッセージを送信するために送信者を作成します。送信側には、リモートピアでキューまたはトピックを識別する ターゲット があります。受信ピアは、メッセージを受信するための受信側を作成します。受信側には、リモートピアでキューまたはトピックを識別する ソース があります。

メッセージの送信は、配信 と呼ばれます。メッセージは送信される内容で、ヘッダーやアノテーションなどのすべてのメタデータが含まれます。配信は、そのコンテンツの移動に関連するプロトコルエクスチェンジです。

配信が完了したことを示すには、送信側または受信側セットのいずれかです。これが設定されていることを知らせると、その配信に関する通信はなくなります。受信側は、メッセージを受諾または拒否するかどうかを指定することもできます。

sudo コマンド

本書では、root 権限を必要とするコマンドには sudo が使用されています。何らかの変更がシステム全体に影響する可能性があるため、sudo を使用する場合は注意が必要です。sudo の詳細は、「sudo コマンドの使用」を参照してください。

ファイルパス

本書では、すべてのファイルパスが Linux、UNIX、および同様のオペレーティングシステムで有効です (例: /home/andrea)。Microsoft Windows では、同等の Windows パスを使用する必要があります (例: C:\Users\andrea)。

変数テキスト

本書には、実際の環境に固有の値に置き換える必要がある変数を含むコードブロックが含まれています。変数テキストは中括弧で囲まれ、斜体の等幅フォントとしてスタイル設定されます。たとえば、以下の例では、<project-dir> を実際の環境の値に置き換えます。

$ cd <project-dir>

パッケージの使用方法は、付録B Red Hat Enterprise Linux パッケージの使用 を参照してください。

Hello World の例では、ブローカーへの接続を作成し、グリーティングが含まれるメッセージを examples キューに送信し、それを受け取ります。成功すると、受け取ったメッセージをコンソールに出力します。

examples ディレクトリーに移動し、helloworld.py の例を実行します。

$ cd /usr/share/proton/examples/python/
$ python helloworld.py
Hello World!

Hello World の例では、ブローカーへの接続を作成し、グリーティングが含まれるメッセージを examples キューに送信し、それを受け取ります。成功すると、受け取ったメッセージをコンソールに出力します。

Hello World の例をダウンロードして実行します。

> curl -o helloworld.py https://raw.githubusercontent.com/apache/qpid-proton/master/python/examples/helloworld.py
> python helloworld.py
Hello World!

このクライアントプログラムは、<connection-url> を使用してサーバーに接続します。ターゲット <address> の送信側は <message-body> が含まれるメッセージを送信し、接続を閉じて終了します。

from __future__ import print_function

import sys

from proton import Message
from proton.handlers import MessagingHandler
from proton.reactor import Container

class SendHandler(MessagingHandler):
    def __init__(self, conn_url, address, message_body):
        super(SendHandler, self).__init__()

        self.conn_url = conn_url
        self.address = address
        self.message_body = message_body

    def on_start(self, event):
        conn = event.container.connect(self.conn_url)

        # To connect with a user and password:
        # conn = event.container.connect(self.conn_url, user="<user>", password="<password>")

        event.container.create_sender(conn, self.address)

    def on_link_opened(self, event):
        print("SEND: Opened sender for target address '{0}'".format
              (event.sender.target.address))

    def on_sendable(self, event):
        message = Message(self.message_body)
        event.sender.send(message)

        print("SEND: Sent message '{0}'".format(message.body))

        event.sender.close()
        event.connection.close()

def main():
    try:
        conn_url, address, message_body = sys.argv[1:4]
    except ValueError:
        sys.exit("Usage: send.py <connection-url> <address> <message-body>")

    handler = SendHandler(conn_url, address, message_body)
    container = Container(handler)
    container.run()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass

サンプルの実行

サンプルプログラムを実行するには、これをローカルファイルにコピーし、python コマンドを使用してこれを呼び出します。詳細は、3章スタートガイド を参照してください。

$ python send.py amqp://localhost queue1 hello

このクライアントプログラムは <connection-url> を使用してサーバーに接続し、ソース <address> のレシーバーを作成し、終了するか <count> メッセージに到達するまでメッセージを受信します。

from __future__ import print_function

import sys

from proton.handlers import MessagingHandler
from proton.reactor import Container

class ReceiveHandler(MessagingHandler):
    def __init__(self, conn_url, address, desired):
        super(ReceiveHandler, self).__init__()

        self.conn_url = conn_url
        self.address = address
        self.desired = desired
        self.received = 0

    def on_start(self, event):
        conn = event.container.connect(self.conn_url)

        # To connect with a user and password:
        # conn = event.container.connect(self.conn_url, user="<user>", password="<password>")

        event.container.create_receiver(conn, self.address)

    def on_link_opened(self, event):
        print("RECEIVE: Created receiver for source address '{0}'".format
              (self.address))

    def on_message(self, event):
        message = event.message

        print("RECEIVE: Received message '{0}'".format(message.body))

        self.received += 1

        if self.received == self.desired:
            event.receiver.close()
            event.connection.close()

def main():
    try:
        conn_url, address = sys.argv[1:3]
    except ValueError:
        sys.exit("Usage: receive.py <connection-url> <address> [<message-count>]")

    try:
        desired = int(sys.argv[3])
    except (IndexError, ValueError):
        desired = 0

    handler = ReceiveHandler(conn_url, address, desired)
    container = Container(handler)
    container.run()

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass

サンプルの実行

サンプルプログラムを実行するには、これをローカルファイルにコピーし、python コマンドを使用してこれを呼び出します。詳細は、3章スタートガイド を参照してください。

$ python receive.py amqp://localhost queue1

AMQ Python は非同期イベント駆動型 API です。アプリケーションがイベントを処理する方法を定義するために、ユーザーは MessagingHandler クラスでコールバックメソッドを実装します。これらのメソッドは、ネットワークアクティビティーとして呼び出され、タイマーが新規イベントをトリガーします。

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        print("The container event loop has started")

    def on_sendable(self, event):
        print("A message can be sent")

    def on_message(self, event):
        print("A message is received")

これらはいくつかの一般的なケースイベントのみです。完全セットは API リファレンス に文書化されています。

event 引数には、イベントが関連するオブジェクトにアクセスするための属性があります。たとえば、on_connection_opened イベントはイベント connection 属性を設定します。

イベントのプライマリーオブジェクトに加えて、イベントのコンテキストを形成するすべてのオブジェクトも設定されます。特定のイベントに対する関連性のない属性は null です。

event.container
event.connection
event.session
event.sender
event.receiver
event.delivery
event.message

コンテナーはトップレベルの API オブジェクトです。これは、接続を作成するエントリーポイントであり、メインのイベントループを実行します。多くの場合、これはグローバルイベントハンドラーで構築されます。

handler = ExampleHandler()
container = Container(handler)
container.run()

各コンテナーインスタンスには、コンテナー ID と呼ばれる一意のアイデンティティーがあります。AMQ Python が接続を確立すると、コンテナー ID をリモートピアに送信します。コンテナー ID を設定するには、これを Container コンストラクターに渡します。

container = Container(handler)
container.container_id = "job-processor-3"

ユーザーが ID を設定しないと、コンテナーが構成されると、ライブラリーは UUID を生成します。

connection URL は、新規接続の確立に使用される情報をエンコードします。

scheme://host[:port]

amqps://example.com
amqps://example.net:56720
amqp://127.0.0.1
amqp://[::1]:2000

リモートサーバーに接続するには、接続 URL を使用して Container.connect() メソッドを呼び出します。通常、これは MessagingHandler.on_start() メソッド内で行われます。

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        event.container.connect("amqp://example.com")

    def on_connection_opened(self, event):
        print("Connection", event.connection, "is open")

セキュアな接続の作成に関する詳細は、7章セキュリティー を参照してください。

再接続すると、クライアントが失われた接続から回復できます。これは、一時的なネットワークまたはコンポーネントの障害後に、分散システムのコンポーネントが再確立された状態にするために使用されます。

AMQ Python はデフォルトで再接続を有効にします。接続が失われたり、接続試行に失敗した場合に、クライアントは簡単な遅延後に再度試行します。遅延は、デフォルトの最大値 10 秒まで、新しい試行ごとに指数関数的に増加します。

再接続を無効にするには、reconnect 接続オプションを False に設定します。

container.connect("amqp://example.com", reconnect=False)

接続試行間の遅延を制御するには、reset() メソッドおよび next() メソッドを実装するクラスを定義し、reconnect 接続オプションをそのクラスのインスタンスに設定します。

class ExampleReconnect(object):
    def __init__(self):
        self.delay = 0

    def reset(self):
        self.delay = 0

    def next(self):
        if self.delay == 0:
            self.delay = 0.1
        else:
            self.delay = min(10, 2 * self.delay)

        return self.delay

container.connect("amqp://example.com", reconnect=ExampleReconnect())

next メソッドは、次の遅延を秒単位で返します。reset メソッドは、再接続プロセスが開始する前に一度呼び出されます。

AMQ Python では、複数の接続エンドポイントを設定できます。ある接続に失敗すると、クライアントはリスト内の次の接続を試みます。一覧に使い切られると、プロセスは最初から開始します。

複数の接続エンドポイントを指定するには、urls 接続オプションを接続 URL の一覧に設定します。

urls = ["amqp://alpha.example.com", "amqp://beta.example.com"]
container.connect(urls=urls)

url オプションおよび urls オプションを同時に使用するのはエラーです。

AMQ Python はインバウンドネットワーク接続を受け入れ、カスタムメッセージングサーバーを構築できます。

接続のリッスンを開始するには、ローカルホストアドレスおよびリッスンするポートが含まれる URL で Container.listen() メソッドを使用します。

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        event.container.listen("0.0.0.0")

    def on_connection_opened(self, event):
        print("New incoming connection", event.connection)

特別な IP アドレス 0.0.0.0 は、利用可能なすべての IPv4 インターフェースでリッスンします。すべての IPv6 インターフェースをリッスンするには、[::0] を使用します。

詳細は、サーバー receive.py の例 を参照してください。

AMQ Python は SSL/TLS を使用して、クライアントとサーバー間の通信を暗号化します。

SSL/TLS を使用してリモートサーバーに接続するには、amqps スキームで接続 URL を使用します。

container.connect("amqps://example.com")

AMQ Python は、ユーザーとパスワードによる接続を認証できます。

認証に使用する認証情報を指定するには、connect() メソッドで user および password オプションを設定します。

container.connect("amqps://example.com", user="alice", password="secret")

AMQ Python は SASL プロトコルを使用して認証を実行します。SASL は多くの異なる 認証 メカニズム を使用できます。2 つのネットワークピアに接続すると、許可されるメカニズムを交換し、両方で許可される最も強力なメカニズムが選択されます。

デフォルトでは、AMQ Python はローカル SASL ライブラリー設定でサポートされるすべてのメカニズムを許可します。許可されるメカニズムを制限し、ネゴシエートできるメカニズムを制御するには、allowed_mechs コネクションオプションを使用します。スペースで区切ったメカニズム名のリストが含まれる文字列を受け入れます。

container.connect("amqps://example.com", allowed_mechs="ANONYMOUS")

この例では、サーバーが他のオプションを提供するように接続しても、ANONYMOUS メカニズムを使用した認証を強制します。有効なメカニズムには、ANONYMOUS、PLAIN、SCRAM-SHA-256、SCRAM-SHA-1、GSSAPI、および EXTERNAL が含まれます。

AMQ Python はデフォルトで SASL を有効にします。これを無効にするには、sasl_enabled 接続オプションを false に設定します。

event.container.connect("amqps://example.com", sasl_enabled=False)

Kerberos は、暗号化されたチケットの交換に基づいて一元管理された認証用のネットワークプロトコルです。詳細は、「Kerberos の使用」を参照してください。

一部のメッセージサーバーは、キューとトピックのオンデマンド作成をサポートします。送信側またはレシーバーが割り当てられている場合、サーバーは送信側のターゲットアドレスまたは受信側ソースアドレスを使用して、アドレスに一致する名前を持つキューまたはトピックを作成します。

メッセージサーバーは通常、キュー (1 対 1 のメッセージ配信用) またはトピック （1 対多のメッセージ配信の場合) を作成します。クライアントは、ソースまたはターゲットに queue または topic 機能を設定することで、希望のものを指定できます。

キューまたはトピックセマンティクスを選択するには、以下の手順に従います。

class CapabilityOptions(SenderOption):
    def apply(self, sender):
        sender.target.capabilities.put_object(symbol("queue"))

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        event.container.create_sender(conn, "jobs", options=CapabilityOptions())

class CapabilityOptions(ReceiverOption):
    def apply(self, receiver):
        receiver.source.capabilities.put_object(symbol("topic"))

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        event.container.create_receiver(conn, "notifications", options=CapabilityOptions())

詳細は、以下の例を参照してください。

永続サブスクリプションは、メッセージの受信側を表すリモートサーバーの状態です。通常、クライアントが閉じられると、メッセージ受信側は破棄されます。ただし、永続サブスクリプションは永続的であるため、クライアントはこれらのサブスクリプションの割り当てを解除してから、後で再度アタッチすることができます。デタッチ中に受信したすべてのメッセージは、クライアントの再割り当て時に利用できます。

永続サブスクリプションは、クライアントコンテナー ID とレシーバー名を組み合わせてサブスクリプション ID を形成することで一意に識別されます。サブスクリプションが回復できるようにするには、これらの値に安定した値が必要です。

永続サブスクリプションを作成するには、以下の手順に従います。

サブスクリプションからデタッチするには、Receiver.detach() メソッドを使用します。サブスクリプションを終了するには、Receiver.close() メソッドを使用します。

詳細は、durable-subscribe.py の例を参照してください。

共有サブスクリプションとは、1 つ以上のメッセージレシーバーを表すリモートサーバーの状態のことです。共有されているので、複数のクライアントは同じメッセージのストリームから消費できます。

クライアントは、レシーバーソースに shared 機能を設定して、共有サブスクリプションを設定します。

共有サブスクリプションは、クライアントコンテナー ID とレシーバー名を組み合わせてサブスクリプション ID を形成することで一意に識別されます。複数のクライアントプロセスで同じサブスクリプションを見つけることができるように、これらの値が安定している必要があります。shared に加えて global 機能が設定されている場合、レシーバー名のみを使用してサブスクリプションを特定します。

永続サブスクリプションを作成するには、以下の手順に従います。

サブスクリプションからデタッチするには、Receiver.detach() メソッドを使用します。サブスクリプションを終了するには、Receiver.close() メソッドを使用します。

詳細は、shared-subscribe.py の例を参照してください。

メッセージを送信するには、on_sendable イベントハンドラーを上書きし、Sender.send() メソッドを呼び出します。sendable イベントは、Sender が少なくとも 1 つのメッセージを送信するのに十分なクレジットがあると実行されます。

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        sender = event.container.create_sender(conn, "jobs")

    def on_sendable(self, event):
        message = Message("job-content")
        event.sender.send(message)

詳細は、send.py の例を参照してください。

メッセージを送信すると、送信側は転送を表す delivery オブジェクトへの参照を維持することができます。メッセージが配信されると、受信側はこれを受け入れるか、拒否します。各配信の結果が送信者に通知されます。

送信されたメッセージの結果を監視するには、on_accepted および on_rejected イベントハンドラーを上書きし、配信状態の更新を send() から返された配信にマップします。

def on_sendable(self, event):
    message = Message(self.message_body)
    delivery = event.sender.send(message)

def on_accepted(self, event):
    print("Delivery", event.delivery, "is accepted")

def on_rejected(self, event):
    print("Delivery", event.delivery, "is rejected")

メッセージの受信には、レシーバーを作成し、on_message イベントハンドラーを上書きします。

class ExampleHandler(MessagingHandler):
    def on_start(self, event):
        conn = event.container.connect("amqp://example.com")
        receiver = event.container.create_receiver(conn, "jobs")

    def on_message(self, event):
        print("Received message", event.message, "from", event.receiver)

詳細は、receive.py の例を参照してください。

配信を明示的に許可または拒否するには、on_message イベントハンドラーの ACCEPTED 状態または REJECTED 状態の Delivery.update() メソッドを使用します。

def on_message(self, event):
    try:
        process_message(event.message)
        event.delivery.update(ACCEPTED)
    except:
        event.delivery.update(REJECTED)

デフォルトでは、配信が明示的な承認されない場合、ライブラリーは on_message の戻り値後に受け入れます。この動作を無効にするには、auto_accept レシーバーオプションを false に設定します。

AMQ Python が出力するすべての例外はProtonException クラスを継承しており、そのクラスは Python のException クラスを継承しています。

以下の例は、AMQ Python から発生した例外をキャッチする方法を示しています。

try:
    # Something that might throw an exception
except ProtonException as e:
    # Handle Proton-specific problems here
except Exception as e:
    # Handle more general problems here
}

API 固有の例外処理が必要ない場合は、ProtonException が継承されてから Exception のみをキャッチする必要があります。

以下の messaging_handler メソッドを上書きすると、プロトコルレベルのエラーを処理することができます。

これらのイベント処理機能は、イベントにある特定のオブジェクトとともにエラー状態が生じるたびに呼び出されます。エラーハンドラーを呼び出すと、適切なクローズハンドラーも呼び出されます。

クライアントは AMQP プロトコルフレームをコンソールに記録できます。通常、このデータは問題を診断する際に重要です。

プロトコルロギングを有効にするには、PN_TRACE_FRM 環境変数を 1 に設定します。

$ export PN_TRACE_FRM=1
$ <your-client-program>

プロトコルロギングを無効にするには、PN_TRACE_FRM 環境変数の設定を解除します。

クライアントは、OpenTracing 標準の Jaeger 実装に基づいて分散トレーシングを提供します。アプリケーションでトレースを有効にするには、以下の手順に従います。

Jaeger の設定に関する詳細は、「Jaeger Sampling」を参照してください。

テストまたはデバッグを行う場合は、Jaeger が特定の操作を追跡できるように強制できます。詳細は、Jaeger Python クライアントのドキュメント を参照してください。

アプリケーションをキャプチャーするトレースを表示するには、Jaeger Getting Started を使用して Jaeger インフラストラクチャーおよびコンソールを実行します。

設定された場合、AMQ Python は MESSAGING_CONNECT_FILE 環境変数の値を使用して設定ファイルを見つけます。

MESSAGING_CONNECT_FILE が設定されていない場合、AMQ Python は以下の場所で connect.json という名前のファイルを検索します。最初の一致で停止します。

connect.json ファイルが見つからない場合、ライブラリーはすべてのオプションにデフォルト値を使用します。

connect.json ファイルには JSON データが含まれ、JavaScript コメントの追加サポートが提供されます。

すべての設定属性は任意で、またはデフォルト値を持っているため、簡単な例にはいくつかの詳細のみが必要になります。

{
    "host": "example.com",
    "user": "alice",
    "password": "secret"
}

SASL および SSL/TLS オプションは、"sasl" および "tls" namespace で入れ子になっています。

{
    "host": "example.com",
    "user": "ortega",
    "password": "secret",
    "sasl": {
        "mechanisms": ["SCRAM-SHA-1", "SCRAM-SHA-256"]
    },
    "tls": {
        "cert": "/home/ortega/cert.pem",
        "key": "/home/ortega/key.pem"
    }
}

オプションキーは、ドット (.) を含む属性は namespace 内でネストされた属性を表します。

AMQP メッセージは AMQP タイプシステムを使用して構成されます。この一般的な形式を使用するのは、異なる言語の AMQP クライアントが、相互運用できることが理由です。

メッセージを送信する場合、AMQ Python は自動的に言語ネイティブの型を AMQP エンコードデータに変換します。メッセージの受信時に、リバース変換が行われます。

AMQP は、JMS メッセージングモデルへの標準的なマッピングを定義します。本項では、そのマッピングのさまざまな側面について説明します。詳細は、「AMQ JMS 相互運用性」を参照してください。

JMS メッセージタイプ

AMQ Python は、本文タイプが異なる、単一のメッセージを提供します。一方、JMS API は異なるメッセージタイプを使用して、さまざまな種類のデータを表します。以下の表は、特定のボディ型が JMS メッセージタイプにマッピングする方法を示しています。

結果として生成される JMS メッセージタイプの明示的な制御を行うために、x-opt-jms-msg-type メッセージアノテーションを設定できます。詳細は、「AMQ JMS 相互運用性」の章を参照してください。

AMQ Broker は AMQP 1.0 クライアントと相互運用するために設計されています。以下をチェックして、ブローカーが AMQP メッセージング用に設定されていることを確認します。

AMQ Interconnect は AMQP 1.0 クライアントと動作します。以下をチェックして、コンポーネントが正しく設定されていることを確認します。