41.2. Consumer インターフェイスの実装
コンシューマーを実装する代替方法
コンシューマーは以下のいずれかの方法で実装できます。
イベント駆動型のコンシューマーの実装
イベント駆動型のコンシューマーでは、処理は外部イベントによって明示的に実行されます。イベントは、リスナーインターフェイスが特定のイベントソースに固有である event-listener インターフェイスを介して受信されます。
例41.4「JMXConsumer 実装」 は、Apache Camel JMX コンポーネント実装から取得した JMXConsumer クラスの実装を示しています。JMXConsumer クラスはイベント駆動型のコンシューマーの例で、org.apache.camel.impl.DefaultConsumer クラスから継承されることで実装されます。この JMXConsumer 例では、イベントは NotificationListener.handleNotification() メソッドの呼び出しによって表されます。これは JMX イベントを受信する標準的な方法です。これらの JMX イベントを受信するには、例41.4「JMXConsumer 実装」 にあるように NotificationListener インターフェイスを実装し、handleNotification() メソッドを上書きする必要があります。
例41.4 JMXConsumer 実装
package org.apache.camel.component.jmx;
import javax.management.Notification;
import javax.management.NotificationListener;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
public class JMXConsumer extends DefaultConsumer implements NotificationListener { 1
JMXEndpoint jmxEndpoint;
public JMXConsumer(JMXEndpoint endpoint, Processor processor) { 2
super(endpoint, processor);
this.jmxEndpoint = endpoint;
}
public void handleNotification(Notification notification, Object handback) { 3
try {
getProcessor().process(jmxEndpoint.createExchange(notification)); 4
} catch (Throwable e) {
handleException(e); 5
}
}
}- 1
JMXConsumerパターンは、DefaultConsumerクラスを拡張してイベント駆動型のコンシューマーの通常のパターンに従います。また、このコンシューマーは (JMX 通知で表される) JMX からのイベントを受信するように設計されているため、NotificationListenerインターフェイスを実装する必要があります。- 2
- 親エンドポイントへの参照
endpointとチェーン内の次のプロセッサーへの参照processorを引数に取るコンストラクターを少なくとも 1 つ実装する必要があります。 - 3
- JMX 通知が到着すると (
NotificationListenerで定義された)、handleNotification()メソッドは JMX によって自動的に呼び出されます。このメソッドの本文には、コンシューマーのイベント処理を実行するコードが含まれている必要があります。handleNotification()呼び出しは JMX レイヤーから発生するため、コンシューマーのスレッドモデルはJMXConsumerクラスではなく、JMX レイヤーによって暗黙的に制御されます。 - 4
- このコード行は、2 つの手順を組み合わせたものです。まず、JMX 通知オブジェクトは、Apache Camel でのイベントの汎用表現であるエクスチェンジオブジェクトに変換されます。次に、新たに作成されたエクスチェンジオブジェクトはルートの次のプロセッサーに渡されます (同期的に呼び出します) 。
- 5
handleException()メソッドは、DefaultConsumerベースクラスによって実装されます。デフォルトでは、org.apache.camel.impl.LoggingExceptionHandlerクラスを使用して例外を処理します。
handleNotification() メソッドは JMX の例に固有です。独自のイベント駆動型のコンシューマーを実装する場合、カスタムコンシューマーに実装する同様のイベントリスナーメソッドを特定する必要があります。
スケジュールされたポーリングコンシューマーの実装
スケジュールされたポーリングコンシューマーでは、ポーリングイベントはタイマークラス java.util.concurrent.ScheduledExecutorService によって自動的に生成されます。生成されたポーリングイベントを受信するには、 ScheduledPollConsumer.poll() メソッドを実装する必要があります (「コンシューマーパターンおよびスレッド」 を参照)。
例41.5「ScheduledPollConsumer 実装」 では、スケジュールされたポーリングパターンに従ってコンシューマーを実装する方法を示します。これは、ScheduledPollConsumer クラスを拡張することで実装されます。
例41.5 ScheduledPollConsumer 実装
import java.util.concurrent.ScheduledExecutorService; import org.apache.camel.Consumer; import org.apache.camel.Endpoint; import org.apache.camel.Exchange; import org.apache.camel.Message; import org.apache.camel.PollingConsumer; import org.apache.camel.Processor; import org.apache.camel.impl.ScheduledPollConsumer; public class pass:quotes[CustomConsumer] extends ScheduledPollConsumer { 1 private final pass:quotes[CustomEndpoint] endpoint; public pass:quotes[CustomConsumer](pass:quotes[CustomEndpoint] endpoint, Processor processor) { 2 super(endpoint, processor); this.endpoint = endpoint; } protected void poll() throws Exception { 3 Exchange exchange = /* Receive exchange object ... */; // Example of a synchronous processor. getProcessor().process(exchange); 4 } @Override protected void doStart() throws Exception { 5 // Pre-Start: // Place code here to execute just before start of processing. super.doStart(); // Post-Start: // Place code here to execute just after start of processing. } @Override protected void doStop() throws Exception { 6 // Pre-Stop: // Place code here to execute just before processing stops. super.doStop(); // Post-Stop: // Place code here to execute just after processing stops. } }
- 1
org.apache.camel.impl.ScheduledPollConsumerクラスを拡張して、スケジュールされたポーリングコンシューマークラス CustomConsumer を実装します。- 2
- 親エンドポイントへの参照
endpointとチェーン内の次のプロセッサーへの参照processorを引数に取るコンストラクターを少なくとも 1 つ実装する必要があります。 - 3
- スケジュールされたポーリングイベントを受信するには、
poll()メソッドを上書きします。これは、受信イベントを取得して処理するコードを配置する場所です (エクスチェンジオブジェクトで表されます)。 - 4
- この例では、イベントは同期的に処理されます。イベントを非同期的に処理する場合は、
getAsyncProcessor()を代わりに呼び出して非同期プロセッサーへの参照を使用する必要があります。イベントを非同期的に処理する方法は、「非同期処理」 を参照してください。 - 5
- (オプション) コンシューマーの開始時にコードを実行する場合は、以下のように
doStart()メソッドを上書きします。 - 6
- (オプション) コンシューマーの停止時にコードを実行する場合は、以下のように
doStop()メソッドを上書きします。
ポーリングコンシューマーの実装
例41.6「PollingConsumerSupport 実装」 では、ポーリングパターンに従ってコンシューマーを実装する方法を説明します。これは、PollingConsumerSupport クラスを拡張することで実装されます。
例41.6 PollingConsumerSupport 実装
import org.apache.camel.Exchange; import org.apache.camel.RuntimeCamelException; import org.apache.camel.impl.PollingConsumerSupport; public class pass:quotes[CustomConsumer] extends PollingConsumerSupport { 1 private final pass:quotes[CustomEndpoint] endpoint; public pass:quotes[CustomConsumer](pass:quotes[CustomEndpoint] endpoint) { 2 super(endpoint); this.endpoint = endpoint; } public Exchange receiveNoWait() { 3 Exchange exchange = /* Obtain an exchange object. */; // Further processing ... return exchange; } public Exchange receive() { 4 // Blocking poll ... } public Exchange receive(long timeout) { 5 // Poll with timeout ... } protected void doStart() throws Exception { 6 // Code to execute whilst starting up. } protected void doStop() throws Exception { // Code to execute whilst shutting down. } }
- 1
org.apache.camel.impl.PollingConsumerSupportクラスを拡張して、ポーリングコンシューマークラス CustomConsumer を実装します。- 2
- 親エンドポイントへの参照
endpointを引数に取るコンストラクターを少なくとも 1 つ実装する必要があります。ポーリングコンシューマーはプロセッサーインスタンスへの参照を必要としません。 - 3
receiveNoWait()メソッドは、イベント (エクスチェンジオブジェクト) を取得するための非ブロッキングアルゴリズムを実装する必要があります。利用できるイベントがない場合は、nullが返されます。- 4
- この
receive()メソッドでは、イベントを取得するためのブロッキングのアルゴリズムを実装する必要があります。このメソッドは、イベントが利用できない状態が続く場合に無期限にブロックできます。 - 5
receive(long timeout)メソッドは、指定したタイムアウト (通常はミリ秒単位で指定) までブロックできるアルゴリズムを実装します。- 6
- コンシューマーの起動またはシャットダウン中に実行するコードを挿入する場合は、
doStart()メソッドとdoStop()メソッドをそれぞれ実装します。
カスタムスレッドの実装
標準のコンシューマーパターンがコンシューマーの実装に適さない場合には、Consumer インターフェイスを直接実装してスレッドコードを作成することができます。ただし、スレッドコードを作成する場合は、「スレッドモデル」 に説明されているように、標準の Apache Camel スレッドモデルに従うことが重要です。
たとえば、camel-core にある SEDA コンポーネントは Apache Camel スレッドモデルに一貫性のある独自のコンシューマースレッドを実装しています。例41.7「カスタムスレッド実装」 クラスがスレッドを実装する方法の概要を SedaConsumer に示しています。
例41.7 カスタムスレッド実装
package org.apache.camel.component.seda;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.camel.Consumer;
import org.apache.camel.Endpoint;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.ShutdownRunningTask;
import org.apache.camel.impl.LoggingExceptionHandler;
import org.apache.camel.impl.ServiceSupport;
import org.apache.camel.util.ServiceHelper;
...
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* A Consumer for the SEDA component.
*
* @version $Revision: 922485 $
*/
public class SedaConsumer extends ServiceSupport implements Consumer, Runnable, ShutdownAware { 1
private static final transient Log LOG = LogFactory.getLog(SedaConsumer.class);
private SedaEndpoint endpoint;
private Processor processor;
private ExecutorService executor;
...
public SedaConsumer(SedaEndpoint endpoint, Processor processor) {
this.endpoint = endpoint;
this.processor = processor;
}
...
public void run() { 2
BlockingQueue<Exchange> queue = endpoint.getQueue();
// Poll the queue and process exchanges
...
}
...
protected void doStart() throws Exception { 3
int poolSize = endpoint.getConcurrentConsumers();
executor = endpoint.getCamelContext().getExecutorServiceStrategy()
.newFixedThreadPool(this, endpoint.getEndpointUri(), poolSize); 4
for (int i = 0; i < poolSize; i++) { 5
executor.execute(this);
}
endpoint.onStarted(this);
}
protected void doStop() throws Exception { 6
endpoint.onStopped(this);
// must shutdown executor on stop to avoid overhead of having them running
endpoint.getCamelContext().getExecutorServiceStrategy().shutdownNow(executor); 7
if (multicast != null) {
ServiceHelper.stopServices(multicast);
}
}
...
//----------
// Implementation of ShutdownAware interface
public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
// deny stopping on shutdown as we want seda consumers to run in case some other queues
// depend on this consumer to run, so it can complete its exchanges
return true;
}
public int getPendingExchangesSize() {
// number of pending messages on the queue
return endpoint.getQueue().size();
}
}- 1
SedaConsumerクラスは、org.apache.camel.impl.ServiceSupportクラスを拡張し、Consumer、Runnable、およびShutdownAwareインターフェイスを実装することで実装されます。- 2
Runnable.run()メソッドを実装し、スレッドで実行中にコンシューマーの動作を定義します。この場合、コンシューマーはループで実行され、新しいエクスチェンジのためにキューをポーリングし、キューの後半でエクスチェンジを処理します。- 3
doStart()メソッドはServiceSupportから継承されます。このメソッドは、コンシューマーの起動時に行う動作を定義するために上書きされます。- 4
- スレッドを直接作成するのではなく、
CamelContextと共に登録されたExecutorServiceStrategyオブジェクトを使用してスレッドプールを作成する必要があり ます。これは、Apache Camel がスレッドの集中管理を実装し、正常なシャットダウンなどの機能をサポートすることができるため、重要になります。詳細は、「スレッドモデル」 を参照してください。 - 5
ExecutorService.execute()メソッドpoolSizeを呼び出し、スレッドを開始します。- 6
doStop()メソッドはServiceSupportから継承されます。このメソッドは、シャットダウン時にコンシューマーの動作を定義するために上書きされます。- 7
executorインスタンスが表すスレッドプールをシャットダウンします。