AMQ OpenWire JMS クライアントの使用

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

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

AMQ OpenWire JMS は、次の AMQ コンポーネントおよびバージョンとの組み合わせでサポートされています。

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

AMQ OpenWire JMS は メッセージ を送受信します。メッセージは、メッセージプロデューサー と コンシューマー を使用して接続されたピア間で転送されます。プロデューサーとコンシューマーは セッション 上で確立されます。セッションは接続上で確立されます。接続は 接続ファクトリー によって作成されます。

送信ピアは、メッセージ送信用のプロデューサーを作成します。プロデューサーには、リモートピアでターゲットキューまたはトピックを識別する 宛先 があります。受信ピアは、メッセージ受信用のコンシューマーを作成します。プロデューサーと同様に、コンシューマーにはリモートピアでソースキューまたはトピックを識別する宛先があります。

宛先は、キュー または トピック のいずれかです。JMS では、キューとトピックはメッセージを保持する名前付きブローカーエンティティーのクライアント側表現です。

キューは、ポイントツーポイントセマンティクスを実装します。各メッセージは 1 つのコンシューマーによってのみ認識され、メッセージは読み取り後にキューから削除されます。トピックはパブリッシュ/サブスクライブセマンティクスを実装します。各メッセージは複数のコンシューマーによって認識され、メッセージは読み取り後も他のコンシューマーで利用できるままになります。

詳細は、JMS tutorial を参照してください。

sudo コマンド

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

ファイルパス

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

変数テキスト

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

$ cd <project-dir>

Red Hat Maven リポジトリーからクライアントライブラリーをダウンロードするように Maven 環境を設定します。

これで、Maven プロジェクトでクライアントを使用できるようになります。

オンラインリポジトリーの代わりに、AMQ OpenWire JMS をファイルベースの Maven リポジトリーとしてローカルファイルシステムにインストールできます。

この例では、exampleQueue という名前のキューにコンシューマーおよびプロデューサーを作成します。テキストメッセージを送信してから受信し、受信したメッセージをコンソールに出力します。

Linux で実行すると、以下のような出力になります。

$ java -cp "target/classes:target/dependency/*" org.apache.activemq.artemis.jms.example.QueueExample
Sent message: This is a text message
Received message: This is a text message

この例のソースコードは <install-dir>/examples/protocols/openwire/queue/src ディレクトリーにあります。追加の例は、<install-dir>/examples/protocols/openwire ディレクトリーにあります。

JMS アプリケーションは InitialContextFactory から取得した JNDI InitialContext オブジェクトを使用して、接続ファクトリーなどの JMS オブジェクトを検索します。AMQ OpenWire JMS は、org.apache.activemq.jndi.ActiveMQInitialContextFactory クラスで InitialContextFactory の実装を提供します。

InitialContextFactory の実装は、InitialContext オブジェクトがインスタンス化されると検出されます。

javax.naming.Context context = new javax.naming.InitialContext();

実装を見つけるには、お使いの環境で JNDI を設定する必要があります。これを実現するには、jndi.properties ファイルを使用する方法とシステムプロパティーを使用する方法の 2 つの主な方法があります。

jndi.properties ファイルの使用

jndi.properties という名前のファイルを作成し、Java クラスパスに配置します。java.naming.factory.initial キーでプロパティーを追加します。

java.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory

Maven ベースのプロジェクトでは、jndi.properties ファイルは <project-dir>/src/main/resources ディレクトリーに配置されます。

システムプロパティーの使用

java.naming.factory.initial システムプロパティーを設定します。

$ java -Djava.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory ...

JMS 接続ファクトリーは、接続を作成するためのエントリーポイントです。これは、アプリケーション固有の設定をエンコードする接続 URI を使用します。

ファクトリー名と接続 URI を設定するには、以下の形式でプロパティーを作成します。この設定は、jndi.properties ファイルに保存するか、対応するシステムプロパティーを設定できます。

connectionFactory.<factory-name> = <connection-uri>

たとえば、以下のように app1 という名前のファクトリーを設定します。

connectionFactory.app1 = tcp://example.net:61616?jms.clientID=backend

その後、JNDI コンテキストを使用して、app1 の名前を使用して設定済みの接続ファクトリーを検索できます。

ConnectionFactory factory = (ConnectionFactory) context.lookup("app1");

接続ファクトリーは、次の形式の接続 URI を使用して設定されます。

<scheme>://<host>:<port>[?<option>=<value>[&<option>=<value>...]]

暗号化されていない接続のスキームは tcp で、SSL/TLS 接続の場合は ssl です。

たとえば、以下はポート 61616 でホスト example.net に接続する接続 URI で、クライアント ID を backend に設定します。

tcp://example.net:61616?jms.clientID=backend

フェイルオーバー URI は次の形式を取ります。

failover:(<connection-uri>[,<connection-uri>])[?<option>=<value>[&<option>=<value>...]]

nested. で始まるオプションは、リストの各接続 URI に適用されます。

使用可能な接続オプションについては、次のセクションで説明します。

Prefetch ポリシーオプション

Prefetch ポリシーは、各 MessageConsumer がリモートピアから取得し、ローカルの prefetch バッファーに保持するメッセージの数を決定します。

prefetch の値は、キューまたは共有サブスクリプションの複数のコンシューマーへのメッセージの分散に影響します。値が大きいと、各コンシューマーに一度に送信されるバッチが大きくなる可能性があります。より均等にラウンドロビンの分散を実現するには、小さい値を使用します。

再配信ポリシーオプション

再配信ポリシーは、クライアント上で再配信されたメッセージの処理方法を制御します。

クライアントは、プロパティー wireFormat.minLargeMessageSize の値を設定することで、大きなメッセージサポートを有効にすることができます。wireFormat.minLargeMessageSize より大きいメッセージは、大きなメッセージと見なされます。

大きなメッセージに書き込むには、BytesMessage.writeBytes() メソッドを使用します。以下の例では、ファイルからバイトを読み取り、メッセージに書き込みます。

BytesMessage message = session.createBytesMessage();
File inputFile = new File(inputFilePath);
InputStream inputStream = new FileInputStream(inputFile);

int numRead;
byte[] buffer = new byte[1024];

while ((numRead = inputStream.read(buffer, 0, buffer.length)) != -1) {
    message.writeBytes(buffer, 0, numRead);
}

大きなメッセージから読み取るには、BytesMessage.readBytes() メソッドを使用します。以下の例では、メッセージからバイトを読み取り、ファイルに書き込みます。

BytesMessage message = (BytesMessage) consumer.receive();
File outputFile = new File(outputFilePath);
OutputStream outputStream = new FileOutputStream(outputFile);

int numRead;
byte buffer[] = new byte[1024];

for (int pos = 0; pos < message.getBodyLength(); pos += buffer.length) {
    numRead = message.readBytes(buffer);
    outputStream.write(buffer, 0, numRead);
}