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

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

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

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

AMQ 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 JMS をファイルベースの Maven リポジトリーとしてローカルファイルシステムにインストールできます。

Hello World の例では、ブローカーへの接続を作成し、グリーティングを含むメッセージを queue キューに送信して、受信しなおします。成功すると、受信したメッセージをコンソールに出力します。

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

$ java -cp "target/classes/:target/dependency/*" org.apache.qpid.jms.example.HelloWorld
Hello world!

サンプルのソースコードは <install-dir>/examples/src/main/java ディレクトリーにあります。JNDI およびロギング設定は、<install-dir>/examples/src/main/resources ディレクトリーにあります。

JMS アプリケーションは InitialContextFactory から取得した JNDI InitialContext オブジェクトを使用して、接続ファクトリーなどの JMS オブジェクトを検索します。AMQ JMS は、org.apache.qpid.jms.jndi.JmsInitialContextFactory クラスで 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.qpid.jms.jndi.JmsInitialContextFactory

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

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

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

$ java -Djava.naming.factory.initial=org.apache.qpid.jms.jndi.JmsInitialContextFactory ...

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

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

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

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

connectionFactory.app1 = amqp://example.net:5672?jms.clientID=backend

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

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

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

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

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

amqp://example.net:5672?jms.clientID=backend

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

フェイルオーバーが設定されると、現在のサーバーへの接続が失われた場合、クライアントは別のサーバーに自動的に再接続できます。フェイルオーバー URI には接頭辞 failover: があり、括弧内にサーバー URI のコンマ区切りリストが含まれます。追加のオプションは最後に指定されます。

failover:(amqp[s]://<host>:<port>[,amqp[s]://<host>:<port>...])[?<option>=<value>[&<option>=<value>...]]

接続 URI の例と同様に、クライアントはフェイルオーバー設定で URI を使用して多数の異なる設定で設定できます。これらの設定については、以下で詳しく説明します。「フェイルオーバーオプション」 セクションは特に興味深いものです。

amqps スキームを使用して SSL/TLS 接続を指定する場合、URI からのホスト名セグメントは JVM の TLS SNI (Server Name Indication) 拡張によって使用でき、TLS ハンドシェイク時に必要なサーバーのホスト名を通信できます。SNI 拡張は、完全修飾ドメイン名 (例: myhost.mydomain) が指定されている場合に自動的に含まれますが、修飾されていない名前 (myhost など) やベア IP アドレスが使用される場合は定義されません。

これらのオプションは、Connection、Session、MessageConsumer、MessageProducer などの JMS オブジェクトの動作を制御します。

Prefetch ポリシーオプション

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

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

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

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

メッセージ ID ポリシーオプション

メッセージ ID ポリシーは、クライアントから送信されたメッセージに割り当てられたメッセージ ID のデータタイプを制御します。

Presettle ポリシーオプション

Presettle ポリシーは、AMQP の事前設定されたメッセージングセマンティクスを使用するように設定されているプロデューサーまたはコンシューマーインスタンスが設定されるタイミングを制御します。

デシリアライズポリシーオプション

デシリアライズポリシーは、シリアライズされた Java Object コンテンツで設定される受信 ObjectMessage からボディーを取得しつつ、どの Java タイプをオブジェクトストリームからデシリアライズするかを制御する手段を提供します。デフォルトでは、ボディーのデシリアライズの試行時にすべてのタイプが信頼されます。デフォルトのデシリアライズポリシーは、ホワイトリストと Java クラスまたはパッケージ名のブラックリストを指定できるようにする URI オプションを提供します。

プレーン TCP を使用してリモートサーバーに接続する場合、以下のオプションは基礎となるソケットの動作を指定します。これらのオプションは、他の設定オプションと共に接続 URI に追加されます。

amqp://localhost:5672?jms.clientID=foo&transport.connectTimeout=30000

TCP トランスポートオプションの完全なセットを以下に示します。

SSL/TLS トランスポートは、amqps URI スキームを使用して有効にします。SSL/TLS トランスポートは TCP ベースのトランスポートの機能を拡張するため、すべての TCP トランスポートオプションは SSL/TLS トランスポート URI で有効です。

amqps://myhost.mydomain:5671

SSL/TLS トランスポートオプションの完全なセットを以下に示します。

以下のオプションは、AMQP ワイヤプロトコルに関連する動作の要素に適用されます。

フェイルオーバー URI は接頭辞 failover: で始まり、括弧内に接続 URI のコンマ区切りリストが含まれます。追加のオプションは最後に指定されます。jms. で始まるオプションは、括弧外にある全体的なフェイルオーバー URI に適用され、その有効期間の Connection オブジェクトに影響します。

failover:(amqp://host1:5672,amqp://host2:5672)?jms.clientID=foo&failover.maxReconnectAttempts=20

括弧内の個々のブローカー詳細は、前のステップで定義した transport. オプションまたは amqp. オプションを使用できます。各ホストが接続されていると、これらが適用されます。

failover:(amqp://host1:5672?amqp.option=value,amqp://host2:5672?transport.option=value)?jms.clientID=foo

フェイルオーバーのすべての設定オプションは以下のとおりです。

フェイルオーバー URI は、AMQP およびトランスポートオプションをすべてネスト化されたブローカー URI に指定する手段として、ネストされたオプションの定義もサポートします。これは、同じ transport. と amqp. を使用して実行できます。非フェイルオーバーブローカー URI について前述されている URI オプションが failover.nested. で接頭辞として付けられます。たとえば、amqp.vhost オプションに同じ値を接続されたすべてのブローカーに適用するには、以下のような URI が含まれる場合があります。

failover:(amqp://host1:5672,amqp://host2:5672)?jms.clientID=foo&failover.nested.amqp.vhost=myhost

クライアントには任意の検出モジュールがあります。これは、接続するブローカー URI が初期 URI で指定されず、検出エージェントと対話して検出されるカスタムフェイルオーバー層を提供します。現在、2 つの検出エージェントの実装があります。これは、ファイルから URI をロードするファイル監視と、クライアントをリッスンするブローカーアドレスをブロードキャストするように設定された ActiveMQ 5.x ブローカーと連携するマルチキャストリスナーです。

検出を使用する際のフェイルオーバー関連のオプションの一般的なセットは前述のものと同じで、主な接頭辞は failover. から discovery. に変更されています。また、検出されたすべてのブローカー URI に共通する URI オプションを指定するために使用される nested 接頭辞があります。たとえば、エージェントの URI の詳細がないと、一般的な検出 URI は以下のようになります。

discovery:(<agent-uri>)?discovery.maxReconnectAttempts=20&discovery.discovered.jms.clientID=foo

ファイルウォッチャー検出エージェントを使用するには、以下のようなエージェント URI を作成します。

discovery:(file:///path/to/monitored-file?updateInterval=60000)

ファイルウォッチャーの検出エージェントの URI オプションを以下に示します。

ActiveMQ 5.x ブローカーでマルチキャスト検出エージェントを使用するには、以下のようなエージェント URI を作成します。

discovery:(multicast://default?group=default)

上記のマルチキャストエージェント URI のホストとして default をホストとして使用することは、エージェントがデフォルトの 239.255.2.3:6155 に置き換える特別な値であることに注意してください。これを変更して、マルチキャスト設定で使用している実際の IP アドレスとポートを指定できます。

マルチキャスト検出エージェントの URI オプションを以下に示します。

通常、JMS を使用するアプリケーションは JNDI を使用してアプリケーションが使用する ConnectionFactory および Destination オブジェクトを取得します。これにより、設定がプログラムから分離され、特定のクライアント実装から分離されます。

これらの例を使用する場合は、前述のように JNDI コンテキストを設定するために、jndi.properties という名前のファイルをクラスパスに配置する必要があります。

jndi.properties ファイルの内容は、以下に示す内容と一致している必要があります。これにより、クライアントの InitialContextFactory 実装を使用し、ローカルサーバーに接続する ConnectionFactory を設定し、queue という名前の宛先キューを定義します。

# Configure the InitialContextFactory class to use
java.naming.factory.initial = org.apache.qpid.jms.jndi.JmsInitialContextFactory

# Configure the ConnectionFactory
connectionfactory.myFactoryLookup = amqp://localhost:5672

# Configure the destination
queue.myDestinationLookup = queue

この例では、最初に JNDI Context を作成し、これを使用して ConnectionFactory および Destination を検索し、ファクトリーを使用して Connection を作成および起動し、Session を作成します。次に、MessageProducer が Destination に作成され、それを使用してメッセージが送信されます。その後、Connection が閉じられ、プログラムは終了します。

この Sender の例の実行可能なバリアントは、<install-dir>/examplesディレクトリー にあります。3章スタートガイド .

package org.jboss.amq.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class Sender {
  public static void main(String[] args) throws Exception {
    try {
      Context context = new InitialContext(); 1

      ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
      Destination destination = (Destination) context.lookup("myDestinationLookup"); 2

      Connection connection = factory.createConnection("<username>", "<password>");
      connection.setExceptionListener(new MyExceptionListener());
      connection.start(); 3

      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 4

      MessageProducer messageProducer = session.createProducer(destination); 5

      TextMessage message = session.createTextMessage("Message Text!"); 6
      messageProducer.send(message, DeliveryMode.NON_PERSISTENT,
                           Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); 7

      connection.close(); 8
    } catch (Exception exp) {
      System.out.println("Caught exception, exiting.");
      exp.printStackTrace(System.out);
      System.exit(1);
    }
  }

  private static class MyExceptionListener implements ExceptionListener {
    @Override
    public void onException(JMSException exception) {
      System.out.println("Connection ExceptionListener fired, exiting.");
      exception.printStackTrace(System.out);
      System.exit(1);
    }
  }
}

これは単なる例であることに注意してください。実際のアプリケーションは、通常有効期限の長い MessageProducer を使用し、時間の経過とともに多数のメッセージを送信します。通常、メッセージごとに Connection、Session、および MessageProducer を開くことは効率的ではありません。

この例では、JNDI コンテキストを作成し、そのコンテキストを使用して ConnectionFactory および Destination を検索し、ファクトリーを使用して Connection を作成および起動し、Session を作成します。次に、Destination に MessageConsumer が作成され、それを使用してメッセージを受信し、その内容がコンソールに出力されます。その後、Connection が閉じられ、プログラムは終了します。送信例 と同じ JNDI 設定が使用されます。

この Receiver の例の実行可能なバリアントは、以前に 3章スタートガイド に記載されている Hello World の例とともに、クライアントディストリビューションのサンプルディレクトリーに含まれます。

package org.jboss.amq.example;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.naming.Context;
import javax.naming.InitialContext;

public class Receiver {
  public static void main(String[] args) throws Exception {
    try {
      Context context = new InitialContext(); 1

      ConnectionFactory factory = (ConnectionFactory) context.lookup("myFactoryLookup");
      Destination destination = (Destination) context.lookup("myDestinationLookup"); 2

      Connection connection = factory.createConnection("<username>", "<password>");
      connection.setExceptionListener(new MyExceptionListener());
      connection.start(); 3

      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 4

      MessageConsumer messageConsumer = session.createConsumer(destination); 5

      Message message = messageConsumer.receive(5000); 6

      if (message == null) { 7
        System.out.println("A message was not received within given time.");
      } else {
        System.out.println("Received message: " + ((TextMessage) message).getText());
      }

      connection.close(); 8
    } catch (Exception exp) {
      System.out.println("Caught exception, exiting.");
      exp.printStackTrace(System.out);
      System.exit(1);
    }
 }

  private static class MyExceptionListener implements ExceptionListener {
    @Override
    public void onException(JMSException exception) {
      System.out.println("Connection ExceptionListener fired, exiting.");
      exception.printStackTrace(System.out);
      System.exit(1);
    }
  }
}

これは単なる例であることに注意してください。実際のアプリケーションは、通常有効期限の長い MessageConsumer を使用し、時間の経過とともに多くのメッセージを受信します。通常、各メッセージの Connection、Session、および MessageConsumer を開くと効率的ではありません。

クライアントは、適切に設定されたサーバーで使用される場合に Kerberos を使用して認証するように設定できます。Kerberos を有効にするには、以下の手順に従います。

使用される正確な設定は、接続に対してクレデンシャルを確立する方法と、使用中の特定の LoginModule によって異なります。Oracle Krb5LoginModule の詳細は、Oracle Krb5LoginModule class reference を参照してください。IBM Java 8 Krb5LoginModule の詳細は、IBM Krb5LoginModule class reference を参照してください。

LoginModule を設定して、プリンシパルの指定や既存のチケットキャッシュまたはキータブを使用するかどうかなど、Kerberos プロセスに使用する認証情報を確立できます。しかし、LoginModule 設定が必要なすべてのクレデンシャルを確立する手段を提供しない場合、ConnectionFactory を使用して Connection を作成する場合、クライアント Connection オブジェクトからユーザー名とパスワードの値が渡されます。

Kerberos は認証の目的でのみサポートされることに注意してください。暗号化には SSL/TLS 接続を使用します。

以下の接続 URI オプションを使用して、Kerberos 認証プロセスに影響を与えることができます。

これまでで説明した amqp. オプションおよび transport. オプションと同様に、これらのオプションはホストごとに指定するか、フェイルオーバー URI のすべてのホストネストされたオプションとして指定する必要があります。

SSL/TLS 接続は、パフォーマンスを向上させるためにネイティブの OpenSSL 実装を使用するように設定できます。OpenSSL を使用するには、transport.useOpenSSL オプションを有効にし、OpenSSL サポートライブラリーをクラスパスで利用できるようにする必要があります。

Red Hat Enterprise Linux でインストールされた OpenSSL ライブラリーを使用するには、openssl パッケージおよび apr RPM パッケージをインストールし、以下の依存関係を POM ファイルに追加します。

<dependency>
  <groupId>io.netty</groupId>
  <artifactId>netty-tcnative</artifactId>
  <version>2.0.29.Final-redhat-00001</version>
</dependency>

OpenSSL ライブラリーの実装の一覧 は、Netty プロジェクトから入手できます。

メッセージングシステムは、メッセージ確認を使用して、メッセージの送信ゴールが完全に行われるかどうかを追跡します。

メッセージが送信されると、メッセージが送信されてから確認応答するまでの期間が発生します (メッセージは in flight (インフライト) です。その間ネットワーク接続が失われた場合、メッセージ配信のステータスは不明となり、配信が完了するまでアプリケーションコードで特別な処理が必要になる場合があります。

以下のセクションでは、接続に失敗した場合にメッセージ配信の条件を説明します。

承認されていない配信とトランザクション以外のプロデューサー

メッセージが進行中の場合は、送信タイムアウトが設定されておらず、経過していないと、再接続後に再度送信されます。

ユーザーアクションは不要です。

トランザクションがコミットされていないトランザクションとのトランザクションプロデューサー

メッセージが進行中の場合は、再接続後に再度送信されます。新しいトランザクションで送信が最初に送信された場合は、再接続後に通常通りに送信が続行されます。トランザクションに以前の送信がある場合、トランザクションは失敗とみなされ、後続のコミット操作によって TransactionRolledBackException が出力されます。

配信を図るには、失敗したトランザクションに属するメッセージを再送信する必要があります。

保留中のコミットとトランザクションプロデューサー

コミットがフライトの場合、トランザクションは失敗とみなされ、後続のコミット操作によって TransactionRolledBackException が出力されます。

配信を図るには、失敗したトランザクションに属するメッセージを再送信する必要があります。

承認されていない配信のある非トランザクションコンシューマー

メッセージが受信してもまだ確認応答されない場合、メッセージを承認するとエラーは生成されませんが、クライアントによるアクションはありません。

受信したメッセージは確認されていないため、プロデューサーは再送信する可能性があります。重複を回避するために、ユーザーはメッセージ ID で重複メッセージを除外する必要があります。

コミットされていないトランザクションを使用したトランザクションコンシューマー

アクティブなトランザクションがまだコミットされていない場合は、失敗とみなされ、保留中の承認はドロップされます。後続のコミット操作によって TransactionRolledBackException が出力されます。

プロデューサーは、トランザクションに属するメッセージを再送信する可能性があります。重複を回避するために、ユーザーはメッセージ ID で重複メッセージを除外する必要があります。

保留中のコミットのあるトランザクションコンシューマー

コミットがフライトの場合、トランザクションは失敗とみなされます。後続のコミット操作によって TransactionRolledBackException が出力されます。

プロデューサーは、トランザクションに属するメッセージを再送信する可能性があります。重複を回避するために、ユーザーはメッセージ ID で重複メッセージを除外する必要があります。

クライアントは、JMS 仕様で定義されたもの以外の追加のセッション確認モードをサポートします。

個別確認応答

このモードでは、セッションが CLIENT_ACKNOWLEDGE モードの場合に使用される Message.acknowledge() メソッドを使用して、メッセージを個別に承認する必要があります。CLIENT_ACKNOWLEDGE モードとは異なり、ターゲットメッセージのみが確認されます。それ以外の配信されたメッセージはすべて承認されていないままになります。このモードを有効にするために使用される整数値は 101 です。

connection.createSession(false, 101);

確認なし

このモードでは、クライアントにディスパッチされる前にサーバーでメッセージを受け入れ、承認はクライアントによって実行されません。クライアントは、このモードをアクティブにするために 2 つの整数値をサポートします (100 および 257)。

connection.createSession(false, 100);

デバッグ時には、Qpid Proton AMQP 1.0 ライブラリーから追加のプロトコルトレースロギングを有効にすると便利です。これを行うには、以下の 2 つの方法があります。

入力および出力バイトの低レベルのトレースを出力するようにクライアントを設定することもできます。これを有効にするには、接続 URI に transport.traceBytes=true オプションを追加し、org.apache.qpid.jms.transports.netty.NettyTcpTransport ロガーをログレベル DEBUG に設定します。

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

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

AMQP メッセージ は AMQP タイプシステム を使用して設定されます。この共通の形式を持つことは、異なる言語の AMQP クライアントが相互運用できる理由の 1 つです。本セクションでは、クライアントと他の AMQP クライアントの使用を支援するために、使用されるさまざまな JMS メッセージタイプに関連して、クライアントによって送受信された AMQP ペイロードに関する動作を文書化します。

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

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