このバージョンはまだ開発中であり、まだ安定しているとは見なされていません。最新の安定バージョンについては、Spring AMQP 4.1.2 を使用してください！ |
汎用 AMQP 1.0 サポート
バージョン 4.1 では、AMQP 1.0 (英語) プロトコルサポート用の spring-amqp-client モジュールが導入されています。
このアーティファクトは Qpid ProtonJ2 クライアントライブラリ [GitHub] (英語) に基づいており、RabbitMQ ブローカーを含む AMQP 1.0 プロトコルをサポートするすべてのピアで動作します。
AMQP 1.0 サポートと対話できるようにするには、この依存関係をプロジェクトに追加する必要があります。
AMQP 1.0 環境
org.apache.qpid:protonj2-client は非常に柔軟で便利な API を備えているため、専用の Spring 実装がなくても Spring アプリケーションで使用できます。たとえば、org.apache.qpid.protonj2.client.Message クラスはビルダーパターンの実装です。いずれにしても、spring-amqp-client は、接続ライフサイクル管理と、メッセージの送受信操作のための AMQP 1.0 プロトコルインタラクションのための高レベルの Spring パターン実装を提供します。AMQP 1.0 環境は、org.apache.qpid.protonj2.client.Client インスタンスから始まります。
@Bean
Client protonClient() {
return Client.create();
} 同じ Client を使用して複数のブローカーに接続できますが、接続設定は接続ごとに指定する必要があります。ただし、この Bean は必須ではなく、SingleAmqpConnectionFactory が内部的にインスタンスを作成します。下記の AMQP 1.0 接続ファクトリも参照してください。
AMQP 1.0 接続ファクトリ
org.springframework.amqp.client.AmqpConnectionFactory 抽象化は、org.apache.qpid.protonj2.client.Connection を管理するために導入されました。SingleAmqpConnectionFactory 実装は、1 つの接続とその設定を管理するために存在します。同じ Connection を複数のプロデューサーとコンシューマー間で共有できます。多重化は、AMQP クライアントライブラリ内部の AMQP 1.0 プロトコル実装のリンク抽象化によって処理されます。Connection は、ConnectionOptions 用に提供される ReconnectOptions を介して回復機能を備えています。
ほとんどの場合、この Bean をプロジェクトに追加するだけで十分です。
@Bean
AmqpConnectionFactory connectionFactory() {
return new SingleAmqpConnectionFactory();
}SingleAmqpConnectionFactory setter を参照して、host、port、user を含むすべての接続固有のオプションを確認してください。より高度な設定は、ProtonJ ライブラリのビルダー API である setConnectionOptions(ConnectionOptions) を使用して行うことができます。すべてのデフォルト設定を使用した上記の SingleAmqpConnectionFactory インスタンスは、localhost:5672 に無制限に再接続します。
SingleAmqpConnectionFactory の場合、Client の注入は省略でき、対応するインスタンスはファクトリによって内部的に作成されます。 |
AmqpClient
AmqpClient は、AMQP 1.0 プロトコルに対するメッセージの送受信操作を実行するための流れるような API です。AmqpConnectionFactory が必要で、AmqpClient.Builder API を介していくつかのデフォルト値を設定できます。DefaultAmqpClient は、AmqpClient 契約の内部実装です。org.apache.qpid:protonj2-client ライブラリには org.apache.qpid.protonj2.client.Message 実装が含まれていますが、AmqpClient は、MessageProperties や MessageConverter などの抽象化を含む、よく知られている org.springframework.amqp.core.Message に基づく API を公開しています。org.apache.qpid.protonj2.client.Message との変換は、ProtonUtils サポートクラスを介して AmqpClient 実装で内部的に行われます。すべての送受信メソッドは、最終的に操作結果を取得するために CompletableFuture を返します。プレーンオブジェクトとのやり取りにはメッセージボディの変換が必要で、デフォルトでは SimpleMessageConverter が使用されます。変換の詳細については、AmqpClient.Builder.messageConverter(MessageConverter) 構成とメッセージコンバーターを参照してください。
通常、このような Bean が 1 台あれば、考えられるすべての送受信操作を実行できます。
@Bean
AmqpClient amqpClient(AmqpConnectionFactory connectionFactory) {
return AmqpClient.builder(connectionFactory)
.defaultToAddress("/queues/some_queue_as_default")
.messageConverter(new JacksonJsonMessageConverter())
.build();
} 上記の例では、defaultToAddress に使用される /queues/ プレフィックスは、RabbitMQ (英語) の宛先アドレス規則の一例です。 |
AmqpClient 操作のサンプルをいくつか示します。
CompletableFuture<Boolean> sendFuture = this.amqpClient.send(Message.create("test_data"));CompletableFuture<Boolean> sendFuture =
this.amqpClient
.to("/queues/test_queue")
.message(new org.springframework.amqp.core.Message("test_data2".getBytes()))
.send();CompletableFuture<Boolean> sendFuture =
this.amqpClient
.to("/queues/test_queue")
.body("convert")
.priority(7)
.header("test_header", "test_value")
.messageId("some_id")
.userId("guest")
.send();AmqpClient は、必要に応じてアドレスからメッセージを受信するための流れるような API も提供します。そのためには、さまざまな受信操作の動作のために ReceiveSpec を返す AmqpClient.from(String fromAddress) メソッドを呼び出す必要があります。それらはすべて、ネイティブ ProtonJ メッセージ、Spring AMQP メッセージ、またはメッセージのボディから変換されたペイロードのいずれかを含む CompletableFuture を生成します。非 Object.class 汎用引数を持つ receiveAndConvert() は、AmqpClient で messageConverter が SmartMessageConverter と同じである必要があります。例: AmqpClient に JacksonJsonMessageConverter が提供されている場合、次の例では適切なデータモデル変換が生成されます。
record TestData(String data) { }
this.amqpClient
.to("/queues/test_queue")
.body(new TestData("convert"))
.send();
CompletableFuture<TestData> receiveFuture =
this.amqpClient.from("/queues/test_queue")
.receiveAndConvert();AmqpClient によって公開される流れるような API の詳細については、これらのメソッドの Javadoc を参照してください。
AMQP 1.0 コンシューマー
AMQP 1.0 プロトコルのイベント駆動型コンシューマーは、フレームワーク内の他のすべてのリスナーコンテナー実装と同様に、AmqpMessageListenerContainer として実装されます。AmqpMessageListenerContainer は AmqpConnectionFactory を必要とし、内部的に org.apache.qpid.protonj2.client.Receiver インスタンスを使用して、指定された queueNames (実質的には AMQP 1.0 アドレス) からメッセージを消費します。デフォルトでは、100 の初期クレジットが受信リンクに使用されます。その後、配信が確定するたびに、新しいメッセージ用に一定数のクレジットが補充されます。消費されたメッセージは、指定された MessageListener によって処理され、autoAccept が false に設定されている場合、AmqpAcknowledgment 実装が、ターゲットメッセージリスナーロジックでの手動配信確定のために、それぞれの Spring AMQP メッセージプロパティに設定されます。
consumersPerQueue オプション（デフォルトは 1）は、消費対象として指定された各アドレスに対して並行処理動作を実装します。
Duration receiveTimeout オプション（デフォルトでは 1 second）は、コンシューマーが停止するまでループで呼び出されるブロッキング操作である Receiver.receive() を制御します。
各コンシューマー（内部の AmqpMessageListenerContainer.AmqpConsumer インスタンス）は、Executor を介して実行されるようにスケジュールされます。AmqpMessageListenerContainer.taskExecutor プロパティは、デフォルトで SimpleAsyncTaskExecutor に設定されています。
メッセージ処理エラーは、ErrorHandler の設定によって処理できます。
Receiver.receive() 操作が失敗すると、受信側リンクが切断され、接続自体が再確立されても ProtonJ クライアントはリンクを復元しません。コンシューマーは、このような失敗を ErrorHandler に報告し (ErrorHandler が指定されていない場合はエラーとしてログに記録)、recoveryBackOff オプションに従って受信側リンクを再開します。recoveryBackOff オプションは、デフォルトでは 5 seconds 間隔と無制限の試行回数を持つ FixedBackOff です。setRecoveryInterval(Duration) は、復旧試行の間隔のみを変更する便利なオプションです。設定された BackOff の試行回数がなくなると、コンシューマーは停止され、コンテナーから削除されます。そのため、コンシューマーが残っていない場合は、AmqpMessageListenerContainer.isRunning() は false を返します。
いくつかの AOP インターセプター（例: TransactionInterceptor）が提供されている場合、MessageListener は AmqpMessageListenerContainer でプロキシされる可能性があります。
AmqpMessageListenerContainer は、pause() および resume() API も提供し、それぞれ AMQP リンククレジットを 0 に設定および補充します。この機能により、Receiver はアクティブな状態を維持しますが、AMQP ピアからの配信は行われません。
以下の例は、AmqpMessageListenerContainer の簡単な設定方法を示しています。
BlockingQueue<Message> receivedMessages = new LinkedBlockingQueue<>();
@Bean
AmqpMessageListenerContainer amqpMessageListenerContainer(AmqpConnectionFactory connectionFactory) {
var amqpMessageListenerContainer = new AmqpMessageListenerContainer(connectionFactory);
amqpMessageListenerContainer.setQueueNames("address1", "address2");
amqpMessageListenerContainer.setConsumersPerQueue(3);
amqpMessageListenerContainer.setAutoAccept(false);
amqpMessageListenerContainer.setReceiveTimeout(Duration.ofMillis(100));
amqpMessageListenerContainer.setupMessageListener(this.receivedMessages::add);
return amqpMessageListenerContainer;
}ProtonJ Delivery の消費
便宜上、Spring AMQP メッセージの代わりにネイティブ ProtonJ Delivery オブジェクトを処理するために ProtonDeliveryListener 契約が提供されています。これは、Delivery インスタンスを完全に制御する必要があるシナリオで役立ちます。たとえば、拒否条件の追加、受信データの InputStream としての処理、ターゲットアプリケーションロジックに従ったリンククレジットの動的な補充などです。ネイティブ Delivery 処理のもう 1 つのシナリオは、プロデューサーとコンシューマーの両方が、ProtonJ ボディのエンコード / デコード内部メカニズムを含むネイティブ ProtonJ Message 契約を扱う場合です (byte[] ボディのみを受け入れる Spring AMQP メッセージとは異なります)。
AmqpMessageListenerContainer が autoAccept = true （デフォルト）に設定されている場合、手動による配送決済はスキップできます。また、AmqpMessageListenerContainer はクレジットの補充を自動的に実行します。 |
ProtonDeliveryListener の実装は、通常の MessageListener として AmqpMessageListenerContainer に注入されるべきです。
BlockingQueue<Delivery> receivedDeliveries = new LinkedBlockingQueue<>();
@Bean
AmqpMessageListenerContainer protonDeliveryListenerContainer(AmqpConnectionFactory connectionFactory) {
var amqpMessageListenerContainer = new AmqpMessageListenerContainer(connectionFactory);
amqpMessageListenerContainer.setQueueNames(TEST_QUEUE_FOR_NATIVE_PROTON);
amqpMessageListenerContainer.setAutoAccept(false);
amqpMessageListenerContainer.setupMessageListener((ProtonDeliveryListener) this.receivedDeliveries::add);
return amqpMessageListenerContainer;
}AMQP 1.0 アノテーション構成
spring-amqp-client は、アノテーションを介して AMQP 1.0 インフラストラクチャを構成する便利な方法を提供します。@EnableAmqp アノテーションは、対象プロジェクトの @Configuration クラスに設定することで、そのインフラストラクチャの登録をトリガーできます。@EnableAmqp アノテーション AmqpDefaultConfiguration によってインポートされるものは、@AmqpListener をサポートするインフラストラクチャ Bean を提供します (下記参照)。
AmqpListenerEndpointRegistry- AMQP リスナーコンテナーをプログラムで登録するためのグローバルコンポーネント。AmqpListenerAnnotationBeanPostProcessor(AmqpDefaultConfiguration.AMQP_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAMEの下) -@AmqpListenerを使用して POJO メソッドを解析し、AmqpListenerEndpointRegistryを介してリスナーコンテナーを登録するBeanPostProcessor。
@AmqpListener
高レベルで便利な設定のために、フレームワークは、フレームワークによって作成された AmqpMessageListenerContainer インスタンスから使用されるリスナー呼び出し子として POJO メソッドをマークするための @AmqpListener アノテーションを提供します。このアノテーションは @RabbitListener、@KafkaListener、@JmsListener と多くの類似点があります。AMQP アドレスを消費する必要があり、他の多くのリスナーコンテナープロパティで構成することもできます。すべての @AmqpListener 属性は、プロパティプレースホルダー (${}) および / または SpEL 式 (#{}) として構成できます。リクエスト / レスポンスメソッドシグネチャーもサポートされており、リクエストメッセージで replyTo が提供されていない場合は、フォールバック応答アドレスとして @SendTo を使用できます。これらのメソッドは、非同期戻り値型を持つことができます。Project Reactor の Mono、CompletableFuture、Kotlin の suspend 関数です。
Monoベースのリクエストレスポンスの例（可能な属性を含む）:
@Bean
AmqpMessageListenerContainerFactory customContainerFactory(AmqpConnectionFactory connectionFactory) {
var containerFactory = new MethodAmqpMessageListenerContainerFactory(connectionFactory);
containerFactory.setAutoStartup(false);
containerFactory.setAutoAccept(true);
containerFactory.setConcurrency(5);
containerFactory.setReceiveTimeout(Duration.ofSeconds(5));
containerFactory.setGracefulShutdownPeriod(Duration.ofSeconds(36));
containerFactory.setInitialCredits(55);
containerFactory.setTaskExecutor(this.taskExecutor);
containerFactory.setAdviceChain(new DebugInterceptor());
containerFactory.setListenerErrorHandler(mock());
containerFactory.setDefaultRequeueRejected(false);
containerFactory.setHeaderMapper(this.headerMapper);
containerFactory.setMessageConverter(this.messageConverter);
return containerFactory;
}
@Bean
TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
threadPoolTaskExecutor.setCorePoolSize(10);
return threadPoolTaskExecutor;
}
@Bean
AmqpHeaderMapper headerMapper() {
return new SimpleAmqpHeaderMapper();
}
@Bean
DebugInterceptor debugInterceptor() {
return new DebugInterceptor();
}
@AmqpListener(
id = "monoListener",
addresses = TEST_QUEUE2 + "," + TEST_QUEUE3,
concurrency = "2",
autoStartup = "#{true}",
autoAccept = "true",
containerFactory = "customContainerFactory",
receiveTimeout = "${receive.timeout:22000}",
gracefulShutdownPeriod = "pt2m",
initialCredits = "101",
defaultRequeueRejected = "true",
executor = "#{taskExecutor}",
messageConverter = "jsonMessageConverter",
headerMapper = "headerMapper",
adviceChain = "debugInterceptor")
Mono<DataOut> requestReplyWithJsonAndOtherAnnotationAttributes(DataIn dataIn) {
DataOut dataOut = new DataOut(dataIn.data + "_out");
return Mono.just(dataOut);
}id 属性は必須ではなく、ターゲットリスナーコンテナーの Bean 名は、クラス名、メソッド名、.listenerContainer サフィックスに基づいて決定されます。addresses は、名前の配列にすることも、各項目をカンマ区切りの名前のセットにすることもできます。これは、外部構成プロパティを介して AMQP アドレスのリストが提供される場合に便利です。containerFactory (下記参照)、executor、messageConverter、headerMapper 属性はオプションであり、上記のサンプルの executor のように SpEL 式として Bean 名前または Bean 参照に構成できます。adviceChain 属性は、Advice 実装の Bean 名のリストです。
Kotlin
suspend関数の例（明示的な@SendToを含む）:
@Bean
fun jsonMessageConverter() = JacksonJsonMessageConverter()
@Bean
fun amqpConnectionFactory(protonClient: Client) =
SingleAmqpConnectionFactory(protonClient)
.setPort(amqpPort())
@Bean
fun amqpClient(connectionFactory: AmqpConnectionFactory, jsonMessageConverter: MessageConverter) =
AmqpClient.builder(connectionFactory)
.messageConverter(jsonMessageConverter)
.build()
@Bean(AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
fun containerFactory(connectionFactory: AmqpConnectionFactory, jsonMessageConverter: MessageConverter) =
MethodAmqpMessageListenerContainerFactory(connectionFactory)
.also { it.setMessageConverter(jsonMessageConverter) }
@AmqpListener(addresses = ["TEST_QUEUE"])
@SendTo("TEST_REPLY_TO")
suspend fun requestReplyAsyncListener(payload: DataIn) = DataOut(payload.data + "_out")
data class DataIn(val data: String)
data class DataOut(val data: String)@AmqpListener は @Repeatable と同じなので、同じサービスメソッドを呼び出す各アノテーションに対して、複数のリスナーコンテナーが登録されます。
AmqpMessageListenerContainerFactory
AmqpMessageListenerContainerFactory 抽象化は、提供された AmqpListenerEndpoint に基づいて AmqpMessageListenerContainer を作成するために使用できます。ファクトリは、AmqpListenerEndpoint インスタンスによって上書きされていない場合、デフォルトのリスナーコンテナープロパティを提供します。フレームワークは、MessageListener インジェクションに基づく SimpleAmqpListenerEndpoint 実装と、特定の POJO メソッド呼び出しに基づく MethodAmqpListenerEndpoint を提供します。どちらのエンドポイント実装も、消費する AMQP アドレスを必要とします。AmqpMessageListenerContainerFactory の MethodAmqpMessageListenerContainerFactory 拡張は、MethodAmqpListenerEndpoint に基づく AmqpMessageListenerContainer インスタンス専用に設計されています (AmqpMessagingListenerAdapter は内部的にメッセージリスナーとして使用されます)。フレームワークは、AmqpListenerAnnotationBeanPostProcessor を介して @AmqpListener アノテーションが解析されるときに、このファクトリの Bean に依存します。POJO メソッドに @AmqpListener(containerFactory) 属性が提供されない場合、フレームワークは AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME を持つ AmqpMessageListenerContainerFactory のグローバル Bean を期待します。
@RabbitListener および RabbitListenerEndpointRegistry インフラストラクチャとは異なり、AmqpListenerEndpointRegistry は AmqpMessageListenerContainerFactory および AmqpListenerEndpoint を使用して AmqpMessageListenerContainer インスタンスを作成し、それをシングルトン Bean としてアプリケーションコンテキストに直接登録します。この AmqpMessageListenerContainer インスタンスの Bean 名は、AmqpListenerEndpoint.getId() の値、または MessageListener クラス名と .listenerContainer サフィックスに基づいて生成された値、あるいは MethodAmqpListenerEndpoint の場合はサービスクラス名、メソッド名、メンションサフィックスを組み合わせた値となります。
詳細については、AmqpListenerEndpoint、AmqpMessageListenerContainerFactory、AmqpListenerEndpointRegistry の Java ドキュメントを参照してください。
