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>

$ 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

$ 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();

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

実装を見つけるには、お使いの環境で JNDI を設定する必要があります。これには、jndi.properties ファイルの使用、システムプロパティーの使用、または初期コンテキスト API の使用の 3 つの方法があります。

jndi.properties ファイルの使用

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

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

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 ...

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

初期コンテキスト API の使用

JNDI 初期コンテキスト API を使用してプロパティーをプログラム的に設定します。

Hashtable<Object, Object> env = new Hashtable<>();

env.put("java.naming.factory.initial", "org.apache.activemq.jndi.ActiveMQInitialContextFactory");

InitialContext context = new InitialContext(env);

Hashtable<Object, Object> env = new Hashtable<>();

env.put("java.naming.factory.initial", "org.apache.activemq.jndi.ActiveMQInitialContextFactory");

InitialContext context = new InitialContext(env);

同じ API を使用して、接続ファクトリー、キュー、およびトピックの JNDI プロパティーを設定できます。

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

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

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

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

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

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

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

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

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

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

コネクションは接続 URI を使用して設定されます。接続 URI は、クエリーパラメーターとして設定されるリモートホスト、ポート、および設定オプションのセットを指定します。利用可能なオプションの詳細は、5章設定オプション を参照してください。

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

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

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

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

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

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

フェイルオーバー URI

再接続およびフェイルオーバーに使用される URI には複数の接続 URI を含めることができます。URI の形式は、以下のとおりです。

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

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

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

JMS は、JNDI を使用してデプロイメント固有のキューとトピックリソースを検索するオプションを提供します。

JNDI でキューおよびトピック名を設定するには、以下の形式でプロパティーを作成します。この設定を jndi.properties ファイルに置くか、対応するシステムプロパティーを設定します。

queue.<lookup-name> = <queue-name>
topic.<lookup-name> = <topic-name>

queue.<lookup-name> = <queue-name>
topic.<lookup-name> = <topic-name>

たとえば、以下のプロパティーは、2 つのデプロイメント固有のリソースの名前、jobs および notifications を定義します。

queue.jobs = app1/work-items
topic.notifications = app1/updates

queue.jobs = app1/work-items
topic.notifications = app1/updates

その後、JNDI 名でリソースを検索できます。

Queue queue = (Queue) context.lookup("jobs");
Topic topic = (Topic) context.lookup("notifications");

Queue queue = (Queue) context.lookup("jobs");
Topic topic = (Topic) context.lookup("notifications");

Prefetch ポリシーオプション

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

prefetch の値は、キューの複数のコンシューマーへのメッセージの分散に影響します。値が大きいと、各コンシューマーに一度に送信されるバッチが大きくなる可能性があります。コンシューマーが異なるレートで動作している場合に、より均等にラウンドロビンの分散を実現するには、小さい値を使用します。

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

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

大きなメッセージに書き込むには、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 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);
}

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);
}