MirroedQueue利用時の設定
実際に本番環境で運用する際には、RabbitMQの冗長化を検討することになります。本章では、最後にRabbitMQを冗長化した場合のアプリケーション側の実装について簡単に説明します。
RabbitMQを冗長化する方法はいくつかありますが、推奨されている方法はMirrored Queueという機能を利用した実装です。
Mirrored Queueそのものの実装についてはここでは説明しませんが、簡単に触れておくと、複数台のRabbitMQサーバでクラスタを構築し、1台のマスタから複数のスレーブへとキューをミラーリングするというものです。障害時にはスレーブがマスタへと昇格し、自動的に切り替わります。ただし、接続元のアプリケーションからの接続が自動的に切り替わるわけではないため、この機能を利用する場合にアプリケーション側でそれを考慮した実装が必要です。
SpringAMQPの場合は、RabbitMQへの接続設定を行なっている箇所で、以下のように設定する必要があります。
@Bean
public ConnectionFactory connectionFactory() {
com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory = new com.rabbitmq.client.ConnectionFactory();
rabbitConnectionFactory.setConnectionTimeout(1000);
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(rabbitConnectionFactory);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
connectionFactory.setAddresses("mqserver1,mqserver2");
return connectionFactory;
}
これまで単に接続先ホストを指定してCachingConnectionFactoryを生成していましたが、まずRabbitMQのConnectionFactoryを生成しコネクションタイムアウトの設定を行います。その後、生成したRabbitMQのConnectionFactoryを引数に指定してCachingConnectionFactoryを生成します。そして生成したCachingConnectionFactoryに対して、setAddressesメソッドを用いてRabbitMQクラスタ内のノードをカンマ区切りの文字列で指定します。setAddressesメソッドでクラスタ内のサーバを登録しておけば、接続しているMQノードがダウンしているのを検知したら、その都度別のノードに接続しにいくようになります。
また、SimpleMessageListenerContainerを利用している場合に、注意すべきことがあります。それは、MirroedQueueを利用し、setAddressesメソッドでクラスタ内のサーバを指定していても、電源障害等によりサーバ自体がダウンした場合は接続中のコネクションは自動では切り替わらないということです。これはサーバ自体がダウンしたことで、ConsumerCancelNotificationという通知がRabbitMQからアプリケーション側に通知されないためです。
従ってSimpleMessageListenerContainerを利用している場合、再度接続をし直す必要があります。Consumerの再起動を行えば簡単に再接続することはできますが、その際、setAddressesメソッドで指定されたノードに対して順番に接続を試みるため、指定したリストの1番目に指定したサーバがダウンしている場合でも延々と接続完了を待ってしまうのです。その結果、Consumerの起動処理自体がタイムアウトし、起動することができません。そのためコネクションタイムアウトを設定したRabbitMQのConectionFactoryを指定して、CachingConnectionFactoryを生成することで接続がタイムアウトした場合は、次のサーバに接続を試みるようにします。
まとめ
ここまで全2回に渡り、SpringAMQPを利用してRabbitMQを使う場合の基本的な実装について解説をしてきました。SpringAMQPのSimpleMessageListnerContainerを使うことで、RabbitMQを利用したメッセージ処理を非同期に実装できます。
また、非同期であるため、複数のConsumerプロセスを起動するだけで簡単にスケールアウトも可能です。
実際の本番環境で利用する場合には、解説した内容以外にもエラーハンドリングなど考慮、実装すべき部分がありますが、複雑になりがちなエラーハンドリングやリトライなどの実装も、SpringAMQPを利用することでわずかなコード量で実装が可能となっています。
SpringAMQPは現在も活発に開発が続いているフレームワークですから、今後もより使いやすく便利になることが期待できるでしょう。機会があればせひ利用してみてください。
