AMQ Broker の設定

ブローカーの設定ファイルはすべて <broker-instance-dir>/etc に保存されます。これらの設定ファイルで設定を編集すると、ブローカーを設定できます。

各ブローカーインスタンスは以下の設定ファイルを使用します。

broker.xml 設定ファイルを編集して、ブローカーの機能の大半を設定します。このファイルにはデフォルト設定が含まれています。これはブローカーを起動し、操作するには十分です。ただし、デフォルト設定の一部を変更し、お使いの環境にブローカーを設定するために新しい設定を追加する必要があります。

デフォルトでは、broker.xml には、以下の機能のデフォルト設定が含まれます。

デフォルトのメッセージ永続性設定

デフォルトでは、AMQ Broker の永続性は、ディスク上のファイルセットで設定される追加のみのファイルジャーナルを使用します。ジャーナルは、メッセージ、トランザクション、およびその他の情報を保存します。

<configuration ...>

   <core ...>

      ...

      <persistence-enabled>true</persistence-enabled>

      <!-- this could be ASYNCIO, MAPPED, NIO
           ASYNCIO: Linux Libaio
           MAPPED: mmap files
           NIO: Plain Java Files
       -->
      <journal-type>ASYNCIO</journal-type>

      <paging-directory>data/paging</paging-directory>

      <bindings-directory>data/bindings</bindings-directory>

      <journal-directory>data/journal</journal-directory>

      <large-messages-directory>data/large-messages</large-messages-directory>

      <journal-datasync>true</journal-datasync>

      <journal-min-files>2</journal-min-files>

      <journal-pool-files>10</journal-pool-files>

      <journal-file-size>10M</journal-file-size>

      <!--
       This value was determined through a calculation.
       Your system could perform 8.62 writes per millisecond
       on the current journal configuration.
       That translates as a sync write every 115999 nanoseconds.

       Note: If you specify 0 the system will perform writes directly to the disk.
             We recommend this to be 0 if you are using journalType=MAPPED and journal-datasync=false.
      -->
      <journal-buffer-timeout>115999</journal-buffer-timeout>

      <!--
        When using ASYNCIO, this will determine the writing queue depth for libaio.
       -->
      <journal-max-io>4096</journal-max-io>

      <!-- how often we are looking for how many bytes are being used on the disk in ms -->
      <disk-scan-period>5000</disk-scan-period>

      <!-- once the disk hits this limit the system will block, or close the connection in certain protocols
           that won't support flow control. -->
      <max-disk-usage>90</max-disk-usage>

      <!-- should the broker detect dead locks and other issues -->
      <critical-analyzer>true</critical-analyzer>

      <critical-analyzer-timeout>120000</critical-analyzer-timeout>

      <critical-analyzer-check-period>60000</critical-analyzer-check-period>

      <critical-analyzer-policy>HALT</critical-analyzer-policy>

      ...

  </core>

</configuration>

デフォルトのアクセプター設定

ブローカーは、acceptor 設定要素を使用して受信クライアント接続をリッスンし、クライアントが接続を作成するためにポートおよびプロトコルを定義します。デフォルトでは、AMQ Broker にはサポートされる各メッセージングプロトコルのアクセプターの設定が含まれます。

<configuration ...>

   <core ...>

      ...

      <acceptors>

        <!-- Acceptor for every supported protocol -->
        <acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300</acceptor>

        <!-- AMQP Acceptor. Listens on default AMQP port for AMQP traffic -->
        <acceptor name="amqp">tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300</acceptor>

        <!-- STOMP Acceptor -->
        <acceptor name="stomp">tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true</acceptor>

        <!-- HornetQ Compatibility Acceptor. Enables HornetQ Core and STOMP for legacy HornetQ clients. -->
        <acceptor name="hornetq">tcp://0.0.0.0:5445?anycastPrefix=jms.queue.;multicastPrefix=jms.topic.;protocols=HORNETQ,STOMP;useEpoll=true</acceptor>

        <!-- MQTT Acceptor -->
        <acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true</acceptor>

      </acceptors>

      ...

  </core>

</configuration>

デフォルトのセキュリティー設定

AMQ Broker には、アドレスに基づいてキューにセキュリティーを適用するための柔軟なロールベースのセキュリティーモデルが含まれています。デフォルト設定ではワイルドカードを使用して amq ロールをすべてのアドレスに適用します (数値記号 # で表されます)。

<configuration ...>

   <core ...>

      ...

      <security-settings>
         <security-setting match="#">
            <permission type="createNonDurableQueue" roles="amq"/>
            <permission type="deleteNonDurableQueue" roles="amq"/>
            <permission type="createDurableQueue" roles="amq"/>
            <permission type="deleteDurableQueue" roles="amq"/>
            <permission type="createAddress" roles="amq"/>
            <permission type="deleteAddress" roles="amq"/>
            <permission type="consume" roles="amq"/>
            <permission type="browse" roles="amq"/>
            <permission type="send" roles="amq"/>
            <!-- we need this otherwise ./artemis data imp wouldn't work -->
            <permission type="manage" roles="amq"/>
         </security-setting>
      </security-settings>

      ...

  </core>

</configuration>

デフォルトのメッセージアドレス設定

AMQ Broker には、作成されたキューまたはトピックに適用されるデフォルトの設定セットを確立するデフォルトアドレスが含まれています。

さらに、デフォルト設定では、DLQ (Dead Letter Queue) の 2 つのキューが、既知の宛先で到達するメッセージを処理します。また、Expiry Queue は有効期限を過ぎたメッセージを保持しているため、元の宛先にルーティングしないでください。

<configuration ...>

   <core ...>

      ...

      <address-settings>
         ...
         <!--default for catch all-->
         <address-setting match="#">
            <dead-letter-address>DLQ</dead-letter-address>
            <expiry-address>ExpiryQueue</expiry-address>
            <redelivery-delay>0</redelivery-delay>
            <!-- with -1 only the global-max-size is in use for limiting -->
            <max-size-bytes>-1</max-size-bytes>
            <message-counter-history-day-limit>10</message-counter-history-day-limit>
            <address-full-policy>PAGE</address-full-policy>
            <auto-create-queues>true</auto-create-queues>
            <auto-create-addresses>true</auto-create-addresses>
            <auto-create-jms-queues>true</auto-create-jms-queues>
            <auto-create-jms-topics>true</auto-create-jms-topics>
         </address-setting>
      </address-settings>

      <addresses>
         <address name="DLQ">
            <anycast>
               <queue name="DLQ" />
            </anycast>
         </address>
         <address name="ExpiryQueue">
            <anycast>
               <queue name="ExpiryQueue" />
            </anycast>
         </address>
      </addresses>

   </core>

</configuration>

デフォルトでは、ブローカーは 5000 ミリ秒ごとに設定ファイルの変更をチェックします。ブローカーが設定ファイルの最後の変更されたタイムスタンプの変更を検出する場合、ブローカーは設定変更の実行を決定します。この場合、ブローカーは設定ファイルを再読み込みして変更を有効にします。

ブローカーが broker.xml 設定ファイルを再読み込みすると、以下のモジュールを再読み込みします。

以下の手順では、ブローカーが broker.xml 設定ファイルへの変更をチェックする間隔を変更する方法を説明します。

共通の設定設定を共有する複数のブローカーがある場合は、個別のファイルに共通設定を定義し、各ブローカーの broker.xml 設定ファイルにこれらのファイルを含めることができます。

ブローカー間で共有できる最も一般的な設定には、以下が含まれます。

$ touch -m <broker-instance-dir>/etc/broker.xml

本書では、sudo コマンドおよびファイルパスについて、以下の表記慣例を使用します。

sudo コマンド

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

sudo の使用の詳細は、sudo コマンド を参照してください。

本書におけるファイルパスの使用

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

AMQ Broker でネットワーク接続を説明する際に最も重要な概念の 1 つが acceptor です。アクセプターは、ブローカーへの接続方法を定義します。以下は、BROKER_INSTANCE_DIR/etc/broker.xml 設定ファイル内で見つかった acceptor の典型的な設定です。

<acceptors>
   <acceptor name="example-acceptor">tcp://localhost:61617</acceptor>
</acceptors>

各 acceptor は acceptors 要素内でグループ化されることに注意してください。サーバーごとに一覧表示できるアクセプターの数の上限はありません。

アクセプターの設定

acceptor に定義された URI のクエリー文字列にキーと値のペアを追加して acceptor を設定します。以下の例のように、セミコロン (';') を使用して複数のキーと値のペアを区切ります。sslEnabled=true で始まる URI の最後に複数のキーと値のペアを追加して、SSL/TLS のアクセプターを設定します。

<acceptor name="example-acceptor">tcp://localhost:61617?sslEnabled=true;key-store-path=/path</acceptor>

connector 設定パラメーターの詳細は、Acceptor and Connector Configuration Parameters を参照してください。

アクセプターはサーバーが接続を受け入れる方法を定義しますが、connector はクライアントによって使用され、サーバーへの接続方法を定義します。

以下は、BROKER_INSTANCE_DIR/etc/broker.xml 設定ファイルで定義される通常の connector です。

<connectors>
   <connector name="example-connector">tcp://localhost:61617</connector>
</connectors>

コネクターは connectors 要素内で定義されることに注意してください。サーバーごとのコネクター数の上限はありません。

コネクターはクライアントによって使用されますが、アクセプターと同様にサーバーで設定されます。重要な理由には、以下の 2 つの理由があります。

コネクターの設定

アクセプターと同様に、コネクターには URI のクエリー文字列に割り当てられた設定があります。以下は、tcpNoDelay パラメーターが false に設定されている connector の例になります。これにより、この接続の Nagle アルゴリズムが無効になります。

<connector name="example-connector">tcp://localhost:61616?tcpNoDelay=false</connector>

connector 設定パラメーターの詳細は、Acceptor and Connector Configuration Parameters を参照してください。

AMQ Broker は Netty を使用して基本的な、暗号化されていない TCP ベースの接続を提供します。この接続は、ブロッキング Java IO または新しい非ブロッキング Java NIO を使用するように設定できます。多くの同時接続でスケーラビリティーを向上させるために、Java NIO が推奨されます。ただし、古い IO を使用すると、何万もの同時接続のサポートを受けても NIO より優れたレイテンシーが発生することがあります。

信頼できないネットワーク全体で接続を実行している場合は、TCP ネットワーク接続が暗号化されないことを認識する必要があります。暗号化が優先される場合には、SSL または HTTPS 設定を使用して、この接続で送信されたメッセージを暗号化することを検討してください。詳細は、トランスポート層セキュリティーの設定 を参照してください。

TCP 接続を使用すると、すべての接続がクライアント側から開始されます。つまり、サーバーはクライアントへの接続を開始しません。これは、ある方向から接続を強制するファイアウォールポリシーで適切に機能します。

TCP 接続では、コネクター URI のホストおよびポートは接続に使用されるアドレスを定義します。

以下の例では、acceptor は TCP 接続として設定されます。この acceptor で設定されたブローカーは、IP 10.10.10.1 およびポート 61617 への TCP 接続を行うクライアントを受け入れます。

<acceptors>
  <acceptor name="tcp-acceptor">tcp://10.10.10.1:61617</acceptor>
  ...
</acceptors>

同様の方法で TCP を使用するようにコネクターを設定します。

<connectors>
  <connector name="tcp-connector">tcp://10.10.10.2:61617</connector>
  ...
</connectors>

上記の connector は、指定された IP およびポート 10.10.10.2 :61617 に TCP 接続を確立する際に、クライアントまたはブローカー自体によって 参照されます。

TCP 接続で利用可能な設定パラメーターの詳細は、Acceptor and Connector Configuration Parameters を参照してください。ほとんどのパラメーターはアクセプターまたはコネクターで使用することができますが、アクセプターでのみ機能するものもあります。

HTTP プロトコルを介した HTTP 接続トンネルは、ファイアウォールが HTTP トラフィックのみを許可する場合に便利です。単一ポートサポートでは、AMQ Broker は HTTP が使用されているかどうかを自動的に検出するため、HTTP のネットワーク接続の設定は TCP の接続の設定と同じです。HTTP の使用方法に関する詳細は、INSTALL_DIR/examples/features/standard/ にある http-transport の例を参照してください。

以下の例では、ブローカーは IP アドレス 10.10.10.1 でポート 80 に接続するクライアントからの HTTP 通信を受け入れます。さらに、ブローカーは HTTP プロトコルが使用されていることを自動検出し、それに応じてクライアントと通信します。

<acceptors>
  <acceptor name="http-acceptor">tcp://10.10.10.1:80</acceptor>
  ...
</acceptors>

HTTP のコネクターの設定は TCP の場合と同じです。

<connectors>
  <connector name="http-connector">tcp://10.10.10.2:80</connector>
  ...
</connectors>

上記の例の設定を使用すると、ブローカーは IP アドレス 10.10.10.2 でポート 80 へのアウトバウンド HTTP 接続を作成し ます。

HTTP 接続は TCP と同じ設定パラメーターを使用しますが、いくつかの設定パラメーターもあります。HTTP 関連およびその他の設定パラメーターの詳細は、Acceptor and Connector Configuration Parameters を参照してください。

SSL/TLS を使用するように接続を設定することもできます。詳細は、トランスポート層セキュリティーの設定 を参照してください。

たとえば、高可用性ソリューションの一部として、複数のブローカーが同じ仮想マシンに共存する場合に、In-VM 接続を使用できます。仮想マシン内のマシン接続は、ブローカーと同じ JVM で実行されているローカルクライアントでも使用できます。in-VM 接続では、URI の認証局部分は一意のサーバー ID を定義します。実際、URI の他の部分は必要ありません。

<acceptors>
  <acceptor name="in-vm-acceptor">vm://0</acceptor>
  ...
</acceptors>

上記の acceptor の例では、ID が 0 のサーバーからの接続を受け入れるようにブローカーに指示します。他のサーバーはブローカーと同じ仮想マシンで実行されている必要があります。

コネクターを in-vm コネクションとして設定する場合は、同じ構文に従います。

<connectors>
  <connector name="in-vm-connector">vm://0</connector>
  ...
</connectors>

上記の例の connector は、クライアントが同じ仮想マシンにある ID が 0 のサーバーへの in-VM 接続を確立する方法を定義します。クライアントはアプリケーションまたはブローカーになります。

コネクターは、クライアントアプリケーションでも間接的に使用されます。サーバー側で connector を定義することなく、JMS 接続ファクトリーを直接クライアント側に設定できます。

Map<String, Object> connectionParams = new HashMap<String, Object>();

connectionParams.put(org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants.PORT_PROP_NAME, 61617);

TransportConfiguration transportConfiguration =
    new TransportConfiguration(
    "org.apache.activemq.artemis.core.remoting.impl.netty.NettyConnectorFactory", connectionParams);

ConnectionFactory connectionFactory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, transportConfiguration);

Connection jmsConnection = connectionFactory.createConnection();

使用する前に、プロトコルをネットワーク接続に関連付ける必要があります。(ネットワーク接続の作成および設定方法の詳細は、ネットワーク接続: アクセプターおよびコネクター を参照してください。) BROKER_INSTANCE_DIR/etc/broker.xml ファイルにあるデフォルト設定には、すでに定義された接続が複数含まれています。便宜上、AMQ Broker にはサポートされる各プロトコルのアクセプターと、すべてのプロトコルをサポートするデフォルトのアクセプターが含まれます。

デフォルトのアクセプターの概要

以下は、broker.xml 設定ファイルにデフォルトで含まれるアクセプターです。

<configuration>
  <core>
    ...
    <acceptors>

      <!-- All-protocols acceptor -->
      <acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300</acceptor>

      <!-- AMQP Acceptor. Listens on default AMQP port for AMQP traffic -->
      <acceptor name="amqp">tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;amqpCredits=1000;amqpLowCredits=300</acceptor>

      <!-- STOMP Acceptor -->
      <acceptor name="stomp">tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true</acceptor>

      <!-- HornetQ Compatibility Acceptor. Enables HornetQ Core and STOMP for legacy HornetQ clients. -->
      <acceptor name="hornetq">tcp://0.0.0.0:5445?anycastPrefix=jms.queue.;multicastPrefix=jms.topic.;protocols=HORNETQ,STOMP;useEpoll=true</acceptor>

      <!-- MQTT Acceptor -->
      <acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true</acceptor>

    </acceptors>
    ...
  </core>
</configuration>

特定のネットワーク設定でプロトコルを有効にする唯一の要件は、protocols パラメーターを アクセプターの URI に追加することです。パラメーターの値は、プロトコル名のコンマ区切りリストである必要があります。protocol パラメーターが URI から省略される場合、すべてのプロトコルが有効になります。

たとえば、AMQP プロトコルを使用して 3232 番ポートでメッセージを受信するためにアクセプターを作成するには、以下の手順に従います。

<acceptor name="ampq">tcp://0.0.0.0:3232?protocols=AMQP</acceptor>

デフォルトのアクセプターの追加パラメーター

最小限のアクセプター設定では、接続 URI の一部としてプロトコルを指定します。ただし、broker.xml 設定ファイルのデフォルトのアクセプターには追加のパラメーターが設定されています。以下の表は、デフォルトのアクセプターに設定された追加パラメーターの詳細を示しています。

ブローカーは AMQP 1.0 仕様をサポートします。AMQP リンクは、ソースとターゲット間のメッセージ (クライアントとブローカー) の一方向プロトコルです。

<acceptors>
  <acceptor name="amqp-acceptor">tcp://localhost:5672?protocols=AMQP</acceptor>
  ...
</acceptors>

上記の例では、ブローカーはデフォルトの AMQP ポートであるポート 5672 で AMQP 1.0 クライアントを受け入れます。

AMQP リンクには、送信者と受信側の 2 つのエンドポイントがあります。送信者がメッセージを送信する場合、ブローカーはこれを内部形式に変換するため、ブローカーの宛先に転送できます。受信側はブローカーの宛先に接続し、メッセージを配信前に AMQP に戻します。

AMQP リンクが動的の場合、一時キューが作成され、リモートソースまたはリモートターゲットアドレスのいずれかが一時キューの名前に設定されます。リンクが動的ではない場合、リモートターゲットまたはソースのアドレスがキューに使用されます。リモートターゲットまたはソースが存在しない場合は、例外が発生します。

リンクターゲットは、基礎となるセッションをトランザクションとして処理するために使用される Coordinator でも、ロールバックまたはコミットします。

プロトコルとその機能の詳細は、AMQP 1.0 仕様を参照してください。

ブローカーは MQTT v3.1.1(および古い v3.1 コードメッセージ形式) をサポートします。MQTT は軽量のクライアントからサーバーへ、パブリッシュ/サブスクライブメッセージングプロトコルです。MQTT は、メッセージングのオーバーヘッドおよびネットワークトラフィック、およびクライアントのコードフットプリントを削減します。このような理由から、MQTT は、センサーやアクチュエーターなどのデバイスを制限するのが適しており、この ng(IoT) のインターネット向けの de facto 標準通信プロトコルがすばやく考えられます。

<acceptors>
  <acceptor name="mqtt">tcp://localhost:1883?protocols=MQTT</acceptor>
  ...
</acceptors>

MQTT には、以下を含む便利な機能が多数含まれています。

MQTT プロトコルに関する情報の最適なソースは、仕様にあります。MQTT v3.1.1 仕様は、OASIS Web サイト からダウンロードできます。

ブローカーは OpenWire protocol プロトコルをサポートします。これにより、JMS クライアントはブローカーと直接対話できます。このプロトコルを使用して、古いバージョンの AMQ Broker と通信します。

現在、AMQ Broker は標準の JMS API のみを使用する OpenWire クライアントをサポートします。

上記の例では、ブローカーは受信 OpenWire コマンドのポート 61616 でリッスンします。

詳細は、INSTALL_DIR/examples/protocols/openwire にある例を参照してください。

STOMP は、STOMP クライアントが STOMP Broker と通信できるようにするテキスト指向のワイヤプロトコルです。ブローカーは STOMP 1.0、1.1、および 1.2 をサポートします。STOMP クライアントは複数の言語やプラットフォームで利用できます。これにより、相互運用性の選択肢があります。

<acceptors>
  <acceptor name="stomp-acceptor">tcp://localhost:61613?protocols=STOMP</acceptor>
  ...
</acceptors>

上記の例では、ブローカーはポート 61613 の STOMP 接続を受け入れます。

STOMP を使用したブローカーの設定例については、 INSTALL_DIR/examples/protocols にある stomp の例を参照してください。

<acceptors>
  <acceptor name="stomp-acceptor">tcp://localhost:61613?protocols=STOMP;stompEnableMessageId=true</acceptor>
  ...
</acceptors>

amq-message-id : STOMP12345

<acceptors>
  <acceptor name="stomp-acceptor">tcp://localhost:61613?protocols=STOMP;connectionTTL=20000</acceptor>
  ...
</acceptors>

<configuration ...>
  ...
  <core ...>
    ...
    <connection-ttl-override>30000</connection-ttl-override>
    ...
  </core>
<configuration>

SEND
destination:/my/stomp/queue

hello queue a
^@

SUBSCRIBE
destination: /other/stomp/queue
ack: client

^@

アドレスおよびキューを設定する場合は、以下の要件に注意してください。

ポイントツーポイントメッセージングは、プロデューサーによって送信されたメッセージが 1 つのコンシューマーのみを持つ一般的なシナリオです。AMQP および JMS メッセージプロデューサーおよびコンシューマーは、たとえば、ポイントツーポイントメッセージングキューを使用できます。address に anycast ルーティングタイプを定義して、そのキューがポイントツーポイント方式でメッセージを受信できるようにします。

anycast を使用してアドレスでメッセージを受信すると、ブローカーはアドレスに関連付けられたキューを見つけ、そこにメッセージをルーティングします。コンシューマーがアドレスから使用の要求を出すと、ブローカーは関連するキューを見つけ、このキューを適切なコンシューマーに関連付けます。複数のコンシューマーが同じキューに接続されている場合、コンシューマーが均等に処理できる前提で、メッセージが各コンシューマー間で均等に分散されます。

図4.1 Point-to-Point

パブリッシュ/サブスクライブのシナリオでは、メッセージはアドレスにサブスクライブしているすべてのコンシューマーに送信されます。JMS トピックおよび MQTT サブスクリプションは、パブリッシュ/サブスクライブメッセージングの 2 つの例です。メッセージが multicast ルーティングタイプのアドレスで受信されると、AMQ ブローカはメッセージのコピーを各キューにルーティングします。コピーのオーバーヘッドを減らすために、各キューにはメッセージへの参照のみが送信され、完全なコピーは送信されません。

図4.2 パブリッシュ - サブスクライブ

anycast ルーティングタイプを使用するアドレスで複数のキューを定義できます。anycast アドレスに送信されるメッセージは、関連するすべてのキューに均等に分散されます。後で説明する完全修飾キュー名を使用して、クライアントを特定のキューに接続させることができます。複数のコンシューマーが同じキューに接続している場合、AMQ Broker はそれらの間でメッセージを配信します。

図4.3 2 つのキューによるポイントツーポイント

ポイントツーポイントとパブリッシュ/サブスクライブの両方のセマンティクスを有効にしてアドレスを定義することができます。通常は推奨されませんが、これは、orders という名前の JMS キューと、orders という名前の JMS トピックが必要な場合などに役立ちます。ルーティングの種類が異なるため、アドレスが異なっているように見えます。

JMS クライアントの例を使用すると、JMS キュープロデューサにーよって送信されるメッセージは、anycast ルーティングタイプを使用してルーティングされます。JMS トピックプロデューサーによって送信されるメッセージは、multicast ルーティング タイプを使用します。さらに、JMS トピックコンシューマーがアタッチすると、独自のサブスクリプションキューにアタッチされます。ただし、JMS キューコンシューマーは anycast キューに割り当てられます。

図4.4 ポイントツーポイントおよびパブリッシュ - サブスクライブ

以下の XML の抜粋は、BROKER_INSTANCE_DIR/etc/broker.xml で、anycast と multicast の両方のルーティングタイプを使用するアドレスの設定の例です。通常、サブスクリプションキューはオンデマンドで作成されるため、multicast ルーティングタイプ内に特定の queue 要素をリストする必要はありません。

<configuration ...>
  <core ...>
    ...
    <address name="foo.orders">
      <anycast>
        <queue name="orders"/>
      </anycast>
      <multicast/>
    </address>
  </core>
</configuration>

ほとんどの場合、プロトコルマネージャーは、クライアントが最初にアドレスへのサブスクライブを要求したときに、自動的にサブスクリプションキューを作成するため、サブスクリプションキューを事前に作成する必要はありません。詳細は 、プロトコルマネージャーとアドレス を参照してください。永続サブスクリプションの場合、生成されたキュー名はクライアント ID とアドレスを連結したものです。

永続サブスクリプションキューの設定

キューが永続サブスクリプションとして設定されると、ブローカーは非アクティブなサブスクライバーのメッセージを格納し、再接続時にサブスクライバーに配信します。そのため、クライアントはサブスクライブ後にキューへ配信される各メッセージを受信することが保証されます。

共有されていない永続的なサブスクリプションの設定

ブローカーは、一度に複数のコンシューマーがキューに接続できないように設定することができます。したがって、このように設定されたキューへのサブスクリプションは共有されません。

非永続的なサブスクリプションキューの設定

非永続的なサブスクリプションは、通常、一時キューを作成および削除する関連するプロトコルマネージャーによって管理されます。

ただし、非永続サブスクリプションキューのように動作するキューを事前に作成する場合は、キューで purge-on-no-consumers 属性を使用できます。purge-on-no-consumers が true に設定されている場合、コンシューマーが接続されるまでキューはメッセージの受信を開始しません。さらに、最後のコンシューマーがキューから切断されると、キューはパージされます (つまり、メッセージが削除されます)。新しいコンシューマーがキューにアタッチされるまで、キューはそれ以上のメッセージを受信しません。

内部的には、ブローカーはアドレスのクライアント要求を特定のキューにマッピングします。ブローカーは、メッセージの送信先となるキューまたはメッセージを受信するキューをクライアントに代わって決定します。ただし、より高度なユースケースでは、クライアントが直接キューを指定する必要がある場合があります。このような場合、クライアントは、アドレス名とキュー名の両方を :: で区切って指定して、完全修飾キュー名 (FQQN) を使用できます。

部分的な順序付けのみが必要とされるキュー全体のメッセージの処理には、queue sharding を使用するのが一般的なパターンです。AMQ Broker では、単一の論理キューとして機能する anycast アドレスを作成してこれを実現できますが、多くの基礎となる物理キューでサポートされます。

上記の設定を使用すると、sharded に送信されるメッセージは q1、q2、および q3 に均等に分散されます。クライアントは、完全修飾キュー名の使用 時に、特定の物理キューに直接接続して、その特定のキューにのみ送信されたメッセージを受け取ることができます。

特定のメッセージを特定のキューに結びつけるために、クライアントはメッセージごとにメッセージグループを指定できます。ブローカーは、グループ化されたメッセージを同じキューにルーティングし、1 つのコンシューマーはそれをすべて処理します。詳細は、メッセージのグループ化 に関する章を参照してください。

last value queueとは、last value プロパティー名が同じメッセージでよりあたらいいものがキューに配置されると、キューからメッセージが破棄されるキューのタイプです。この動作により、last value キューは、同じプロパティーのメッセージは最後の値のみを保持します。

last value キューに対する単純なユースケースは、株式の株価しか監視するためのものです。特定の株式の最新値のみが関心があります。

以下を使用して、last value キューを設定できます。

上記の各メソッドでは、Core API を除いて、last value キー (最後の値のプロパティーとも呼ばれます) のカスタム値を指定するか、この値を未設定のままにすることができるので、代わりにキーがデフォルト値に設定されます。last value キーのデフォルト値は _AMQ_LVQ_NAME です。

Core API の場合、定数の last value キー Message.HDR_LAST_VALUE_NAME を使用してlast value キューを作成し、last value メッセージを識別します。

<address name="my.address">
<multicast>
<queue name="prices1" last-value-key="stock_ticker"/>
</multicast>
</address>

<address name="my.address">
<multicast>
<queue name="prices1" last-value="true"/>
</multicast>
</address>

Queue queue = session.createQueue("my.destination.name?last-value-key=stock_ticker");
Topic topic = session.createTopic("my.destination.name?last-value-key=stock_ticker");

Queue queue = session.createQueue("my.destination.name?last-value=true");
Topic topic = session.createTopic("my.destination.name?last-value=true");

void createQueue(
SimpleString address,
RoutingType routingType,
SimpleString queueName,
SimpleString filter,
Boolean durable,
Boolean autoCreated,
int maxConsumers,
Boolean purgeOnNoConsumers,
Boolean exclusive,
Boolean lastValue
)

<address-setting match="lastValueQueue">
   <default-last-value-key>stock_ticker</default-last-value-key>
</address-setting>

<address name="my.address">
<multicast>
<queue name="prices1" last-value-key="stock_ticker"/>
</multicast>
</address>

TextMessage message = session.createTextMessage("First message with last value property set");
message.setStringProperty("stock_ticker", "36.83");
producer.send(message);

TextMessage message = session.createTextMessage("Second message with last value property set");
message.setStringProperty("stock_ticker", "37.02");
producer.send(message);

TextMessage messageReceived = (TextMessage)messageConsumer.receive(5000);
System.out.format("Received message: %s\n", messageReceived.getText());

<address name="my.address">
   <multicast>
      <queue name="orders1" last-value-key="stock_ticker" non-destructive="true" />
   </multicast>
</address>

Queue queue = session.createQueue("my.destination.name?last-value-key=stock_ticker&non-destructive=true");
Topic topic = session.createTopic("my.destination.name?last-value-key=stock_ticker&non-destructive=true");

<address-setting match="lastValueQueue">
   <default-last-value-key>stock_ticker </default-last-value-key>
   <default-non-destructive>true</default-non-destructive>
</address-setting>

max-consumers 属性を使用して、特定のキューに接続するコンシューマーの数を制限します。max-consumers フラグを 1 に設定して、排他的コンシューマーを作成します。デフォルト値は -1 で、無制限のコンシューマーを設定します。

専用キューは、すべてのメッセージを一度に 1 つのコンシューマーだけにルーティングする特別なキューです。この設定は、すべてのメッセージを同じコンシューマーによって順次処理する必要がある場合に便利です。キューに複数のコンシューマーがある場合は、1 つのコンシューマーのみがメッセージを受信します。このコンシューマーが切断されると、別のコンシューマーが選択されます。

<configuration ...>
  <core ...>
    ...
    <address name="foo.bar">
      <multicast>
        <queue name="orders1" exclusive="true"/>
      </multicast>
    </address>
  </core>
</configuration>

Queue queue = session.createQueue("my.destination.name?exclusive=true");
Topic topic = session.createTopic("my.destination.name?exclusive=true");

<address-setting match="myQueue">
    <default-exclusive-queue>true</default-exclusive-queue>
</address-setting>

通常、AMQ Broker のキューは first-in、first-out(FIFO) セマンティクスを使用します。これは、ブローカーがキューの末尾にメッセージを追加し、それらをヘッドから削除することを意味します。リングキューは、指定された数のメッセージを保持する特別なタイプのキューです。ブローカーは、新規メッセージが到達し、キューがすでに指定された数のメッセージを保持した場合にキューの先頭にメッセージを削除することで、固定キューのサイズを維持します。

たとえば、サイズ 3 で設定されたリングキューと、メッセージ A、B、C、および D を順番に送信するプロデューサーについて考えてみます。メッセージ C がキューに到達すると、キュー内のメッセージ数が設定済みのリングサイズに達したことになります。この時点で、メッセージ A はキューのヘッドにあり、メッセージ C はテイールにあります。メッセージ D がキューに到達すると、ブローカーはキューの末尾にメッセージを追加します。固定キューサイズを維持するために、ブローカーはキュー (メッセージ A) の先頭にメッセージを削除します。メッセージ B がキューの先頭にあるようになりました。

通常、anycast と multicast の両方を使用しているアドレスでメッセージを受信した場合に、anycast のいずれかのキューがメッセージと、multicast のすべてのキューを受信します。ただし、クライアントは、アドレスに接続する際に特別な接頭辞を指定して、anycast と multicast キャストのどちらで接続するかを指定することができます。接頭辞は、acceptor の URL 内で anycastPrefix および multicastPrefix パラメーターを使用して指定されるカスタム値です。

<configuration ...>
  <core ...>
    ...
      <acceptors>
         <!-- Acceptor for every supported protocol -->
         <acceptor name="artemis">tcp://0.0.0.0:61616?protocols=AMQP;anycastPrefix=anycast://</acceptor>
      </acceptors>
    ...
  </core>
</configuration>

<configuration ...>
  <core ...>
    ...
      <acceptors>
         <!-- Acceptor for every supported protocol -->
         <acceptor name="artemis">tcp://0.0.0.0:61616?protocols=AMQP;multicastPrefix=multicast://</acceptor>
      </acceptors>
    ...
  </core>
</configuration>

アドレスを retroactive として設定すると、アドレスに送信されたメッセージを保持することができます。これには、アドレスにキューがバインドされていない場合も含まれます。キューが後に作成され、アドレスにバインドされると、ブローカーはメッセージをこれらのキューにアクティブに配信します。アドレスが retroactive として設定されていない場合や、キューがバインドされていない場合、ブローカーはそのアドレスに送信されたメッセージを破棄します。

Retactive アドレスを設定する場合、ブローカーは ring queue と呼ばれるタイプのキューの内部インスタンスを作成します。リングキューは、指定された数のメッセージを保持する特別なタイプのキューです。キューが指定されたサイズに達すると、キューに到達する次のメッセージがキューから最も古いメッセージを強制的に強制的に実行します。Retroactive アドレスを設定する場合は、内部リングキューのサイズを間接的に指定します。デフォルトでは、内部キューは multicast ルーティングタイプを使用します。

Retactive アドレスによって使用される内部リングキューは、管理 API 経由で公開されます。メトリクスを検査し、キューが空など、その他の一般的な管理操作を実行できます。リングキューは、アドレスの全体的なメモリー使用量にも貢献し、メッセージのページングなどの動作に影響を与えます。

以下の手順では、アドレスを Retroactive として設定する方法を説明します。

プロトコルマネージャーは、プロトコル固有の概念を AMQ Broker アドレスモデルの概念 (キューとルーティングの種類) にマップします。たとえば、クライアントがアドレス /house/room1/lights および /house/room2/lights を含む MQTT サブスクリプションパケットを送信すると、MQTT プロトコルマネージャーは、2 つのアドレスに multicast セマンティクスが必要であることを認識します。そのため、プロトコルマネージャーはまず、両方のアドレスでmulticast が有効になっていることを確認します。そうでない場合は、動的に作成を試みます。成功すれば、プロトコル・マネージャーは、クライアントが要求した各サブスクリプションのために特別なサブスクリプションキューを作成します。

各プロトコルの動作は若干異なります。以下の表は、様々なタイプの queue へのサブスクライブフレームが要求されたときに、典型的に起こることを説明しています。

デフォルトでは、AMQ Broker は、OpenWire クライアントがブローカーに接続されているときにアドレスおよびキューに関するアドバイザリーメッセージを作成します。アドバイザリーメッセージは、ブローカーによって作成された内部管理アドレスに送信されます。これらのアドレスは、ユーザーがデプロイされたアドレスおよびキューと同じ表示内の AMQ コンソールに表示されます。有用な情報が提供されますが、ブローカーによって多数の宛先を管理する場合に、アドバイザリーメッセージにより不要な結果が生じる可能性があります。たとえば、メッセージはメモリー使用量やステンケーション接続リソースを増やす可能性があります。また、アドバイザリーメッセージを送信するために作成されたすべてのアドレスを表示しようとすると、AMQ コンソールが明確になる可能性があります。このような状況を回避するには、supportAdvisory パラメーターと suppressInternalManagementObjects パラメーターを使用して、ブローカー側でのアドバイスメッセージの動作を管理します。

BROKER_INSTANCE_DIR/etc/broker.xml 設定ファイルを編集してこれらのパラメーターを使用し、URL を使用して OpenWire アクセプターのパラメーターを設定します。以下に例を示します。

<acceptor name="artemis">tcp://127.0.0.1:61616?protocols=CORE,AMQP,OPENWIRE;supportAdvisory=false;suppressInternalManagementObjects=false</acceptor>

AMQ Broker には、メッセージの配信方法と時期、試行回数、およびメッセージの有効期限を制御する設定可能なオプションがいくつかあります。これらの設定オプションはすべて、<address-setting> 設定要素内に存在します。ワイルドカード構文を使用して、AMQ Broker に単一の <address-setting> を複数の宛先に適用させることができます。

AMQ Broker ワイルドカード構文

AMQ Broker は、セキュリティー設定、アドレス設定、およびコンシューマーの作成時に、ワイルドカードを表すために特定の構文を使用します。

マッチングは文字による文字は実行されませんが、各区切り文字境界では文字ごとに一致します。たとえば、名前に my を使用してキューを照合しようとする address-setting は、myqueue という名前のキューとは一致しません。

キューに複数の address-setting がマッチする場合、ブローカーは、ベースラインとして最も合致の少ない設定を使用して設定をオーバーレイします。リテラル式はワイルドカードよりも限定的であり、* は # よりも限定的です。たとえば、my.queue と my.* の両方がキュー my.queue に一致します。この場合、ワイルドカード式はリテラルよりも具体的なため、ブローカーは最初に my.* にある設定を適用します。次に、ブローカーは my.queue address-setting の設定をオーバーレイし、my.* と共有される設定が上書きされます。以下の設定では、キュー my.queue の max-delivery-attempts が 3 に設定され、last-value-queue が false に設定されます。

<address-setting match="my.*">
    <max-delivery-attempts>3</max-delivery-attempts>
    <last-value-queue>true</last-value-queue>
</address-setting>
<address-setting match="my.queue">
    <last-value-queue>false</last-value-queue>
</address-setting>

以下の表の例は、ワイルドカードを使用してアドレスセットを照合する方法を示しています。

ワイルドカード構文の設定

設定を broker.xml に追加することで、ワイルドカードアドレスに使用される構文をカスタマイズできます。

<configuration>
  <core>
    ...
    <wildcard-addresses> 1
      <enabled>true</enabled> 2
      <delimiter>,</delimiter> 3
      <any-words>@</any-words> 4
      <single-word>$</single-word> 5
    </wildcard-addresses>
    ...
  </core>
</configuration>

アドレスおよびキューを自動的に作成し、使用されなくなったら削除するように AMQ Broker を設定できます。これにより、クライアントが接続する前に各アドレスを事前に設定する必要がなくなります。

キューとアドレスの自動作成と削除は、address-setting ごとに設定されます。設定は、address-setting に一致する任意のアドレスまたはキューに適用されます。ワイルドカード構文を使用してアドレスとキューを address-setting に一致させる方法は、アドレスの設定 を参照してください。

以下の表は、address -setting を設定してキューとアドレス を自動的に作成し、削除する際に利用できる設定要素の一覧です。

<configuration ...>
 <core ...>
  ...
  <address-settings>
    <address-setting match="activemq.#"> 1
      <auto-create-addresses>true</auto-create-addresses> 2
      <auto-delete-addresses>true</auto-delete-addresses> 3
      <auto-create-queues>true</auto-create-queues> 4
      <auto-delete-queues>true</auto-delete-queues> 5
      <default-address-routing-type>ANYCAST</default-address-routing-type> 6
    </address-setting>
  </address-settings>
  ...
 </core>
</configuration>

最後の値キュー以外のキューでは、非破壊的なコンシューマーしかない場合、ブローカーはキューからメッセージを削除できず、キューのサイズが徐々に増加します。キューサイズで制約のない増加を防ぐには、メッセージの有効期限が切れるタイミングを設定し、ブローカーが期限切れのメッセージを移動するアドレスを指定します。

クライアントにメッセージの配信に失敗した場合は、ブローカーがメッセージの配信を継続しようとしない場合があります。無限配信を試行するのを防ぐために、dead letter address と 1 つ以上の dead letter queuesを定義できます。指定の数の配信試行後、ブローカーは元のキューから未配信メッセージを削除し、そのメッセージを設定済みの dead letter アドレスに送信します。システム管理者は、デッド文字キューから未配信メッセージを後で消費してメッセージを検査できます。

指定のキューに dead letter アドレスを設定しない場合、ブローカーは指定された数の配信試行後にキューから未配信メッセージを完全に削除します。

dead letter キューから消費される配信されていないメッセージには、以下のプロパティーがあります。

ブローカーは、期限切れの AMQP メッセージまたは未配信の AMQP メッセージを、設定した期限切れまたは dead letter キューに移動する前に、アノテーションおよびプロパティーをメッセージに適用します。クライアントは、これらのプロパティーまたはアノテーションに基づいてフィルターを作成し、有効期限または dad letter キューから消費する特定のメッセージを選択できます。

以下の表は、ブローカーが期限切れの AMQP メッセージまたは配信されないアノテーションおよび内部プロパティーを表しています。

ログイン認証情報のないユーザー、または認証情報の検証に失敗したユーザーには、guest アカウントを使用してブローカーへの制限付きアクセスを許可できます。

コマンドラインの切り替えを使用して --allow-anonymous (--require-login の逆)、ゲストアクセスを有効にし、ブローカーインスタンスを作成できます。

ゲストアカウントは、login.config ファイルで設定されます。

activemq {
  org.apache.activemq.artemis.spi.core.security.jaas.GuestLoginModule required
      org.apache.activemq.jaas.guest.user="guest" 1
      org.apache.activemq.jaas.guest.role="guest"; 2
};

ゲストログインモジュールを使用すると、認証情報を持たないユーザー (設定方法によっては、認証情報が無効なユーザーも) がブローカーにアクセスできます。org.apache.activemq.artemis.spi.core.security.jaas.GuestLoginModule により実装されます。

ゲストログインモジュールは、プロパティーログインモジュールなどの別のログインモジュールと組み合わせて使用するのが一般的です。そのユースケースの詳細は、「複数のログインモジュールの使用」セクションを参照してください。

基本的なユーザー名とパスワードの検証が必要な場合は、Properties ログインモジュールを使用して定義します。このログインモジュールは、一連のローカルプロパティーファイルに対してユーザーの認証情報をチェックします。

ユーザーとそれに対応するパスワードは、BROKER_INSTANCE_DIR/etc/artemis-users.properties ファイルにリストされています。利用可能なロールとそれらのロールが割り当てられているユーザーは、BROKER_INSTANCE_DIR/etc/artemis-roles.properties ファイルで定義されています。

これらのファイルは両方とも、BROKER_INSTANCE_DIR/etc/login.config ファイルで参照されます。

JAAS の詳細は、Java ベンダーのドキュメントを参照してください。たとえば、Oracle には、login.config の設定に関するチュートリアルがあります。

activemq { 1
    org.apache.activemq.artemis.spi.core.security.jaas.PropertiesLoginModule required 2 3
        org.apache.activemq.jaas.properties.user="artemis-users.properties"; 4
        org.apache.activemq.jaas.properties.role="artemis-roles.properties";
};

以下は、前の例にリストされている各成功状態の説明です。

これらのフラグと認証プロセスの詳細は、Oracle のドキュメント を参照してください。

user1=secret 1
user2=swordfish 2
user3=myPassword 3

admin=user1,user2 1
developer=user3 2

<jaas-security domain="activemq"/>

権限は、broker.xml の <security-setting> 要素を介してアドレスに基づいてキューに対して定義されます。<security-setting> の複数のインスタンスを <security-settings> で定義できます。アドレスの完全一致を使用することができます。または、ワイルドカード文字 # と * を使ったワイルドカード一致を使用することも可能です。

アドレスに一致する一連のキューに、別の権限を指定できます。これらの権限は次のとおりです。

パーミッションごとに、そのパーミッションが付与されているロールのリストが指定されています。ユーザーにこれらのロールのいずれかが割り当てられている場合、そのアドレスのセットに対するそのアクセス許可が付与されます。

<security-settings>
    <security-setting match="queue1"> 1
        <permission type="send" roles="producer"/> 2
    </security-setting>
</security-settings>

<security-settings>
    <security-setting match="queue1"> 1
        <permission type="consume" roles="consumer"/> 2
    </security-setting>
</security-settings>

<security-settings>
    <security-setting match="#"> 1
        <permission type="createDurableQueue" roles="guest"/>
        <permission type="deleteDurableQueue" roles="guest"/>
        <permission type="createNonDurableQueue" roles="guest"/>
        <permission type="deleteNonDurableQueue" roles="guest"/>
        <permission type="createAddress" roles="guest"/>
        <permission type="deleteAddress" roles="guest"/>
        <permission type="send" roles="guest"/>
        <permission type="browse" roles="guest"/>
        <permission type="consume" roles="guest"/>
        <permission type="manage" roles="guest"/>
    </security-setting>
</security-settings>

<address name="ExempleQueue">
    <anycast>
       <queue name="ExampleQueue" user="admin" />
    </anycast>
</address>

ロールベースのアクセス制御 (RBAC) を使用して、MBean の属性とメソッドへのアクセスを制限します。RBAC を使用すると、管理者は、ロールに基づいて、Web コンソール、管理インターフェイス、コアメッセージなどのすべてのユーザーに適切なレベルのアクセスを許可できます。RBAC は、BROKER_INSTANCE_DIR/etc/management.xml 設定ファイルの authorization 要素を使用して設定されます。authorization 要素内で、Whitelist、default-access、および role-access サブ要素を設定できます。

<match domain="org.apache.activemq.artemis" key="subcomponent=queues">
   <access method="list*" roles="view,update,amq"/>
   <access method="get*" roles="view,update,amq"/>
   <access method="is*" roles="view,update,amq"/>
   <access method="set*" roles="update,amq"/>
   <access method="*" roles="amq"/>
</match>

<match domain="org.apache.activemq.artemis" key="queue=exampleQueue">
   <access method="list*" roles="view,update,amq"/>
   <access method="get*" roles="view,update,amq"/>
   <access method="is*" roles="view,update,amq"/>
   <access method="set*" roles="update,amq"/>
   <access method="*" roles="amq"/>
</match>

<match domain="org.apache.activemq.artemis" key="queue=example*">
   <access method="list*" roles="view,update,amq"/>
   <access method="get*" roles="view,update,amq"/>
   <access method="is*" roles="view,update,amq"/>
   <access method="set*" roles="update,amq"/>
   <access method="*" roles="amq"/>
</match>

<match domain="org.apache.activemq.artemis" key="queue=example*">

<match domain="org.apache.activemq.artemis" key="queue=example.sub*">

AMQ Broker 7.1.0 以降では、デフォルトでローカルホストからのみ AMQ 管理コンソールにアクセスできます。リモートアクセスを有効にするには、BROKER_INSTANCE_DIR/etc/jolokia-access.xml の設定を変更する必要があります。詳細は、AMQ 管理コンソールと AMQ ブローカ接続の保護 を参照してください。

トランスポート層セキュリティー (TLS) には、次の 2 つの基本的な使用例があります。

<acceptor name="artemis">tcp://0.0.0.0:61616</acceptor>

<acceptor name="artemis">tcp://0.0.0.0:61616?sslEnabled=true;keyStorePath=../etc/broker.keystore;keyStorePassword=1234!</acceptor>

<acceptor name="artemis">tcp://0.0.0.0:61616?sslEnabled=true;keyStorePath=../etc/broker.keystore;keyStorePassword=1234!;needClientAuth=true</acceptor>

<acceptor name="artemis">tcp://0.0.0.0:61616?sslEnabled=true;keyStorePath=../etc/broker.keystore;keyStorePassword=1234!;needClientAuth=true;trustStorePath=../etc/client.truststore;trustStorePassword=5678!</acceptor>

activemq {
    org.apache.activemq.artemis.spi.core.security.jaas.TextFileCertificateLoginModule 1
        debug=true 2
        org.apache.activemq.jaas.textfiledn.user="artemis-users.properties" 3
        org.apache.activemq.jaas.textfiledn.role="artemis-roles.properties"; 4
};

system=CN=system,O=Progress,C=US 1
user=CN=humble user,O=Progress,C=US
guest=CN=anon,O=Progress,C=DE

admins=system
users=system,user 1
guests=guest

activemq {
  org.apache.activemq.artemis.spi.core.security.jaas.PropertiesLoginModule sufficient 1
      debug=true
      org.apache.activemq.jaas.properties.user="artemis-users.properties"
      org.apache.activemq.jaas.properties.role="artemis-roles.properties";

  org.apache.activemq.artemis.spi.core.security.jaas.GuestLoginModule sufficient 2
      debug=true
      org.apache.activemq.jaas.guest.user="guest"
      org.apache.activemq.jaas.guest.role="restricted";
};

activemq {
    org.apache.activemq.artemis.spi.core.security.jaas.GuestLoginModule sufficient 1
        debug=true
       credentialsInvalidate=true 2
       org.apache.activemq.jaas.guest.user="guest"
       org.apache.activemq.jaas.guest.role="guests";

    org.apache.activemq.artemis.spi.core.security.jaas.PropertiesLoginModule requisite 3
        debug=true
        org.apache.activemq.jaas.properties.user="artemis-users.properties"
        org.apache.activemq.jaas.properties.role="artemis-roles.properties";
};

<security-setting match="globalqueues.europe.#"> 1
   <permission type="createDurableQueue" roles="admin"/> 2
   <permission type="deleteDurableQueue" roles="admin"/> 3
   <permission type="createNonDurableQueue" roles="admin, guest, europe-users"/> 4
   <permission type="deleteNonDurableQueue" roles="admin, guest, europe-users"/> 5
   <permission type="send" roles="admin, europe-users"/> 6
   <permission type="consume" roles="admin, europe-users"/> 7
</security-setting>

<security-setting match="globalqueues.europe.orders.#">
   <permission type="send" roles="europe-users"/>
   <permission type="consume" roles="europe-users"/>
</security-setting>

<resource-limit-settings>
   <resource-limit-setting match="myUser">
      <max-connections>5</max-connections>
      <max-queues>3</max-queues>
   </resource-limit-setting>
</resource-limit-settings>

AMQP プロトコルでメッセージを送受信する場合、クライアントは Simple Authentication and Security Layer (SASL) フレームワークの GSSAPI メカニズムを使用して、AMQ Broker が認証する Kerberos セキュリティー認証情報を送信できます。Kerberos 認証情報は、認証されたユーザーを LDAP ディレクトリーまたはテキストベースのプロパティーファイルで設定された割り当てられたロールにマッピングすることで、承認にも使用できます。

Transport Layer Sockets (TLS) と SASL を併用して、メッセージングアプリケーションのセキュリティーを確保できます。SASL はユーザー認証を行い、TLS はデータの整合性を確保します。

AMQ Broker が Kerberos 認証情報を認証および承認する前に、Kerberos インフラストラクチャーをデプロイし、設定する必要があります。Kerberos のデプロイメントに関する詳細は、オペレーティングシステムのドキュメントを参照してください。たとえば、オペレーティングシステムが RHEL 7 の場合は、システムレベル認証ガイドの Kerberos の使用 の章を参照してください。Windows 用の Kerberos 認証の概要 も利用できます。

<acceptor name="amqp">
tcp://0.0.0.0:5672?protocols=AMQP;saslMechanisms=GSSAPI,PLAIN;saslLoginConfigScope=alternative-sasl-gssapi`
</acceptor>

デフォルトでは、AMQ Broker はすべてのパスワードをプレーンテキストとして設定ファイルに保存します。承認されていないアクセスを防ぐために、適切なパーミッションですべての設定ファイルのセキュリティーを保護するようにしてください。また、プレーンテキストのパスワードを暗号化するか、マスクして、不要なビューアーが読み込まれないようにすることもできます。

マスクされたパスワードは、プレーンテキストのパスワードを暗号化したものです。暗号化バージョンは、AMQ Broker によって提供される mask コマンドラインユーティリティーによって生成されます。mask ユーティリティーの詳細は、コマンドラインのヘルプドキュメントを参照してください。

$ BROKER_INSTANCE_DIR/bin/artemis help mask

パスワードをマスクするには、プレーンテキストの値を暗号化された値に置き換えます。マスクされたパスワードは、実際の値が必要になったときに復号化されるように、識別子 ENC() でラップする必要があります。

以下の例では、BROKER_INSTANCE_DIR/etc/bootstrap.xml 設定ファイルには、keyStorePassword および trustStorePassword 属性のマスクされたパスワードが含まれます。

<web bind="https://localhost:8443" path="web"
     keyStorePassword="ENC(-342e71445830a32f95220e791dd51e82)"
     trustStorePassword="ENC(32f94e9a68c45d89d962ee7dc68cb9d1)">
    <app url="activemq-branding" war="activemq-branding.war"/>
</web>

以下の設定ファイルでは、マスクされたパスワードのみを使用できます。

設定ファイルは BROKER_INSTANCE_DIR/etc にあります。

設定ファイルには、マスクされたパスワードが含まれるようになりました。パスワードは ENC() 識別子でラップされるため、AMQ Broker では使用前にこれを復号化します。

メッセージの発信元の追跡およびロギングを有効にするには (セキュリティー監査の目的など)、_AMQ_VALIDATED_USER メッセージキーを使用できます。

broker.xml 設定ファイルで populate-validated-user オプションが true に設定されている場合には、ブローカーは _AMQ_VALIDATED_USER キーを使用して検証済みユーザーの名前をメッセージに追加します。JMS および STOMP クライアントの場合には、このメッセージキーは JMSXUserID キーにマッピングされます。

自分の SSL 証明書を基に認証されるユーザーの場合、ブローカーによって設定される検証されたユーザー名は、証明書の識別名 (DN) マップの名前です。

broker.xml 設定ファイルで security-enabled が false で、populate-validated-user が true の場合に、ブローカーはクライアントが指定するユーザー名 (ある場合) を指定します。populate-validated-user オプションのデフォルトは false です。

メッセージの送信時に、クライアントによりユーザー名 (つまり、JMSXUser ID キー) がまだ入力されていないメッセージを拒否するようにブローカーを設定できます。ブローカーはこれらのクライアントによって送信されたメッセージについて検証されたユーザー名自体を設定できないため、このオプションが AMQP クライアントに役立ちます。

クライアントによって JMSXUserID が設定されていないメッセージを拒否するようにブローカーを設定するには、以下の設定を broker.xml 設定ファイルに追加します。

<reject-empty-validated-user>true</reject-empty-validated-user>

デフォルトでは、reject-empty-validated-user は false に設定されます。

セキュリティーはデフォルトで有効になっています。ブローカーのセキュリティーは、broker.xml 設定ファイルの <core> 要素に <security-enabled> パラメーターを設定して有効または無効にすることができます。

ブローカーは、セキュリティーマネージャー と呼ばれるコンポーネントを使用して認証および承認を処理します。デフォルトでは、AMQ Broker は org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager セキュリティーマネージャーを使用します。このデフォルトのセキュリティーマネージャーは、JAAS および Red Hat JBoss Enterprise Application Platform(JBoss EAP) のセキュリティーとの統合を提供します。

ただし、システム管理者は、ブローカーセキュリティーの実装をより詳細に制御する必要がある場合があります。このような場合には、ブローカー設定でカスタムセキュリティーマネージャーを指定できます。カスタムセキュリティーマネージャーは、org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager5 インターフェイスを実装するユーザー定義のクラスです。

ブローカージャーナルは、ディスク上の append-only のファイルセットです。各ファイルは固定サイズに事前作成され、最初にパディングが入力されます。メッセージング操作がブローカーで実行されると、レコードがジャーナルの最後に追加されます。レコードを追加すると、ブローカーはディスクヘッドの移動およびランダムアクセス操作を最小限に抑えることができます。これは通常、ディスク上で最も遅い操作です。1 つのジャーナルファイルがいっぱいになると、ブローカーは新しいジャーナルファイルを使用します。

ジャーナルファイルサイズは設定可能です。各ファイルによって使用されるディスクリプラーの数を最小限に抑えることができます。最新のディスクトポロジーは複雑で、ブローカーはファイルのマッピング先の cylinder を制御することはできません。したがって、ジャーナルファイルのサイジングは、正確な情報ではありません。

その他の永続性関連の機能には、以下が含まれます。

ジャーナルの大半は Java で書かれています。ただし、実際のファイルシステムの操作は抽象化されるため、プラグ可能な実装を各種使用できます。AMQ Broker には、2 つの実装が同梱されています。

yum install libaio

永続性の設定は、BROKER_INSTANCE_DIR/etc/broker.xml ファイルで維持されます。ブローカーのデフォルト設定はジャーナルベースの永続化を使用し、以下の要素が含まれます。

<configuration>
  <core>
    ...
    <persistence-enabled>true</persistence-enabled>
    <journal-type>ASYNCIO</journal-type>
    <bindings-directory>./data/bindings</bindings-directory>
    <journal-directory>./data/journal</journal-directory>
    <journal-datasync>true</journal-datasync>
    <journal-min-files>2</journal-min-files>
    <journal-pool-files>-1</journal-pool-files>
    ...
  </core>
</configuration>

JDBC 永続ストアは JDBC 接続を使用してメッセージとバインディングデータをデータベーステーブルに保存します。テーブルのデータは AMQ Broker ジャーナルエンコーディングを使用してエンコードされます。サポートされるデータベースの詳細は、Red Hat カスタマーポータルの Red Hat AMQ 7 Supported Configurations を参照してください。

メッセージングシステムに、ゼロ永続化が必要になる場合があります。ゼロ永続化を実行するためのブローカーを設定することは簡単です。BROKER_INSTANCE_DIR/etc/broker.xml の persistence-enabled パラメーターを false に設定します。

このパラメーターを false に設定すると、ゼロ 永続化が発生することに注意してください。つまり、バインディングデータ、メッセージデータ、大きなメッセージデータ、重複した ID キャッシュまたはページングデータが永続化されません。

メッセージはファイルシステムのアドレスごとに保存されます。各アドレスには、メッセージが複数のファイル (ページファイル) に保存される個別のフォルダーがあります。各ファイルには、最大に設定されたサイズ (page-size-bytes) までのメッセージが含まれます。システムは必要に応じてファイルを移動し、すべてのメッセージがそのポイントに確認応答されるとすぐにページファイルが削除されます。

ブラウザーは page-cursor システムを通じて読み込まれます。

セレクターを持つコンシューマーもページファイルを検索し、基準に見合わないメッセージを無視します。

ページングディレクトリーの場所を設定するには、以下の例のように paging-directory 設定要素をブローカーのメイン設定ファイル BROKER_INSTANCE_DIR/etc/broker.xml に追加します。

<configuration ...>
  ...
  <core ...>
    <paging-directory>/somewhere/paging-directory</paging-directory>
    ...
  </core>
</configuration>

AMQ Broker は、設定された場所でページされる各アドレスに 1 つのディレクトリーを作成します。

ページングの設定は、以下の例のように特定の address-settings に要素を追加して、アドレスレベルで行われます。

<address-settings>
   <address-setting match="jms.paged.queue">
      <max-size-bytes>104857600</max-size-bytes>
      <page-size-bytes>10485760</page-size-bytes>
      <address-full-policy>PAGE</address-full-policy>
   </address-setting>
</address-settings>

上記の例では、アドレス jms.paged.queue に送信されたメッセージがメモリーの 104857600 バイトを超えると、ブローカーはページングを開始します。

これは、アドレス設定で利用可能なパラメーターの一覧です。

アドレスごとにメモリー制限を設定することは、ブローカーが異なる使用パターンを持つアドレスを多数管理する場合など、実用的ではないことがあります。このような状況では、global-max-size パラメーターを使用して、ブローカーが受信メッセージに関連付けられたアドレスに設定されたページモードに入る前に、ブローカーが使用可能なメモリー量にグローバル制限を設定します。

global-max-size のデフォルト値は、Java 仮想マシン (JVM) で利用可能な最大メモリーの半分です。broker.xml 設定ファイルで設定すると、このパラメーターに独自の値を指定できます。global-max-size の値はバイト単位ですが、便宜上バイト表記 ("K"、Mb、"GB" など) を使用できます。

以下の手順では、broker.xml 設定ファイルで global-max-size パラメーターを設定する方法を説明します。

global-max-size パラメーターの設定

ブローカーが受信メッセージをブロックする前に、ブローカーが使用する物理ディスクの量を制限することができます。max-disk-usage を broker.xml 設定ファイルに追加し、ページングメッセージ時にブローカーが使用できるディスク領域の割合の値を指定します。max-disk-usage のデフォルト値は 90 です。これは、制限がディスク領域の 90 パーセントに設定されることを意味します。

max-disk-usage の設定

最大サイズに達したときにメッセージをページングするのではなく、アドレスが満杯になったときにメッセージをドロップするように設定できます。

これを行うには、アドレス設定で address-full-policy を DROP に設定します。

最大サイズに達したときにメッセージをページングせずに、アドレスが満杯になったときにプロデューサーがさらにメッセージを送信しないように設定することもできます。そのため、サーバーでメモリーが使い切られるのを防ぐことができます。

サーバー上でメモリーが解放されると、プロデューサーは自動的にブロック解除され、送信を継続できます。

これを行うには、アドレス設定で address-full-policy を BLOCK に設定します。

デフォルトの設定では、10 MiB のデータがアドレスに含まれていると、すべてのアドレスがプロデューサーをブロックするように設定されています。

マルチキャストキューがバインドされているアドレス (Topic の JMS サブスクリプションなど) にメッセージがルーティングされた場合は、メッセージのコピーはメモリー内に 1 つだけ存在します。各キューは参照のみを処理します。このため、メモリーは、メッセージを参照するすべてのキューがメッセージを配信した後のみ解放されます。

1 つのレイジー (遅延) サブスクリプションがある場合、ページングシステム上の追加のストレージを介して送信されたメッセージがすべてのキューにあるため、アドレス全体が IO パフォーマンスの影響を受けます。

以下に例を示します。

この例では、他のすべての 9 キューはページシステムからメッセージを消費します。これにより、これが望ましくない状態である場合にパフォーマンスの問題が発生する可能性があります。

以下の手順では、ブローカーが大きなメッセージファイルを保存するディスクまたはデータベーステーブルにディレクトリーを指定する方法を説明します。

以下の手順は、指定したサイズよりも大きい AMQP メッセージを大規模メッセージとして処理するように AMQP アクセプターを設定する方法を説明します。

以下の手順は、指定したサイズよりも大きい STOMP メッセージをサイズの大きいメッセージとして処理するように STOMP アクセプターを設定する方法を説明します。

前の例では、ブローカーはポート 61613 で STOMP メッセージを受け入れるように設定されています。stompMinLargeMessageSize の値に基づいて、アクセプターが 204800 バイトよりも大きい STOMP メッセージ (200 キロバイト以上) を受信する場合、ブローカーはメッセージを大きなメッセージとして格納します。このプロパティーの値を明示的に指定しない場合、 ブローカーは 102400 (100 キロバイト) のデフォルト値を使用します。

大きなメッセージを使用するクライアントを作成している Java 開発者には、2 つのオプションがあります。

1 つのオプションとして、InputStream および OutputStream のインスタンスを使用することができます。たとえば、FileInputStream を使用して、物理ディスク上の大きなファイルから作成されたメッセージを送信します。その後、受信者が FileOutputStream を使用して、メッセージをローカルファイルシステムの場所にストリーミングできます。

別のオプションとして、JMS BytesMessage または StreamMessage を直接ストリーミングする方法もあります。以下に例を示します。

BytesMessage rm = (BytesMessage)cons.receive(10000);
byte data[] = new byte[1024];
for (int i = 0; i < rm.getBodyLength(); i += 1024)
{
   int numberOfBytes = rm.readBytes(data);
   // Do whatever you want with the data
}

クライアントとサーバー間のネットワーク接続に失敗してオンラインに戻る可能性があるため、AMQ Broker は非アクティブなサーバー側のリソースをクリーンアップします。この待機時間は Time-to-live (TTL) と呼ばれます。ネットワークベースの接続のデフォルトの TTL は 60000 ミリ秒 (1 分 ) です。in-VM 接続のデフォルトの TTL は -1 です。つまり、ブローカーはブローカー側で接続をタイムアウトしません。

ブローカーでの Time-To-Live の設定

クライアントが独自の接続 TTL を指定しないようにするには、ブローカー側でグローバル値を設定します。これは、ブローカー設定で connection-ttl-override 要素を指定して実行できます。

connection-ttl-check-interval 要素によって決定されるように、TTL 違反の接続をチェックするロジックはブローカーで定期的に実行されます。

クライアント上での Time-To-Live の設定

デフォルトでは、クライアントは独自の接続に TTL を設定できます。以下の例は、Core JMS クライアントを使用して Time-To-Live を設定する方法を示しています。

サーバー側で受信されたパケットの多くは、リモート スレッドで実行されます。これらのパケットは短時間実行される操作を表し、パフォーマンス上の理由から常に リモーティング スレッドで実行されます。ただし、一部のパケットタイプは リモーティング スレッドではなくスレッドプールを使用して実行され、ネットワークのレイテンシーが少し追加されます。

スレッドプールを使用するパケットタイプは、以下に示す Java クラス内に実装されます。クラスはすべて org.apache.actiinvemq.artemis.core.protocol.core.impl.wireformat パッケージにあります。

クライアントアプリケーションは、デッド接続が発生しないように、終了する前に制御された方法でリソースを閉じる必要があります。Java では、finally ブロック内で接続を閉じることが推奨されます。

Connection jmsConnection = null;
try {
   ConnectionFactory jmsConnectionFactory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(...);
   jmsConnection = jmsConnectionFactory.createConnection();
   ...use the connection...
}
finally {
   if (jmsConnection != null) {
      jmsConnection.close();
   }
}

コンシューマーフロー制御は、クライアントがブローカーからメッセージを消費する際に、ブローカーとクライアント間のデータフローを制御します。AMQ Broker クライアントは、デフォルトでメッセージをバッファーしてからコンシューマーに配信します。バッファーがない場合、クライアントはまず、消費する前にブローカーから各メッセージを要求する必要があります。このタイプのラウンドトリップ通信はコストがかかります。メモリー不足の問題によりコンシューマーがメッセージをすばやく処理できず、バッファーが受信メッセージでオーバーフローを開始するため、クライアント側のデータのフローを制限することが重要になります。

コンシューマーウィンドウベースのフロー制御と同様に、AMQ Broker はプロデューサーからブローカーに送信されるデータ量をブローカーに制限し、ブローカーが大量のデータで過負荷にならないようにすることができます。プロデューサーの場合、ウィンドウサイズは 1 度にインフライトにできるバイト数を決定します。

<configuration>
  <core>
    ...
    <address-settings>
       <address-setting match="my.blocking.queue"> 1
          <max-size-bytes>300000</max-size-bytes>  2
          <address-full-policy>BLOCK</address-full-policy> 3
       </address-setting>
    </address-settings>
  </core>
</configuration>

以下の例は、Core JMS クライアントを使用してメッセージのグループ化を使用する方法を示しています。

グループ ID を独自に指定する代わりに、ID を自動的に生成することができます。この方法でグループ化されたメッセージは、1 つのコンシューマーによって順次処理されます。

複製メッセージ検出を有効にするには、メッセージプロパティー _AMQ_DUPL_ID に一意の値を指定します。ブローカーがメッセージを受信すると、_AMQ_DUPL_ID の値があるかどうかを確認します。存在する場合、ブローカーはメモリーキャッシュを確認し、その値を持つメッセージをすでに受信したかどうかを確認します。同じ値を持つメッセージが見つかると、受信メッセージは無視されます。

ブローカーは、_AMQ_DUPL_ID プロパティーの受信値のキャッシュを維持します。各アドレスには個別のキャッシュがあります。キャッシュは円形で固定されます。新しいエントリーは、キャッシュ領域の要求に合わせて最も古いエントリーを置き換えます。

<configuration>
  <core>
    ...
    <id-cache-size>5000</id-cache-size> 1
    <persist-id-cache>false</persist-id-cache> 2
  </core>
</configuration>

重複検出を使用してブローカー間でメッセージを移動すると、XA トランザクションを使用してメッセージを消費するのと同じ 1 度と 1 回のみ配信を保証できますが、オーバーヘッドが少なく、XA を使用するよりも設定がはるかに簡単になります。

トランザクションでメッセージを送信する場合は、トランザクション内のすべてのメッセージに _AMQ_DUPL_ID を設定する必要はなく、そのうちの 1 つにのみ設定する必要はありません。ブローカーがトランザクションのメッセージに対して重複メッセージを検出すると、トランザクション全体を無視します。

クラスター接続を設定して、クラスター全体に移動する各メッセージの重複 ID を挿入できます。

独自の受信インターセプターおよび発信インターセプターを作成できます。すべてのインターセプターはプロトコル固有で、サーバーに出入りするパケットに対して呼び出されます。これにより、監査パケットなどのビジネス要件を満たすインターセプターを作成できます。インターセプターは、傍受するパケットを変更できます。これにより、危険が伴います。そのため、注意して使用してください。

インターセプターとその依存関係は、ブローカーの Java クラスに配置する必要があります。BROKER_INSTANCE_DIR/libディレクトリーは、デフォルトでクラスパスに含まれているため、使用することができます。

インターセプターを作成したら、それを使用するようにブローカーを設定する必要があります。

クライアントはインターセプターを使用して、クライアントからサーバーに送信したパケットを、またはサーバーがクライアントへインターセプトできます。ブローカーサイドインターセプターの場合は、false を返す場合、他のインターセプターは呼び出されず、クライアントはさらにパケットを処理しません。このプロセスは、blocking 方式で発信パケットが送信される場合を除き、クライアントに透過的に行われます。このような場合、ブロック送信は信頼性を提供するため、ActiveMQException が呼び出し元に出力されます。出力されたActiveMQException には、false を返したインターセプターの名前が含まれています。

サーバーと同様に、クライアントインターセプタークラスとその依存関係を適切にインスタンス化および呼び出すには、クライアントの Java クラスパスに追加する必要があります。

迂回により、クライアントアプリケーションロジックを変更せずに、あるアドレスにルーティングされたメッセージを他のアドレスに透過的に迂回できます。ブローカーサーバーの迂回のセットをメッセージのルーティングテーブルの種類と見なすことができます。

迂回は 排他的 にすることができます。つまり、メッセージは元のアドレスに入りずに指定されたフォワードアドレスに迂回されます。

迂回は 排他的 ではない場合もあります。つまり、ブローカーはメッセージのコピーを指定された転送アドレスに送信する一方で、メッセージを元のアドレスに引き続き送信します。そのため、メッセージフローを分割するのに排他的でない迂回を使用できます。たとえば、注文キューに送信された全順序を個別に監視する場合は、メッセージフローを分割できます。

アドレスに排他的な迂回と非排他的な迂回の両方が設定されている場合、ブローカーは特別な迂回を最初に処理します。特定のメッセージがすでに特別な迂回によって迂回されている場合、ブローカーはそのメッセージの特別でない迂回を処理しません。この場合、メッセージは元のアドレスには決して送信されません。

ブローカーがメッセージを迂回すると、ブローカーは新しいメッセージ ID を割り当て、メッセージアドレスを新しい転送アドレスに設定します。元のメッセージ ID およびアドレスの値は、_AMQ_ORIG_ADDRESS ( 文字列タイプ ) および _AMQ_ORIG_MESSAGE_ID (long タイプ ) メッセージプロパティーから取得できます。Core API を使用している場合は、Message.HDR_ORIGINAL_ADDRESS および Message.HDR_ORIG_MESSAGE_ID プロパティーを使用します。

ブローカーインスタンスで迂回を設定するには、broker.xml 設定ファイルの core 要素に divert 要素を追加します。

<core>
...
   <divert name= >
        <address> </address>
        <forwarding-address> </forwarding-address>
        <filter string= >
        <routing-type> </routing-type>
        <exclusive> </exclusive>
   </divert>
...
</core>

ルーティングタイプの制御は、メッセージにルーティングタイプがすでに設定されているものの、別のルーティングタイプを使用するアドレスにメッセージを迂回する必要がある場合に役立ちます。たとえば、ブローカーは、迂回の routing-type パラメーターを MULTICAST に設定しない限り、anycast ルーティングタイプのメッセージを マルチキャスト を使用するキューにルーティングできません。迂回の routing-type パラメーターの有効な値は ANYCAST、MULTICAST、PASS、および STRIP です。デフォルト値は STRIP です。

以下のサブセクションは、排他的な迂回および排他的でない迂回の設定例を示しています。

<divert name="prices-divert">
   <address>priceUpdates</address>
   <forwarding-address>priceForwarding</forwarding-address>
   <filter string="office='New York'"/>
   <exclusive>true</exclusive>
</divert>

<divert name="order-divert">
   <address>orders</address>
   <forwarding-address>spyTopic</forwarding-address>
   <exclusive>false</exclusive>
</divert>

BROKER_INSTANCE_DIR/etc/broker.xml で設定するキューにフィルターを追加できます。フィルター式に一致するメッセージのみがキューに格納されます。

JMS 仕様は、セレクターで使用されると String プロパティーを数値型に変換してはならないことを示しています。たとえば、メッセージの age プロパティーが String 値 21 に設定されていると、セレクターの age > 18 は一致できません。この制限により、STOMP クライアントは String プロパティーでメッセージを送信できるため制限されます。

文字列を数値に変換するフィルターの設定

String プロパティーを数値型に変換するには、接頭辞 convert_string_expressions: を filter の値に追加します。

ブローカーは、期限切れの AMQP メッセージまたは未配信の AMQP メッセージを、設定した期限切れまたは dead letter キューに移動する前に、アノテーションおよびプロパティーをメッセージに適用します。クライアントはこれらのプロパティーまたはアノテーションに基づいてフィルターを作成し、期限切れまたは dead letter キューから消費する特定のメッセージを選択できます。

以下は、メッセージプロパティーとアノテーションに基づくフィルターの例です。このアプローチではブローカーによる処理が少なくなるため、プロパティーに基づくフィルターリングは可能な場合に推奨される方法です。

ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672");
Connection connection = factory.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
connection.start();
javax.jms.Queue queue = session.createQueue("my_DLQ");
MessageConsumer consumer = session.createConsumer(queue, "_AMQ_ORIG_ADDRESS='original_address_name'");
Message message = consumer.receive();

ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672");
Connection connection = factory.createConnection();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
connection.start();
javax.jms.Queue queue = session.createQueue("my_DLQ");
MessageConsumer consumer = session.createConsumer(queue, "\"m.x-opt-ORIG-ADDRESS\"='original_address_name'");
Message message = consumer.receive();

ブローカークラスターを作成する前に、いくつかの重要なクラスターリングの概念を理解する必要があります。

17.1.1. ブローカークラスターがメッセージ負荷のバランスを取る方法

ブローカーがクラスターを形成するとき、AMQ Broker はブローカー間でメッセージの負荷を自動的に分散します。これにより、クラスターが高メッセージスループットを維持できるようになります。

4 つのブローカーの対称クラスターについて考えてみましょう。各ブローカーは OrderQueue という名前のキューで設定されます。OrderProducer クライアントは Broker1 に接続し、メッセージを OrderQueue に送信します。Broker1 は、ラウンドロビン方式でメッセージを他のブローカーに転送します。各ブローカーに接続されている OrderConsumer クライアントはメッセージを消費します。正確な順序は、ブローカーが起動した順序によって異なります。

図17.1 メッセージの負荷分散

メッセージの負荷分散がない場合、Broker1 に送信されたメッセージは Broker1 に残り、OrderConsumer1 のみがそれらを消費できます。

AMQ Broker はデフォルトでメッセージを自動的に負荷分散しますが、一致するコンシューマーを持つブローカーへのメッセージのみを負荷分散するようにクラスターを設定できます。また、コンシューマーのないキューからコンシューマーを持つキューからメッセージを自動的に再分配するようにメッセージ再分配を設定することもできます。

関連情報

• メッセージの負荷分散ポリシーは、各ブローカーのクラスター接続の message-load-balancing プロパティーで設定されます。詳細は、付録C クラスター接続設定要素 を参照してください。
• メッセージ再分配に関する詳細は、「メッセージ再分配の設定」 を参照してください。

17.1.4. 一般的なブローカークラスタートポロジー

ブローカーに接続し、対称 または チェーン クラスタートポロジーのいずれかを形成できます。実装するトポロジーは、環境およびメッセージングの要件によって異なります。

対称クラスター

対称クラスターでは、すべてのブローカーが他のすべてのブローカーに接続されます。つまり、すべてのブローカーは他のすべてのブローカーから複数のホップを削除できません。

図17.2 対称クラスター

対称クラスターの各ブローカーは、クラスター内の他のすべてのブローカーに存在するすべてのキューと、これらのキューでリッスンしているコンシューマーを認識します。そのため、対称クラスターは、メッセージをチェーンクラスターよりも最適に負荷分散および再分散できます。

対称クラスターはチェーンクラスターよりも設定が簡単ですが、ネットワーク制限が原因でブローカーが直接接続されないようにする環境では使用が困難になる可能性があります。

チェーンクラスター

チェーンクラスターでは、クラスターの各ブローカーはクラスター内のすべてのブローカーに直接接続されません。代わりに、ブローカーはチェーンの最後ごとにブローカーと、チェーンの前のブローカーおよび次のブローカーに接続するすべてのブローカーでチェーンを形成します。

図17.3 チェーンクラスター

チェーンクラスターは対称クラスターよりも設定が難しくなりますが、ブローカーが別個のネットワーク上にあり、直接接続できない場合に役立ちます。チェーンクラスターを使用すると、中間ブローカーは 2 つのブローカーを間接的に接続し、2 つのブローカーが直接接続されていない場合でも、メッセージ間のフローが可能になります。

クラスターに参加する各ブローカーにクラスター接続を設定して、ブローカークラスターを作成します。クラスター接続は、ブローカーが他のブローカーに接続する方法を定義します。

静的検出または動的検出を使用するブローカークラスターを作成できます ( UDP マルチキャストまたは JGroups のいずれか )。

ブローカークラスターの作成後に、高可用性 (HA) を実装してその信頼性を強化できます。HA では、1 つ以上のブローカーがオフラインになっても、ブローカークラスターは機能し続けます。

HA の実装には、複数のステップが関係します。

17.3.5. Configuring high availability with colocated backups

ライブバックアップグループを設定するのではなく、バックアップブローカーを別のライブブローカーと同じ JVM にコロケートできます。この設定では、各ライブブローカーは別のライブブローカーを要求し、その JVM でバックアップブローカーを作成し、開始します。

図17.4 コロケートされたライブブローカーおよびバックアップブローカー

コロケーションは、共有ストアまたはレプリケーションのいずれかを高可用性 (HA) ポリシーとして使用できます。新しいバックアップブローカーは、これを作成するライブブローカーからその設定を継承します。バックアップの名前は、colocated_backup_n に設定されます。n は、ライブブローカーが作成したバックアップの数です。

さらに、バックアップブローカーは、コネクターと、これを作成するライブブローカーからアクセプターの設定を継承します。デフォルトでは、ポートオフセット 100 がそれぞれに適用されます。たとえば、ライブブローカーにポート 61616 のアクセプターがある場合、作成される最初のバックアップブローカーはポート 61716 を使用し、2 番目のバックアップは 61816 を使用します。

ジャーナル、大きなメッセージ、ページングのディレクトリーは、選択した HA ポリシーに従って設定されます。共有ストアを選択する場合、要求ブローカーはターゲットブローカーに対し、使用するディレクトリーについて通知します。レプリケーションが選択されている場合、ディレクトリーは作成ブローカーから継承され、新しいバックアップの名前が追加されます。

この手順では、共有のストア HA を使用するようにクラスター内の各ブローカーを設定し、バックアップを作成してクラスター内の別のブローカーと共存させるよう要求します。

手順

1. 最初のブローカーの <broker-instance-dir>/etc/broker.xml 設定ファイルを開きます。

2. HA ポリシーとコロケーションを使用するようにブローカーを設定します。

   この例では、ブローカーは共有ストア HA およびコロケーションで設定されます。

   <configuration>
       <core>
           ...
           <ha-policy>
               <shared-store>
                   <colocated>
                       <request-backup>true</request-backup>
                       <max-backups>1</max-backups>
                       <backup-request-retries>-1</backup-request-retries>
                       <backup-request-retry-interval>5000</backup-request-retry-interval/>
                       <backup-port-offset>150</backup-port-offset>
                       <excludes>
                           <connector-ref>remote-connector</connector-ref>
                       </excludes>
                       <master>
                           <failover-on-shutdown>true</failover-on-shutdown>
                       </master>
                       <slave>
                           <failover-on-shutdown>true</failover-on-shutdown>
                           <allow-failback>true</allow-failback>
                           <restart-backup>true</restart-backup>
                       </slave>
                   </colocated>
               </shared-store>
           </ha-policy>
           ...
       </core>
   </configuration>

   request-backup

   このプロパティーを true に設定すると、このブローカーはクラスター内の別のライブブローカーによって作成されるバックアップブローカーを要求します。

   max-backups

   このブローカーが作成できるバックアップブローカーの数。このプロパティーを 0 に設定すると、このブローカーはクラスターの他のブローカーからのバックアップ要求を受け入れません。

   backup-request-retries

   このブローカーがバックアップブローカーの作成を要求する回数。デフォルトは、無制限を意味する -1 です。

   backup-request-retry-interval

   バックアップブローカーを作成する要求を再試行する前にブローカーが待機する時間 ( ミリ秒単位 )。デフォルトは 5000 (5 秒) です。

   backup-port-offset

   新しいバックアップブローカーにアクセプターおよびコネクターに使用するポートオフセット。このブローカーがクラスター内の別のブローカーのバックアップを作成する要求を受信すると、この量によってポートオフセットでバックアップブローカーが作成されます。デフォルトは 100 です。

   excludes ( オプション )

   バックアップポートのオフセットからコネクターを除外します。バックアップポートのオフセットから除外する必要のある外部ブローカーのコネクターを設定している場合は、コネクターごとに <connector-ref> を追加します。

   master

   このブローカーの共有ストアまたはレプリケーションフェイルオーバーの設定。

   slave

   このブローカーのバックアップの共有ストアまたはレプリケーションのフェイルオーバー設定。

3. クラスター内の残りのブローカーごとに、この手順を繰り返します。

関連情報

• コロケートバックアップを使用するブローカークラスターの例については、HA example programs を参照してください。

ブローカークラスターがオンデマンドメッセージの負荷分散を使用する場合、メッセージ再分配 を設定して、メッセージを消費するコンシューマーを持たないキューでメッセージがストックにならないようにすることができます。

本セクションでは、以下の情報を提供します。

メッセージのグループ化により、クライアントは特定のタイプのメッセージのグループを、同じコンシューマーによって順次処理できます。クラスターの各ブローカーにグルーピングハンドラーを追加すると、クライアントはグループ化されたメッセージをクラスター内のブローカーに送信し、それらのメッセージを同じコンシューマーによって正しい順序で消費できるようにします。

グルーピングハンドラーには、ローカルハンドラーとリモートハンドラーの 2 つのタイプがあります。ブローカークラスターは、特定のグループのすべてのメッセージを適切なキューにルーティングし、目的のコンシューマーが正しい順序でそれらを使用できるようにします。

AMQ JMS クライアントを使用してクラスターに接続できます。JMS を使用すると、メッセージングクライアントを設定して、ブローカーのリストを動的または静的に検出できます。クライアント側の負荷分散を設定して、クラスター全体で接続から作成されたクライアントセッションを分散することもできます。

Red Hat Ceph Storage は、クラスター化されたオブジェクトストレージシステムです。Red Hat Ceph Storage は、オブジェクトおよびポリシーベースのレプリケーションのデータシャーディングを使用して、データの整合性とシステムの可用性を保証します。

Red Hat Ceph Storage は CRUSH (Controlled Replication Under Scalable Hashing) と呼ばれるアルゴリズムを使用して、データストレージの場所を自動的に計算してデータを保存および取得する方法を決定します。CRUSH マップ と呼ばれる Ceph アイテムを設定します。これは、クラスターのトポロジーの詳細と、ストレージクラスター全体でデータの複製方法を指定します。

CRUSH マップには、オブジェクトストレージデバイス (OSD) の一覧、デバイスを障害ドメイン階層に集約するためのバケットの一覧、および CRUSH に Ceph クラスターのプールでデータを複製する方法を指示するルールが含まれます。

インストールの基礎となる物理組織を反映することで、CRUSH マップはモデル化し、物理近接性、共有電源ソース、共有ネットワークなどの関連するデバイス障害の潜在的なソースとなります。この情報をクラスターマップにエンコードすることで、CRUSH は異なる障害ドメイン ( データセンターなど ) 間でオブジェクトレプリカを分離しつつ、ストレージクラスター全体でデータの擬似分散を維持できます。これは、データ損失を回避し、クラスターが動作が低下した状態で動作できるようにします。

Red Hat Ceph Storage クラスターでは、動作に多数のノード ( 物理または仮想 ) が必要です。クラスターには以下のタイプのノードが含まれている必要があります。