AMQ Python クライアントの使用

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

AMQ Python は、以下に示す OS および言語のバージョンをサポートします。詳細は、「Red Hat AMQ 7 Supported Configurations」を参照してください。

AMQ Python は、以下の AMQ コンポーネントおよびバージョンと組み合わせてサポートされます。

本セクションでは、コア API エンティティーを紹介し、コア API が連携する方法を説明します。

AMQ Python は メッセージ を送受信します。メッセージは、senders と receivers を介して、接続されたピアの間で転送されます。送信側および受信側は セッション 上で確立されます。セッションは接続上で確立されます。接続は、一意に識別された 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 リファレンス に文書化されています。

イベント引数には、イベントが関係するオブジェクトにアクセスするための属性が含まれます。たとえば、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 を生成します。

接続 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（） メソッドを実装するクラスを定義し、再接続 接続オプションをそのクラスのインスタンスに設定します。

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 はインバウンドネットワーク接続を受け入れ、カスタムメッセージングサーバーを構築できます。

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

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_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 receiver オプションを 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 が継承されるため 、ProtonException のみをキャッチする必要があります。

以下の 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 クライアントが相互に対話できる理由の 1 つです。

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

AMQP は JMS メッセージングモデルへの標準マッピングを定義します。本セクションでは、そのマッピングのさまざまな側面について説明します。詳細は、AMQ JMS Interoperability の章を参照してください。

JMS メッセージタイプ

AMQ Python は、本文タイプが異なる、単一のメッセージを提供します。一方、JMS API は異なるメッセージタイプを使用してさまざまな種類のデータを表します。次の表は、特定の本文タイプが JMS メッセージタイプにどのようにマップされるかを示しています。

作成される JMS メッセージタイプをさらに明示的に制御するには、x-opt-jms-msg-type メッセージアノテーションを設定できます。詳細は、AMQ JMS Interoperability の章を参照してください。

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

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