Amazon MQ ブローカーへの Java アプリケーションの接続

Amazon MQ ActiveMQ ブローカーを作成したら、ブローカーにアプリケーションを接続できます。以下の例では、Java Message Service (JMS) を使用してブローカーへの接続を作成し、キューを作成して、メッセージを送信する方法を説明します。完全な Java の実用例については、「Working Java Example」を参照してください。

ActiveMQ ブローカーには、さまざまな ActiveMQ クライアントを使用して接続できます。ActiveMQ クライアントを使用することをお勧めします。

前提条件

VPC 属性 を有効にする

VPC 内でブローカーにアクセスできることを確実にするには、enableDnsHostnames および enableDnsSupport VPC 属性を有効にする必要があります。詳細については、Amazon VPC ユーザーガイドの「VPC の DNS サポート」を参照してください。

インバウンド接続を有効にする

1. Amazon MQ コンソールにサインインします。

2. ブローカーのリストからブローカーの名前 (MyBroker など) を選択します。

3. [MyBroker] ページの [Connections] (接続) セクションで、ブローカーのウェブコンソール URL とワイヤレベルプロトコルのアドレスとポートをメモします。

4. [Details] (詳細) セクションの [Security and network] (セキュリティとネットワーク) で、セキュリティグループの名前または をクリックします。

   EC2 ダッシュボードの [セキュリティグループ] ページが表示されます。

5. セキュリティグループのリストから、セキュリティグループを選択します。

6. ページ下部で、[インバウンド] を選択し、次に [編集] を選択します。

7. [Edit inbound rules] (インバウンドルールの編集) ダイアログボックスで、パブリックアクセスを許可する URL またはエンドポイントごとにルールを追加します (以下の例は、これをブローカーのウェブコンソールに対して行う方法を説明しています)。

   1. [Add Rule] (ルールの追加) をクリックします。

   2. [タイプ] で、[カスタム TCP] を選択します。

   3. [Port Range] (ポート範囲) にはウェブコンソールポート (8162) を入力します。

   4. [Source] (ソース) では、[Custom] (カスタム) が選択された状態のままにしておき、ウェブコンソールにアクセスできるようにするシステムの IP アドレスを入力します (192.0.2.1 など)。

   5. [Save] (保存) をクリックします。

      これで、ブローカーはインバウンド接続を受け入れることができます。

Java の依存関係を追加する

activemq-client.jar パッケージと activemq-pool.jar パッケージを Java クラスパスに追加します。以下の例は、Maven プロジェクトの pom.xml ファイルにあるこれらの依存関係を示しています。

<dependencies>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.15.8</version>
    </dependency>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
        <version>5.15.8</version>
    </dependency>
</dependencies>

activemq-client.jar の詳細については、Apache ActiveMQ ドキュメントの「Initial Configuration」を参照してください。

重要

以下のコード例では、プロデューサーとコンシューマーが単一のスレッド内で実行されます。実稼働システム (またはブローカーインスタンスのフェイルオーバーをテストする) には、プロデューサーとコンシューマーが個別のホストまたはスレッドで実行されるようにしてください。

メッセージプロデューサーを作成してメッセージを送信する

1. ブローカーのエンドポイントを使用してメッセージプロデューサーの JMS プール接続ファクトリを作成してから、ファクトリに対して createConnection メソッドを呼び出します。

   注記

   アクティブ/スタンバイブローカーの場合、Amazon MQ は 2 つの ActiveMQ ウェブコンソール URL を提供しますが、一度に 1 つの URL しかアクティブになりません。同様に、Amazon MQ はワイヤレベルプロトコルごとに 2 つのエンドポイントを提供しますが、ペアごとに一度に 1 つのエンドポイントしかアクティブになりません。-1 および -2 サフィックスは冗長ペアを表します。詳細については、「Broker Architecture」を参照してください。

   ワイヤレベルプロトコルのエンドポイントについては、フェイルオーバートランスポートを使用することによって、アプリケーションがエンドポイントのどちらか一方に接続することを許可できます。

   // Create a connection factory.
   final ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(wireLevelEndpoint);
   
   // Pass the sign-in credentials.
   connectionFactory.setUserName(activeMqUsername);
   connectionFactory.setPassword(activeMqPassword);
   
   // Create a pooled connection factory.
   final PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
   pooledConnectionFactory.setConnectionFactory(connectionFactory);
   pooledConnectionFactory.setMaxConnections(10);
   
   // Establish a connection for the producer.
   final Connection producerConnection = pooledConnectionFactory.createConnection();
   producerConnection.start(); 
   
   // Close all connections in the pool.
   pooledConnectionFactory.clear();

   注記

   メッセージプロデューサーは、常に PooledConnectionFactory クラスを使用する必要があります。詳細については、「常に接続プールを使用する」を参照してください。

2. セッション、MyQueue という名前のキュー、およびメッセージプロデューサーを作成します。

   // Create a session.
   final Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   
   // Create a queue named "MyQueue".
   final Destination producerDestination = producerSession.createQueue("MyQueue");
   
   // Create a producer from the session to the queue.
   final MessageProducer producer = producerSession.createProducer(producerDestination);
   producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

3. メッセージ文字列 "Hello from Amazon MQ!" を作成してから、メッセージを送信します。

   // Create a message.
   final String text = "Hello from Amazon MQ!";
   TextMessage producerMessage = producerSession.createTextMessage(text);
   
   // Send the message.
   producer.send(producerMessage);
   System.out.println("Message sent.");

4. プロデューサーをクリーンアップします。

   producer.close();
   producerSession.close();
   producerConnection.close();

メッセージコンシューマーを作成してメッセージを受信する

1. ブローカーのエンドポイントを使用してメッセージプロデューサーの JMS 接続ファクトリを作成してから、ファクトリに対して createConnection メソッドを呼び出します。

   // Create a connection factory.
   final ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(wireLevelEndpoint);
   
   // Pass the sign-in credentials.
   connectionFactory.setUserName(activeMqUsername);
   connectionFactory.setPassword(activeMqPassword);
   
   // Establish a connection for the consumer.
   final Connection consumerConnection = connectionFactory.createConnection();
   consumerConnection.start();

   注記

   メッセージコンシューマーには、PooledConnectionFactory クラスを一切使用しないでください。詳細については、「常に接続プールを使用する」を参照してください。

2. セッション、MyQueue という名前のキュー、およびメッセージコンシューマーを作成します。

   // Create a session.
   final Session consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   
   // Create a queue named "MyQueue".
   final Destination consumerDestination = consumerSession.createQueue("MyQueue");
   
   // Create a message consumer from the session to the queue.
   final MessageConsumer consumer = consumerSession.createConsumer(consumerDestination);

3. メッセージの待機を開始し、メッセージの到着時にメッセージを受信します。

   // Begin to wait for messages.
   final Message consumerMessage = consumer.receive(1000);
   
   // Receive the message when it arrives.
   final TextMessage consumerTextMessage = (TextMessage) consumerMessage;
   System.out.println("Message received: " + consumerTextMessage.getText());

   注記

   AWS メッセージングサービス (Amazon SQS など) とは異なり、コンシューマーはブローカーに常時接続されています。

4. コンシューマー、セッション、および接続を閉じます。

   consumer.close();
   consumerSession.close();
   consumerConnection.close();