Azure Service Bus JMS 2.0 開発者ガイド

このガイドでは、Java Message Service (JMS) 2.0 API を使用してAzure Service Busとの通信を成功させるために役立つ詳細情報を提供します。

Java 開発者として、Azure Service Bus を初めて使用する場合は、次の記事をお読みください。

Java Message Service (JMS) プログラミング モデル

Java Message Service API プログラミング モデルについては、次のセクションで説明します。

JMS - 構成要素

JMS アプリケーションと通信するには、次の構成要素を使用します。

接続ファクトリ

クライアントは、接続ファクトリ オブジェクトを使用して JMS プロバイダーに接続します。 接続ファクトリは、管理者が定義する一連の接続構成パラメーターをカプセル化します。

各接続ファクトリは、 ConnectionFactory、 QueueConnectionFactory、または TopicConnectionFactory インターフェイスのインスタンスです。

これらのインターフェイスは、Azure Service Busへの接続を簡略化するために、それぞれ ServiceBusJmsConnectionFactory、ServiceBusJmsQueueConnectionFactory、または ServiceBusJmsTopicConnectionFactory によって実装されます。

TokenCredential tokenCredential = new DefaultAzureCredentialBuilder().build();

String host = "<YourNamespaceName>.servicebus.windows.net";
ConnectionFactory factory = new ServiceBusJmsConnectionFactory(tokenCredential, host, null); 

TokenCredential tokenCredential = new DefaultAzureCredentialBuilder()
                                                .managedIdentityClientId("<clientIDOfUserAssignedIdentity>")
                                                .build();

String host = "<YourNamespaceName>.servicebus.windows.net";
ConnectionFactory factory = new ServiceBusJmsConnectionFactory(tokenCredential, host, null); 

TokenCredential tokenCredential = new ClientSecretCredentialBuilder()
                .tenantId("")
                .clientId("")
                .clientSecret("")
                .build();

ConnectionFactory factory = new ServiceBusJmsConnectionFactory(SERVICE_BUS_CONNECTION_STRING, null);

JMS の送信先

宛先は、クライアントが生成するメッセージのターゲットと、クライアントが使用するメッセージのソースを指定するために使用するオブジェクトです。

送信先は、Azure Service Bus 内のエンティティにマッピングされます - キュー (ポイント間のシナリオの場合) とトピック (パブリッシュ/サブスクライブのシナリオの場合)。

接続

接続は、JMS プロバイダーとの仮想接続をカプセル化します。 Azure Service Bus では、AMQP 経由のアプリケーションと Azure Service Bus の間のステートフル接続を表します。

次の例に示すように、接続ファクトリから接続を作成します。

Connection connection = factory.createConnection();

セッション

セッションは、メッセージを生成および使用するためのシングル スレッド コンテキストです。 これを使用して、メッセージ、メッセージ プロデューサー、およびコンシューマーを作成します。 また、送信と受信をアトミック作業単位にグループ化するためのトランザクション コンテキストも提供します。

次の例に示すように、接続オブジェクトからセッションを作成します。

Session session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

セッション モード

次のいずれかのモードでセッションを作成します。

セッション モードを指定しない場合、既定値は Session.AUTO_ACKNOWLEDGE です。

JMSContext

JMSContext は、接続オブジェクトとセッション オブジェクトによって提供される機能を組み合わせたものです。 接続ファクトリ オブジェクトから作成します。

JMSContext context = connectionFactory.createContext();

JMSContext のモード

Session オブジェクトと同様に、セッション モードで説明したのと同じ受信確認モードを使用して JMSContext を作成できます。

JMSContext context = connectionFactory.createContext(JMSContext.AUTO_ACKNOWLEDGE);

モードを指定しない場合、既定値は JMSContext.AUTO_ACKNOWLEDGE です。

JMS メッセージ プロデューサー

メッセージ プロデューサーは、JMSContext または Session を使用して作成するオブジェクトです。 送信先にメッセージを送信する場合に使用します。

次の例に示すように、スタンドアロン オブジェクトとして作成できます。

JMSProducer producer = context.createProducer();

または、メッセージを送信する必要があるときに実行時に作成することもできます。

context.createProducer().send(destination, message);

JMS メッセージ コンシューマー

メッセージ コンシューマーは、JMSContext または Session によって作成されるオブジェクトです。 これを使用して、送信先に送信されたメッセージを受信します。 次の例に示すように作成します。

JMSConsumer consumer = context.createConsumer(dest);

receive() メソッドを介した同期受信

メッセージ コンシューマーは、 receive() メソッドを介して送信先からメッセージを受信する同期的な方法を提供します。

引数またはタイムアウトを指定しない場合、または 0のタイムアウトを指定した場合、メッセージが到着しない限り、または接続が切断されない限り、コンシューマーは無期限にブロックされます (どちらか早い方)。

Message m = consumer.receive();
Message m = consumer.receive(0);

0 以外の正の引数を指定すると、コンシューマーはタイマーの有効期限が切れるまでブロックします。

Message m = consumer.receive(1000); // time out after one second.

JMS メッセージ リスナーを使用した非同期受信

メッセージ リスナーは、送信先でのメッセージの非同期処理に使用するオブジェクトです。 MessageListener インターフェイスを実装します。このインターフェイスには、特定のビジネス ロジックを実行する必要があるonMessage メソッドが含まれています。

setMessageListener メソッドを使用して、メッセージ リスナー オブジェクトをインスタンス化し、特定のメッセージ コンシューマーに対して登録する必要があります。

Listener myListener = new Listener();
consumer.setMessageListener(myListener);

トピックからの使用

JMS メッセージ・コンシューマーをキューまたはトピックである宛先に対して作成します。

キュー上のコンシューマーは、クライアント アプリケーションとAzure Service Busの間のセッション (および接続) のコンテキストに存在する単なるクライアント側オブジェクトです。

ただし、トピックに関するコンシューマーには、次の 2 つの部分があります。

• Session (または JMSContext) のコンテキストに存在する クライアント側オブジェクト 。
• Azure Service Bus 上のエンティティである サブスクリプション 。

サブスクリプションは ここに 記載されており、次のいずれかの種類にすることができます。

• 共有の永続的サブスクリプション
• 共有の非永続的サブスクリプション
• 非共有の永続的サブスクリプション
• 非共有の非永続的サブスクリプション

JMS キュー ブラウザー

JMS API は、アプリケーションがキュー内のメッセージを参照し、各メッセージのヘッダー値を表示するために使用できる QueueBrowser オブジェクトを提供します。

次の例に示すように、JMSContext を使用してキュー ブラウザーを作成できます。

QueueBrowser browser = context.createBrowser(queue);

JMS メッセージ セレクター

受信側アプリケーションでは、メッセージ セレクターを使用して、受信したメッセージをフィルター処理できます。 メッセージ セレクターを使用すると、受信側のアプリケーションは、その責任を負うのではなく、JMS プロバイダー (この場合はAzure Service Bus) にメッセージをフィルター処理する作業をオフロードします。

セレクターは、次のいずれかのコンシューマーを作成するときに使用できます。

• 共有永続サブスクリプション
• 非共有の永続的なサブスクリプション
• 非永続的な共有サブスクリプション
• 非共有の非永続的サブスクリプション
• キュー コンシューマー
• キュー ブラウザー

スケジュールされたメッセージ (配信の遅延)

JMS 2.0 では、setDeliveryDelayまたはMessageProducerで JMSProducer メソッドを使用して、メッセージの将来の配信のスケジュールを設定できます。 このプロパティを設定すると、Service Busはメッセージを受け入れますが、遅延期間が経過した後にのみコンシューマーに表示されます。

MessageProducer producer = session.createProducer(queue);

// Schedule a message for delivery 30 seconds from now
producer.setDeliveryDelay(30000);
producer.send(session.createTextMessage("Scheduled message"));

完全な作業サンプルについては、azure-servicebus-jms-samples リポジトリの QueueScheduledSend.java を参照してください。

接続ファクトリの選定と信頼性

Spring Boot または JMS 接続を管理する他のフレームワークで ServiceBusJmsConnectionFactory を使用する場合は、信頼できる操作を確保するために 、送信側 と リスナー に適した接続ファクトリ ラッパーを選択します。

推奨される構成

リスナーが避けるべきこと

CachingConnectionFactory また、キャッシュされたセッションが古い基になる接続を参照する可能性があるため、リスナー コンテナーで問題が発生する可能性もあります。 リスナーの場合、ロー・ファクトリーにより、各コンテナが独立して新規接続を確立できるようになります。

Spring Cloud Azureの既定値

spring-cloud-azure-starter-servicebus-jms (バージョン 6.2.0 以降) を使用する場合、スターターは既定でこのファクトリ分離を適用します。

以前のバージョン (6.2.0 より前) では、送信者とリスナーの両方が既定で ServiceBusJmsConnectionFactory を使用するため、送信者は送信ごとに新しい接続を作成します。

例外リスナーの追加

例外リスナーがない場合、接続の切断は完全に無音になります。 監視できるように、送信者とリスナーの両方のファクトリに jakarta.jms.ExceptionListener を追加します。

connection.setExceptionListener(exception -> {
    log.error("JMS connection error: {}", exception.getMessage(), exception);
});

Spring Boot で、 CachingConnectionFactory (送信者の場合) と DefaultJmsListenerContainerFactory (リスナーの場合) に例外リスナーを設定します。

これらのすべてのパターンを示す完全な作業サンプルについては、azure-servicebus-jms-samples リポジトリの Spring Boot JMS Resilience サンプルを参照してください。

配信不能キュー

Azure Service Bus内のすべてのキューとトピックのサブスクリプションには、デッドレターキュー (DLQ) が関連付けられています。 システムは、DLQ に配信または処理できないメッセージを自動的に移動します。 たとえば、メッセージが最大配信数を超えた場合、または有効期限 (TTL) が切れると、システムはメッセージを DLQ に移動します。

JMS では、完全パスを構築し、それを使用して JmsQueue を作成することで、DLQ に別の宛先としてアクセスします。 特別な API は必要ありません。

キュー DLQ パスの形式:

<queue-name>/$deadletterqueue

トピック サブスクリプション DLQ パスの形式:

<topic-name>/Subscriptions/<subscription-name>/$deadletterqueue

例 - キューの配信不能キューからの使用:

import org.apache.qpid.jms.JmsQueue;

// Construct the DLQ path for a queue named "orders"
String dlqPath = "orders/$deadletterqueue";
JmsQueue dlqDestination = new JmsQueue(dlqPath);

// Create a consumer on the DLQ and receive messages
MessageConsumer dlqConsumer = session.createConsumer(dlqDestination);
Message message = dlqConsumer.receive(5000);

デッドレター メッセージには、メッセージが配信不能となった理由を説明するメタデータ プロパティが含まれます。

message.getStringProperty()を使用して、これらのプロパティを読み取る:

String reason = message.getStringProperty("DeadLetterReason");
String description = message.getStringProperty("DeadLetterErrorDescription");

完全な作業サンプルについては、azure-servicebus-jms-samples リポジトリの QueueDeadLetterReceive.java を参照してください。

AMQP のディスポジションと Service Bus 操作のマッピング

AMQP の処理が Service Bus 操作にどのように変換されるかを次に示します。

ACCEPTED = 1; -> Complete()
REJECTED = 2; -> DeadLetter()
RELEASED = 3; (just unlock the message in service bus, will then get redelivered)
MODIFIED_FAILED = 4; -> Abandon() which increases delivery count
MODIFIED_FAILED_UNDELIVERABLE = 5; -> Defer()

概要

この開発者ガイドでは、Java Message Service (JMS) を使用するクライアント アプリケーションJava Azure Service Busに接続する方法について説明します。

次のステップ

Azure Service Busの詳細と、Java Message Service (JMS) エンティティの詳細については、次の記事を参照してください。

• `Azure Service Bus 用の JMS とネイティブ SDK の間を選択する`
• Service Bus - キュー、トピック、サブスクリプション
• サービスバス - Java メッセージサービス エンティティ
• Azure Service Bus での AMQP 1.0 のサポート
• Service Bus AMQP 1.0 開発者ガイド
• Service Bus キューの使用
• Java Message Service API (外部 Oracle ドキュメント)
• ActiveMQ から Service Bus に移行する方法について説明します