Java の RabbitMQ を使用した非同期 Web-ワーカーモデル

最終更新日 2019年12月16日(月)

ワーカープロセスを使用して、長時間実行の Web リクエストを非同期的に処理することはベストプラクティスです。このアーキテクチャパターンについての詳細は、「Worker dyno、バックグラウンドジョブおよびキューイング​」の記事を参照してください。

この記事では、Spring MVC​ と RabbitMQ​ を含むサンプルの Java アプリケーションを使用してこのパターンを紹介します。これには、Heroku アドオンカタログにある RabbitMQ アドオンの 1 つである CloudAMQP アドオン​が利用されています。

はじめに

以下の手順に従って、このアプリケーションを Heroku アカウントに複製します。

  1. このサンプルアプリケーションを Heroku アカウントに複製​します。
  2. http://yourappname.herokuapp.com/spring/bigOp​ に移動して、アプリケーションを試します。
  3. http://yourappname.herokuapp.com/​ の手順に従って、Eclipse または Heroku CLI を使用して変更を加えます。

このアプリケーションのサンプルコード​は、GitHub で入手できます。この記事の残りの部分では、このパターンを実装するために使用される、アプリケーション内のいくつかの重要な Java クラスについて詳細に説明します。

アプリケーションの概要

このアプリケーションは、web​ と worker​ という 2 つのプロセスで構成されています。

  • web​ : Web リクエストを受信し、それを処理のために RabbitMQ のキューに入れる単純な Spring MVC アプリ。
  • worker​ : Spring AMQP を使用して RabbitMQ からメッセージを読み取って処理するスタンドアロンの Java アプリケーション。

これらは独立したプロセスなので、特定のアプリケーションのニーズに基づいて個別にスケーリングできます。Heroku のプロセスモデルへの理解を深めるには、「プロセスモデル​」の記事を参照してください。

RabbitMQ の設定

RabbitMQ の設定は、RabbitConfiguration.java​ を使用して行われます。これは、CloudAMQP​ アドオンによって提供される CLOUDAMQP_URL​ 環境変数を読み取り、それをアプリケーションの残りの部分から使用できるようにします。

 @Bean
 public ConnectionFactory connectionFactory() {
     final URI rabbitMqUrl;
     try {
         rabbitMqUrl = new URI(getEnvOrThrow("CLOUDAMQP_URL"));
     } catch (URISyntaxException e) {
         throw new RuntimeException(e);
     }

     final CachingConnectionFactory factory = new CachingConnectionFactory();
     factory.setUsername(rabbitMqUrl.getUserInfo().split(":")[0]);
     factory.setPassword(rabbitMqUrl.getUserInfo().split(":")[1]);
     factory.setHost(rabbitMqUrl.getHost());
     factory.setPort(rabbitMqUrl.getPort());
     factory.setVirtualHost(rabbitMqUrl.getPath().substring(1));

     return factory;
 }

Cloud AMQP アドオンを既存の Heroku アプリケーションに追加する場合は、次のコマンドを使用します。

$ heroku addons:create cloudamqp

Web プロセス

Spring MVC コントローラー BigOperationWebController.java​ は、Web リクエストを RabbitMQ のキューに入れます。このクラスには、Spring AMQP テンプレートとキュー設定 @autowired​ が含まれています。

@Autowired private AmqpTemplate amqpTemplate;
@Autowired private Queue rabbitQueue;

Web リクエストがコントローラーによって受信されると、それらのリクエストは AMPQ メッセージに変換され、RabbitMQ に送信されます。AmqpTemplate​ には次の行が含まれているため、これが容易になっています。

amqpTemplate.convertAndSend(rabbitQueue.getName(), bigOp);

web​ プロセスは、ただちにユーザーに確認ページを返します。

ワーカープロセス

worker​ プロセスは、Spring Web アプリケーションのコンテキスト外の個別のプロセスとして実行されています。そのため、その設定は RabbitConfiguration​ から明示的に接続されている必要があります。BigOperationWorker​ は、worker​ プロセスで実行されるメインの Java クラスであり、次のように RabbitMQ の設定をロードします。

ApplicationContext rabbitConfig = new AnnotationConfigApplicationContext(RabbitConfiguration.class);
ConnectionFactory rabbitConnectionFactory = rabbitConfig.getBean(ConnectionFactory.class);
Queue rabbitQueue = rabbitConfig.getBean(Queue.class);
MessageConverter messageConverter = new SimpleMessageConverter();

Spring は、キューからメッセージを受信し、そこに挿入される MessageListener に委任するための便宜クラス SimpleMessageListenerContainer​ を提供します。SimpleMessageListenerContainer​ の RabbitMQ 接続は、次のように設定されます。

SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer();
listenerContainer.setConnectionFactory(rabbitConnectionFactory);
listenerContainer.setQueueNames(rabbitQueue.getName());

このリスナーは、MessageListener​ インターフェースを実装することによって定義されます。長時間実行の BigOperation​ は、このリスナーから呼び出されます。

 listenerContainer.setMessageListener(new MessageListener() {
         public void onMessage(Message message) {
             // message is converted back into model object
             final BigOperation bigOp = (BigOperation) messageConverter.fromMessage(message);

             // simply printing out the operation, but expensive computation would happen here
             System.out.println("Received from RabbitMQ: " + bigOp);
         }
     });

キューのメッセージのリッスンを開始するには、リスナーコンテナを起動する必要があります。

listenerContainer.start();

ワーカープロセスのスケーリング

このアプリケーションの Procfile​ では、web​ と worker​ が次のように定義されます。

web: java $JAVA_OPTS -jar web/target/dependency/webapp-runner.jar --port $PORT web/target/*.war
worker: sh worker/target/bin/worker

worker​ プロセスは、次のように heroku scale​ コマンドを使用して、web​ プロセスとは独立してスケーリングできます。

$ heroku scale worker=1

このアプリケーションをテストするには、http://yourappname.herokuapp.com/spring/bigOp​ に移動し、テキストフィールドに任意の文字列を入力します。送信すると、ただちに Web リクエストが返されたことがわかり、次のアプリケーションログが表示されます。

$ heroku logs --tail
16:50:29 web.1 | Sent to RabbitMQ: BigOperation{name='text you entered'}
16:50:30 worker.1 | Received from RabbitMQ: BigOperation{name='text you entered'}

Maven プロジェクトの設定

このアプリケーションは、web、worker (2 つプロセスのそれぞれに対応)、および BigOperation.java​ モデルクラスと RabbitConfiguration.java​ 設定クラスを含む一般的なモジュールの 3 つのモジュールを含む Maven マルチモジュールプロジェクトとして構造化されています。