狠狠撸

Spring AMQP × RabbitMQ
@Keisuke69

自己紹介
西谷圭介

TIS株式会社

eXcaleというPaaSやってます
http://www.excale.net/
Twitter ID : @Keisuke69
Github : Keisuke69

Spring AMQPとは？
Springのサブプロジェクトの1つ

Java版と.Net版がある

比較的若いプロダクト

最初のGAリリースが2011/08

最新は1.1.3

AMQPベースのメッセージングをSpringっぽく扱える

アプリケーションロジックとインフラ周りの設定を切り離せる

メッセージ送受信用に抽象化されたtemplateを提供

POJOによるメッセージ送受信をサポート

AMQP
メッセージングミドルウェアのオープンなプロトコル仕様

プラットフォーム問わず、実装言語非依存

インターオペラビリティ高い

APIだけでなくワイヤレベルプロトコルである

JMSはAPIだけ

ビジネス上のリアルな問題を解決するようデザインされてい
る

http://www.amqp.org/

基本モデル
Server

Publisher Exchange

Binding

Consumer Message Queue

Virtual Host

Exchangeがメッセージを受け取ってMessage Queueへと受け渡す

Message QueueはMessageをためてConsumerに渡す

ExchangeとMessageQueueを対応付けるのがBinding

ExchangeのタイプはDirect、Fanout、Topic、Headerがある

MessageQueueを複数にする構成も可能

RabbitMQ
Erlangで書かれているブローカー

VMwareによって開発されている

AMQPのインプリメンテーション

クラスタリングやキューのミラーリングが可能

本来AMQPのSpecではクラスタ等については
定义されていない

なぜメッセージングか

コンポーネントを疎結合にできる

非同期

負荷分散とスケーラビリティ

Java-based container
con?guration
従来、XMLで定義していたSpringの各種設定をJavaクラス
で定義

冗長になりがちなXMLを書かなくていいので個人的に
はおすすめ

@Con?gurationをつけたクラスに、@Beanをつけたメ
ソッドを実装してbean登録

ApplicationContextはAnnotationCon?gApplicationContextで

今回のサンプルは全てこの方式で設定

ConnectionFactory
RabbitMQへの接続を管理するインター
フェース

com.rabbitmq.client.Connectionのラッ
パー

CachingConnectionFactoryだけが実装ク
ラスとして提供されている

Template
AmqpTemplate
汎用インターフェース

RabbitOperation
RabbitMQ用インターフェース

RabbitTemplate
実際のインプリメンテーション

実情として現時点ではサポートしているのがRabbitMQのみ

後々、他のAMQPミドルウェアをサポート予定（らしい）

Template
各種メソッドおよびコールバックを提供

メッセージの送受信は基本的にTemplateを利用して行
う
void send(Message message) throws AmqpException;

void convertAndSend(Object message) throws AmqpException;

Message receive() throws AmqpException;

Object receiveAndConvert() throws AmqpException;

MessageConverter
Spring AMQPの大きな特徴

オブジェクトとメッセージ間で変換が行
える

デフォルトではSimpleMessageConverter

JsonMessageConverter
MarshallingMessageConverter

Con?guration
@Configuration
public class SampleConfig {

@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
return connectionFactory;
}

@Bean
public AmqpAdmin amqpAdmin() {
return new RabbitAdmin(connectionFactory());
}

@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setRoutingKey("sample_queue");
template.setQueue("sample_queue");
return template;
}

@Bean
public Queue helloWorldQueue() {
return new Queue("sample_queue");
}
}

メッセージの送信
AmqpTemplateを使って行う

下記サンプルではアノテーションベースの設定を行なっ
ているのでAnnotationCon?tgApplicationContextを使用

public class Publisher {
? ? public static void main(String[] args){
? ? ? ? ApplicationContext context = new AnnotationConfigApplicationContext(SampleConfig.class);
? ? ? ? AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
? ? ? ? amqpTemplate.convertAndSend("Hello world");
? ? ? ? System.out.println("Sent message.");
? ? }
}

メッセージの受信
送信と同じくAmqpTemplateを使う
public class Consumer {
? ? ? ? ApplicationContext context = new
AnnotationConfigApplicationContext(SampleConfig.class);
? ? ? ? String message = (String)amqpTemplate.receiveAndConvert();
? ? ? ? System.out.println("Receive: " + message);
? ? }
}

MessageConverter
Templateのbean登録時に利用したいコンバータをセットする

今回は JsonMessageConverterを使用

JsonMessageConverterを利用する場合、Jacksonパーサ必須

@Configuration
public class SampleConfig {

（省略）

@Bean
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
template.setRoutingKey("sample_queue");
template.setQueue("sample_queue");
template.setMessageConverter(new JsonMessageConverter());
return template;
}

（省略）
}

MessageConverter
POJOを用意する

今回はシンプルにkeyとmessageというフィールドを用意してそれらに対するgetter/setter
を定義したものを用意

DB用のモデルクラスなどと兼用することも可能

public class SimplePojo {
? ? private String key;
? ? private String message;
? ?
? ? public String getKey() {
? ? ? ? return key;
? ? }
? ? public void setKey(String key) {
? ? ? ? this.key = key;
? ? }
? ? public String getMessage() {
? ? ? ? return message;
? ? }
? ? public void setMessage(String message) {
? ? ? ? this.message = message;
? ? }
}

コンバータを使った送信
POJOをnewしたオブジェクトに値をセット

AmqpTemplateのconvertAndSendの引数としてオブジェクトを渡
す
public class Publisher {
? ? ? ? ApplicationContext context = new
AnnotationConfigApplicationContext(SampleConfig.class);
? ? ? ? SimplePojo simplePojo = new SimplePojo();
? ? ? ? simplePojo.setKey("ABC");
? ? ? ? simplePojo.setMessage("This is message made by pojo.");
? ? ? ? amqpTemplate.convertAndSend(simplePojo);
? ? ? ? System.out.println("Sent message.");
? ? }
}

コンバータを使った受信
同期処理

AmqpTemplate.receiveAndConvert()
変換先の型にキャストする必要がある

public static void main(String[] args){
ApplicationContext context = new AnnotationConfigApplicationContext(SampleConfig.class);
AmqpTemplate amqpTemplate = context.getBean(RabbitTemplate.class);
SimplePojo simplePojo = (SimplePojo)amqpTemplate.receiveAndConvert();
System.out.println("Receive message: key = " + simplePojo.getKey() + ", message = " + simplePojo.getMessage());
}
}

MessageListener

非同期受信のためのインターフェース

メッセージドリブンなコールバック処理を簡
単に実装できる

SimpleMessageListenerContainer
Springが提供する軽量リスナーコンテナ

メッセージ受信時の処理をハンドラとして実装

MessageListener
HelloWorldHandlerというハンドラクラスをMessageListenerAdapterで紐付け

もちろん、MessageConverterも利用可能

@Configuration
public class SampleListenerConfig {
? ? String queueName = "sample_queue";
? ?
? ? @Bean
? ? public ConnectionFactory connectionFactory() {
? ? ? ? CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");
? ? ? ? connectionFactory.setUsername("guest");
? ? ? ? connectionFactory.setPassword("guest");
? ? ? ? return connectionFactory;
? ? }
? ? @Bean
? ? public SimpleMessageListenerContainer listenerContainer(){
? ? ? ? SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
? ? ? ? container.setConnectionFactory(connectionFactory());
? ? ? ? container.setQueueNames(queueName);
? ? ? ? container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler(),new JsonMessageConverter()));
? ? ? ? return container;
? ? }
}

ハンドラクラス
特別なクラスの継承やインターフェースの実装は不要

handleMessageというメソッドはデフォルト

変更可能

メソッドの引数として変換先の型を指定する

自動的に変換されて、getterでアクセス可能

public class HelloWorldHandler {
? ? public void handleMessage(SimplePojo simplePojo) {
? ? ? ? System.out.println("Received: Key = " + simplePojo.getKey()
???????????????? + ", message = " + simplePojo.getMessage());
? ? }
}

惭别蝉蝉补驳别尝颈蝉迟别苍别谤の起动
基本的に変わらない

標準ではApplicationContext起動時に自動でリスナーも
起動される

受信時の処理がハンドラで行われるようになっている
ためその辺りの記述が不要

? ? ? ? new AnnotationConfigApplicationContext(SampleListenerConfig.class);
? ? }
}

困ったところ
メッセージの型が固定になる

ヘッダの値を元に変換先の型が特定されている（FQCN）

PublisherがSpring以外でもこのヘッダのセットが必要

我々の場合、MQにメッセージを送る側が異なる言語で実装されていた

独自のクラスマッパーを用意

とは言ってもFQCNでの指定を不要にしただけ

1つの型につき、定義できる処理が1つ

複数のキュー、型を利用したい場合に大量に定義が必要、かつ1つの型で複数の処理
は定義できない

同じ型で異なる処理をさせたかった

汎用ハンドラクラスを用意し、型と命名規則で実際のハンドラクラスへ処理を実装

サンプル
（独自ClassMapper）
DefaultClassMapperを継承してtoClassメ
ソッドをオーバーライド

メッセージヘッダの"__TypeId__"に指定
された型名（FQCNじゃない）を使って
指定のパッケージ配下からクラスを探
しだして返却

サンプル
（独自ClassMapper）
@Override
public Class<?> toClass(MessageProperties properties) {
if (targetClass != null) {
return targetClass;
}

Map<String, Object> headers = properties.getHeaders();
Object classIdFieldNameValue = headers.get(getClassIdFieldName());
String classId = null;
if (classIdFieldNameValue != null) {
classId = classIdFieldNameValue.toString();
}
String className = null;
try {
className = ClassResolver.scan(classId, this.basePackage);
if (className != null) {
return ClassUtils.forName(className, getClass()
.getClassLoader());
}
} catch (IOException e1) {

} catch (ClassNotFoundException e) {

}
return super.toClass(properties);
}

サンプル
（汎用ハンドラクラス）
メッセージヘッダのTypeId+Handlerという名前で委
譲先のハンドラクラスを決定

“__Method__”ヘッダにセットされたメソッドを実行

全オブジェクト分のSimpleListenerContainerのbean登
録が不要に

1つのキューで複数のオブジェクトを元にしたメッ
セージの送受信が可能

サンプル
（汎用ハンドラクラス）
public class GenericHandler {

private String path;

@Autowired
BeanFactory beanFactory;

public GenericHandler() {
this.path = "org.hoge";
}

public GenericHandler(String path) {
this.path = path;
}

public void handleMessage(Object obj, MessageProperties message) {
String typeId = null;
String methodName = null;
try {
Map<String, Object> headers = message.getHeaders();
typeId = (String) headers.get("__TypeId__");
methodName = ((String) headers.get("__Method__"));
methodName = StringUtils.uncapitalize(methodName);

String delegateClassName = ClassResolver.scan(typeId + "Handler",this.path);
Object delegate = beanFactory.getBean(ClassUtils.forName(delegateClassName, getClass().getClassLoader()));

MethodInvoker methodInvoker = new MethodInvoker();
methodInvoker.setTargetObject(delegate);
methodInvoker.setTargetMethod(methodName);
Object[] objects = { obj };
methodInvoker.setArguments(objects);
methodInvoker.prepare();
methodInvoker.invoke();
} catch (Exception e) {
throw new HandlerExecutionFailureException(e);
}
}
}

その他の考慮事項
RabbitMQを冗長化した場合の話

特にMirroredQueueを利用する場合
→Codezineさんに寄稿した記事に書いておきました
?http://codezine.jp/article/detail/6943

エラー時のハンドリング

RetryTemplateとMessageRecovererとか

トランザクション

@TransactionalとTransactionManager

なぜメッセージングか
コンポーネントを疎結合にできる

MQを境界としてコンポーネント間の依存関係を切り離せる

メッセージのフォーマットさえ正しければ言語が異なってもOK

非同期

疎結合になるため、障害時などの影響を小さくできる

部分的なスケールアウトなどが可能に

負荷分散とスケーラビリティ

Consumerプロセスを増やすだけでスケールできる

SpringMQを使えば1プロセス内の並行度の設定も可能

Queueを増やすことも簡単、Exchangeの構成だけでラウンドロビン等も可能

Spring AMQPとRabbitMQ使うと

簡単に非同期アプリが書けますよ！

http://www.excale.net/

狠狠撸

Spring AMQP × RabbitMQ

Recommended

More Related Content

What's hot (20)

Similar to Spring AMQP × RabbitMQ (20)

More from Keisuke Nishitani (20)

Recently uploaded (8)

Spring AMQP × RabbitMQ