ProducerとConsumerの実行
では、これまでに用意したConsumerとProducerの両クラスを実行してください。Producer側では、5秒おきに送信しているメッセージ内容が標準出力に表示されます。
Producerの標準出力例
2013-01-02 11:14:37,527 INFO ation.AnnotationConfigApplicationContext: 510 - Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@6be4b53d: startup date [Wed Jan 02 11:14:37 JST 2013]; root of context hierarchy 2013-01-02 11:14:37,867 INFO ation.AnnotationConfigApplicationContext:1374 - Bean 'org.springframework.scheduling.annotation.SchedulingConfiguration' of type [class org.springframework.scheduling.annotation.SchedulingConfiguration$$EnhancerByCGLIB$$db59c7fc] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying) 2013-01-02 11:14:37,938 INFO ctory.support.DefaultListableBeanFactory: 577 - Pre-instantiating singletons in org.springframework.beans.factory.support.DefaultListableBeanFactory@36927633: defining beans [org.springframework.context.annotation.internalConfigurationAnnotationProcessor,org.springframework.context.annotation.internalAutowiredAnnotationProcessor,org.springframework.context.annotation.internalRequiredAnnotationProcessor,org.springframework.context.annotation.internalCommonAnnotationProcessor,producerConfig,org.springframework.context.annotation.ConfigurationClassPostProcessor.importAwareProcessor,org.springframework.scheduling.annotation.SchedulingConfiguration,org.springframework.context.annotation.internalScheduledAnnotationProcessor,scheduledProducerTask,connectionFactory,rabbitTemplate]; root of factory hierarchy Message sent at Wed Jan 02 11:14:38 JST 2013 Message sent at Wed Jan 02 11:14:43 JST 2013 Message sent at Wed Jan 02 11:14:48 JST 2013 Message sent at Wed Jan 02 11:14:53 JST 2013
Consumer側で受け取ったメッセージが標準出力に出力されますので、Producer側と同じメッセージが出力されます。
Consumerの標準出力例
2013-01-02 11:17:11,247 INFO ation.AnnotationConfigApplicationContext: 510 - Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@428d9b22: startup date [Wed Jan 02 11:17:11 JST 2013]; root of context hierarchy 2013-01-02 11:17:11,556 INFO ctory.support.DefaultListableBeanFactory: 577 - Pre-instantiating singletons in org.springframework.beans.factory.support.DefaultListableBeanFactory@3f7de242: defining beans [org.springframework.context.annotation.internalConfigurationAnnotationProcessor,org.springframework.context.annotation.internalAutowiredAnnotationProcessor,org.springframework.context.annotation.internalRequiredAnnotationProcessor,org.springframework.context.annotation.internalCommonAnnotationProcessor,sampleListenerConfig,org.springframework.context.annotation.ConfigurationClassPostProcessor.importAwareProcessor,connectionFactory,listenerContainer]; root of factory hierarchy 2013-01-02 11:17:11,757 INFO ontext.support.DefaultLifecycleProcessor: 334 - Starting beans in phase 2147483647 Received: Message sent at Wed Jan 02 11:14:38 JST 2013 Received: Message sent at Wed Jan 02 11:14:43 JST 2013 Received: Message sent at Wed Jan 02 11:14:48 JST 2013
そのまま起動し続けるとProducer側では5秒おきにメッセージ送信が、Consumer側ではメッセージを受け取るたびにhandleMessageに定義した処理内容が実行され続けることが分かります。今回のサンプルでは、Producerでメッセージ送信されるとほぼ同時にConsumer側のハンドラが実行されるため実感しにくいのですが、Consumer側では5秒おきに処理が実行されているわけではなく、あくまでもメッセージを受け取るたびに処理が実行されています。
メッセージコンバータを利用する
SimpleMessageListenerContainerを利用した場合でも、前回紹介したメッセージコンバータが利用可能です。メッセージコンバータの指定は、MessageListenerAdapterの初期化時にハンドラクラスに加えて利用するコンバータクラスをインスタンス化して指定します。前回同様にシンプルなJavaオブジェクト（POJO）を用意し、このオブジェクトとメッセージを送受信時に変換します。まずは以下のようなオブジェクトを用意してください。
SimplePojo.java
public class SimplePojo { private int key; private String message; public int getKey() { return key; } public void setKey(int key) { this.key = key; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } }
次に、リスナー側でメッセージコンバータとして前回同様に、JsonMessageConverterを利用するようにSampleListenerConfigを変更します。
SampleListenerConfig.java（変更前）
（省略）
@Bean
public SimpleMessageListenerContainer listenerContainer(){
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueNames(queueName);
container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
return container;
}
（省略）
SampleListenerConfig.java（変更後）
（省略）
@Bean
public SimpleMessageListenerContainer listenerContainer(){
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueNames(queueName);
container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler(),new JsonMessageConverter()));
return container;
}
（省略）
送信側のソースを作成したSimplePojoを利用する形に変更します。SimplePojoのkeyには、AtomicIntegerを利用したカウンターを使って送信ごとにインクリメントした値をセットするようにしています。
ScheduledProducerTask.java（変更前）
（省略）
@Scheduled(fixedDelay=5000)
public void sendMessage() {
String message = "Message sent at " + new Date();
rabbitTemplate.convertAndSend(message);
System.out.println(message);
}
（省略）
ScheduledProducerTask.java（変更後）
（省略）
private final AtomicInteger counter = new AtomicInteger();
@Scheduled(fixedDelay=5000)
public void sendMessage() {
String message = "Message sent at " + new Date();
SimplePojo simplePojo = new SimplePojo();
simplePojo.setKey(counter.incrementAndGet());
simplePojo.setMessage(message);
rabbitTemplate.convertAndSend(simplePojo);
System.out.println(message);
}
（省略）
最後に受信側のハンドラクラスであるHelloWorldHandlerを変更します。
ScheduledProducerTask.java（変更前）
public class HelloWorldHandler { public void handleMessage(String text) { System.out.println("Received: " + text); } }
HelloWorldHandler.java（変更後）
public class HelloWorldHandler { public void handleMessage(SimplePojo simplePojo) { System.out.println("Received: Key = " + simplePojo.getKey() + ", message = " + simplePojo.getMessage()); } }
HelloWorldHandlerでは、handleMessageメソッドの引数にメッセージの受信時に変換するオブジェクトを与えています。こうすることで、Json形式で送信されたメッセージが指定されたオブジェクトに変換されます。
異なる言語のアプリケーションからメッセージを送信するには？
JsonMessageConverterでは、メッセージ受信時の変換先となる型の特定を、メッセージにセットされたヘッダの値を元に行なっています。SpringAMQPがメッセージを送信する際にメッセージのプロパティに"__TypeId__"というヘッダでそのメッセージの変換に必要なクラス名をFQCNでセットしています。JsonMessageConverterのデフォルトのクラスマッパーでは、この値を読み取ることで変換先のクラスが特定されています。
従って、異なる言語やSpringAMQPを利用していないアプリケーションからメッセージを送信する際は、メッセージの送信時に"__TypeId__"ヘッダに変換先クラスをセットしておく必要があります。
Rubyから送信する例（抜粋）
EM.run do （省略） exchange.publish({ :key => 1, :message => "message" }.to_json, :message_id => UUIDTools::UUID.random_create.to_s, :routing_key => "sample_queue", :headers => { :__TypeId__ => "org.sample.pojo.SimplePojo" }, :content_type => "application/json" ) do amqpConnection.disconnect {EM.stop} end end
なお、クラスマッパーは独自に用意することも可能です。設定時にJsonMessageConverterにご自分で用意したクラスマッパーをセットするだけです。
クラスマッパーのセット
@Bean
public SimpleMessageListenerContainer listenerContainer(){
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory());
container.setQueueNames(queueName);
JsonMessageConverter jsonMessageConverter = new JsonMessageConverter();
CustomClassMapper classMapper = new CustomClassMapper();
jsonMessageConverter.setClassMapper(classMapper);
container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler(),jsonMessageConverter));
return container;
}
