ステップ 1: ActiveMQ ブローカーを作成するステップ 2: ブローカーに Java アプリケーションを接続するステップ 3: (オプション) AWS Lambda 関数に接続するステップ 4: ブローカーを削除する次のステップ

ActiveMQ ブローカーの作成と接続

ブローカーは、Amazon MQ で実行されるメッセージブローカー環境です。これは、Amazon MQ の基本的な構成要素です。ブローカーインスタンスのクラス (m5、t3) およびサイズ (large、micro) を組み合わせた説明がブローカーインスタンスタイプ (mq.m5.large など) になります。詳細については、「ブローカー」を参照してください。

ステップ 1: ActiveMQ ブローカーを作成する

最初に実行する最も一般的な Amazon MQ タスクは、ブローカーの作成です。次の例は、を使用して基本的なブローカー AWS Management Console を作成する方法を示しています。

Amazon MQ コンソールにサインインします。
[Select broker engine] (ブローカーエンジンの選択) ページで [Apache ActiveMQ] を選択します。
[Select deployment and storage] (デプロイとストレージタイプの選択) ページの [Deployment mode and storage type] (デプロイモードとストレージタイプ) セクションで、以下を実行します。
1. [Deployment mode] (デプロイモード) を選択します ([Active/standby broker] (アクティブ/スタンバイブローカー)など)。詳細については、「Broker Architecture」を参照してください。
  - 単一インスタンスブローカーは 1 つのアベイラビリティーゾーンにある 1 つのブローカーで構成されます。ブローカーは、アプリケーション、および Amazon EBS または Amazon EFS ストレージボリュームと通信します。詳細については、「Amazon MQ 単一インスタンスブローカー」を参照してください。
  - 高可用性対応のアクティブ/スタンバイブローカーは、2 つの異なるアベイラビリティーゾーンにある 2 つのブローカーで構成され、冗長ペアで設定されます。これらのブローカーは、アプリケーションおよび Amazon EFS と同期的に通信します。詳細については、「高可用性対応の Amazon MQ アクティブ/スタンバイブローカー」を参照してください。
  - ブローカーのネットワークのサンプル設計図の詳細については、「サンプル設計図」を参照してください。
2. [Storage type] (ストレージタイプ) を選択します (EBS など)。詳細については、「Storage」を参照してください。
  
  注記
  Amazon EBS は単一のアベイラビリティーゾーン内でデータをレプリケートし、ActiveMQ アクティブ/スタンバイデプロイモードをサポートしません。
3. [Next] (次へ) をクリックします。
[Configure settings] (設定の定義) ページの [Details] (詳細) セクションで、以下を実行します。
1. [Broker name] (ブローカー名) を入力します。
  
  重要
  個人を特定できる情報 (PII) などの機密情報や秘匿性の高い情報はタグに追加しないでください。ブローカー名には、 CloudWatch ログを含む他の AWS のサービスからアクセスできます。ブローカー名は、プライベートデータや機密データとして使用することを意図していません。
2. [Broker instance type] (ブローカーインスタンスタイプ) を選択します (mq.m5.large など)。詳細については、「Broker instance types」を参照してください。
[ActiveMQ Web Console access] (ActiveMQ ウェブコンソールアクセス) セクションで、[Username] (ユーザーネーム) と [Password] (パスワード) を入力します。ブローカーのユーザー名とパスワードには、以下の制限が適用されます:
- ユーザーネームに使用できるのは、英数字、ダッシュ、ピリオド、アンダースコア、およびチルダ (- . _ ~) のみです。
- パスワードは 12 文字以上の長さで、一意の文字を少なくとも 4 つ含める必要があり、カンマ、コロン、または等号 (,:=) は使用できません。
重要
個人を特定できる情報 (PII) などの機密情報や秘匿性の高い情報はブローカーのユーザー名に追加しないでください。ブローカーのユーザー名は、 CloudWatch ログを含む他の AWS のサービスからアクセスできます。ブローカーのユーザー名は、プライベートデータや機密データとして使用することを意図していません。
[Deploy] (デプロイ) をクリックします。

Amazon MQ がブローカーを作成している間は、[Creation in progress] (作成中) ステータスが表示されます。

ブローカーの作成には約 15 分かかります。

ブローカーが正常に作成されると、Amazon MQ が [Running] (実行中) ステータスを表示します。
を選択しますMyBroker。

MyBroker ページの Connect セクションで、ブローカーの ActiveMQ ウェブコンソール URL を書き留めます。次に例を示します。
```
https://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:8162
```
また、ブローカーのワイヤレベルプロトコルの [Endpoints] (エンドポイント) もメモしておきます。 OpenWire エンドポイントの例を次に示します。
```
ssl://b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9-1.mq.us-east-2.amazonaws.com:61617
```

ステップ 2: ブローカーに Java アプリケーションを接続する

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

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

前提条件

VPC 属性を有効にする

注記

既存の Amazon MQ ブローカーのパブリックアクセシビリティを無効にすることはできません。

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

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

Amazon MQ コンソールにサインインします。
ブローカーリストから、ブローカーの名前を選択します (例: MyBroker）。
MyBroker ページの Connections セクションで、ブローカーのウェブコンソール URL とワイヤレベルのプロトコルのアドレスとポートを書き留めます。
[Details] (詳細) セクションの [Security and network] (セキュリティとネットワーク) で、セキュリティグループの名前またはをクリックします。

EC2 ダッシュボードの [セキュリティグループ] ページが表示されます。
セキュリティグループのリストから、セキュリティグループを選択します。
ページ下部で、[インバウンド] を選択し、次に [編集] を選択します。
[Edit inbound rules] (インバウンドルールの編集) ダイアログボックスで、パブリックアクセスを許可する URL またはエンドポイントごとにルールを追加します (以下の例は、これをブローカーのウェブコンソールに対して行う方法を説明しています)。
1. ルールの追加] を選択します。
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.16</version>
    </dependency>
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
        <version>5.15.16</version>
    </dependency>
</dependencies>

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

重要

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

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

ブローカーのエンドポイントを使用してメッセージプロデューサーの 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 クラスを使用する必要があります。詳細については、「常に接続プールを使用する」を参照してください。

セッション、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);

メッセージ文字列 "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.");

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


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

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

ブローカーのエンドポイントを使用してメッセージプロデューサーの 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 クラスを一切使用しないでください。詳細については、「常に接続プールを使用する」を参照してください。

セッション、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);

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


// 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 など) とは異なり、コンシューマーは常にブローカーに接続されます。

コンシューマー、セッション、および接続を閉じます。
```
consumer.close();
consumerSession.close();
consumerConnection.close();
```

ステップ 3: (オプション) AWS Lambda 関数に接続する

AWS Lambda は、Amazon MQ ブローカーに接続して、Amazon MQ ブローカーからのメッセージを使用できます。ブローカーを Lambda に接続するときは、キューからメッセージを読み取り、関数 synchronously を呼び出すイベントソースマッピングを作成します。作成するイベントソースマッピングは、ブローカーからメッセージをバッチで読み取り、それらを JSON オブジェクト形式の Lambda ペイロードに変換します。

ブローカーを Lambda 関数に接続する

Lambda 関数 execution role に以下の IAM ロール許可を追加します。
注記
必要な IAM 許可がない場合、関数は Amazon MQ リソースからレコードを正常に読み取ることができません。
(オプション) パブリックアクセシビリティがないブローカーを作成した場合は、次のいずれかを実行して、Lambda のブローカーへの接続を許可する必要があります。
- パブリックサブネットごとに 1 つの NAT ゲートウェイを設定します。詳細については、AWS Lambda デベロッパーガイドの「VPC に接続した関数のインターネットアクセスとサービスアクセス」を参照してください。
- VPC エンドポイントを使用して、Amazon Virtual Private Cloud (Amazon VPC) と Lambda 間の接続を作成します。Amazon VPC は、 AWS Security Token Service （AWS STS) および Secrets Manager エンドポイントにも接続する必要があります。詳細については、AWS Lambda デベロッパーガイドの「Lambda のインターフェイス VPC エンドポイントの設定」を参照してください。
AWS Management Consoleを使用して、Lambda 関数のイベントソースとしてブローカーを設定します。create-event-source-mapping AWS Command Line Interface コマンドを使用することもできます。

ブローカーから取り込まれたメッセージを処理するための Lambda 関数のコードをいくつか記述します。イベントソースマッピングによって取得される Lambda ペイロードは、ブローカーのエンジンタイプに依存します。以下は、Amazon MQ for ActiveMQ キューの Lambda ペイロードの例です。

注記

この例では、testQueue がキューの名前です。


{
  "eventSource": "aws:amq",
  "eventSourceArn": "arn:aws:mq:us-west-2:112556298976:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
  "messages": {
    [
      {
        "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
        "messageType": "jms/text-message",
        "data": "QUJDOkFBQUE=",
        "connectionId": "myJMSCoID",
        "redelivered": false,
        "destination": {
          "physicalname": "testQueue" 
        }, 
        "timestamp": 1598827811958,
        "brokerInTime": 1598827811958,
        "brokerOutTime": 1598827811959
      },
      {
        "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
        "messageType":"jms/bytes-message",
        "data": "3DTOOW7crj51prgVLQaGQ82S48k=",
        "connectionId": "myJMSCoID1",
        "persistent": false,
        "destination": {
          "physicalname": "testQueue" 
        }, 
        "timestamp": 1598827811958,
        "brokerInTime": 1598827811958,
        "brokerOutTime": 1598827811959
      }
    ]
  }
}

Amazon MQ の Lambda への接続、Amazon MQ イベントソースに対して Lambda がサポートするオプション、およびイベントソースマッピングエラーの詳細については、AWS Lambda デベロッパーガイドの「Amazon MQ で Lambda を使用する」を参照してください。

ステップ 4: ブローカーを削除する

Amazon MQ ブローカーを使用しない場合 (および近い将来使用することが予想される場合）、 AWS コストを削減するために Amazon MQ から削除することがベストプラクティスです。

以下の例では、 AWS Management Consoleを使用してブローカーを削除する方法を説明します。

Amazon MQ コンソールにサインインします。
ブローカーリストからブローカー ( などMyBroker) を選択し、「削除」を選択します。
Delete MyBroker? ダイアログボックスで、delete「」と入力し、「削除」を選択します。
ブローカーの削除には約 5 分かかります。

次のステップ

ブローカーを作成してアプリケーションを接続し、メッセージを送受信したので、次の操作を試してください。

Amazon MQ のベストプラクティスと Amazon MQ REST API をよく理解した上で、Amazon MQ への移行を計画することもできます。

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

開始方法

RabbitMQ ブローカーの作成と接続