Spring BootでRabbit MQクラスタを利用する


## 概要
dependencyにspring-boot-starter-amqpを追加しapplication.ymlに接続先などを設定するだけで利用可能になる。
まだ検証途中だがいろいろと試してみたので一旦メモ。

## RabbitMQのクラスタ環境で利用する

実運用ではRaabbitMQはほぼクラスタ構成にするはずなので、単純にクラスタ構成を組んだ場合以下のようになる。

![スクリーンショット 2016-01-11 19.46.43.png](https://qiita-image-store.s3.amazonaws.com/0/39230/56f969a9-aa14-cfd9-aad1-5bee32cefba2.png "スクリーンショット 2016-01-11 19.46.43.png")

```java
@Configuration
public class RabbitConfiguration {
	
	@Bean
	public Exchange testExchange() {
		return new DirectExchange("test.exchange");
	}
	
	@Bean
	public Queue queueX() {
		return new Queue("queue.x");
	}
	
	@Bean
	public Queue queueY() {
		return new Queue("queue.y");
	}
	
	@Bean
	@Autowired
	public Binding queueXBinding(@Qualifier("queueX") Queue queue,Exchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with("routing-key-x").noargs();
	}

	@Bean
	@Autowired
	public Binding queueXBinding(@Qualifier("queueY") Queue queue,Exchange exchange) {
		return BindingBuilder.bind(queue).to(exchange).with("routing-key-y").noargs();
	}


}

```

```java
@Autowired
private RabbitMessagingTemplate template;	

@Transactional
public void send() {		
	Message<String> message = new GenericMessage<>("data")
	template.send("test.exchange", "routing-key-x",message);
}
```

```java
@RabbitListener(queues="queue.X")
public class QueueXListener {	
	@RabbitHandler
	public void onMessage(@Payload String data){
	}
}
```

```yaml
spring:
    rabbitmq:
        addresses: node1:5672,node2:5672
        username: guest
        password: guest
```

* RabbitMessagingTemplateはspring.rabbitmq.addressesに指定したうちの一つのNodeに対してメッセージを送信
* Exchangeは指定されたルーティングキーに従ってQueueにメッセージをルーティング
* Consumer(@RabbitListener)は指定されたキューに接続しメッセージがQueueに入ったらメッセージを取り出し処理開始

## キューをミラーリングして利用する

クラスタ構成だけでは、ノード間でキューを認識してルーティングが可能になるだけなので、上の図ではNode1、Node2のいずれにメッセージを送信しても対象のqueue Xにメッセージが到達するが、Node1に障害が発生するとqueue Xにメッセージを送信できなくなる。
そのため、実運用ではキューを各ノードでミラーリングすることになる。
ミラーリングしたキューの利用に伴うSpringの設定変更は不要。

![スクリーンショット 2016-01-11 19.50.09.png](https://qiita-image-store.s3.amazonaws.com/0/39230/ef675581-50ec-34d9-1d96-54414292aac5.png "スクリーンショット 2016-01-11 19.50.09.png")

![スクリーンショット 2016-01-11 20.01.09.png](https://qiita-image-store.s3.amazonaws.com/0/39230/8effd9c8-3558-b7f1-7689-2c591173db9a.png "スクリーンショット 2016-01-11 20.01.09.png")

* Node2がダウンするとNode1のQueue Yがmasterに昇格する。
* ConsumerはNode1に接続して処理を継続する。
* Node1/Node2が稼働中はConsumerはどのノードに接続してもMasterからデキューして処理する。
* Node2を停止しただけでNode1のqueue Yがmasterになるので、試してはいないがNode1とNode2の接続が切れるとスプリットブレイン症候群に陥り、切断中はmasterが二つ存在した状態になり、メッセージの2重配信が発生するはず。3ノードクラスタを構築してcluster_partition_handlingパラメータでいろいろできそうではある。

## Consumerが処理中に接続が切れたりノードが停止した場合

Consumerがメッセージの処理を開始すると、メッセージの状態はReadyからUnackedになるが、メッセージを処理中にRabbitMQとの接続が切れたり、接続中のノードが停止した場合にはメッセージの状態がUnackedからReadyに戻る。そして同じキューに対するRabbitListenerがデキューし処理を再開することになる。
そのため、同じキューに対するRabbitListenerが複数存在したり、RabbitListenrの同時実行数が2以上になっている場合には、2重でメッセージが処理されることになる。

![スクリーンショット 2016-01-12 19.34.16.png](https://qiita-image-store.s3.amazonaws.com/0/39230/c4f4990f-8aca-43f6-b5e3-2f601e6dfcf2.png "スクリーンショット 2016-01-12 19.34.16.png")

## リトライ処理

Consumer側でExceptionが発生した場合にリトライするにはStatelessRetryOperationsInterceptorを利用する。
StatefullRetryOperationsInterceptorもあるが、予期しないExceptionが発生した場合、内部でリトライできないようにマークしているため、機能しない。

```java
@Bean
@Autowired	
public SimpleRabbitListenerContainerFactory springRetryConnectionFactory(ConnectionFactory cf) {
	SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
	factory.setConnectionFactory(cf);
	//リトライ処理追加3秒インターバルで3回リトライする
	factory.setAdviceChain(RetryInterceptorBuilder.stateless()
		.maxAttempts(3).backOffOptions(3000, 1.0, 3000).build());
	factory.setDefaultRequeueRejected(false);
	return factory;
}
```

```java
//containerFactoryに上で定義したBeanを指定
@Component
@RabbitListener(queues="default.queue",containerFactory="springRetryConnectionFactory")
public class DefaultQueueListener {
}
```

Exceptionが発生するとStatelessRetryOperationInterceptorがRabbitListenerの処理を再実行することになる。指定したリトライ回数を超過するとRabbitMQにエラーとして返却するため、DLQが指定されていればDLQに移動になる。

![スクリーンショット 2016-01-12 20.23.50.png](https://qiita-image-store.s3.amazonaws.com/0/39230/52c1adbf-e32e-bfb8-10a7-e3d6a7312b96.png "スクリーンショット 2016-01-12 20.23.50.png")

## リトライ処理②

たいていの場合上記のリトライでも十分だが、一回エラーになったら次のキューを優先してリトライ処理を後回しにしたい場合には、Rabbit MQ側の構成の工夫で対応することになる。

![スクリーンショット 2016-01-12 20.33.42.png](https://qiita-image-store.s3.amazonaws.com/0/39230/d19b21c3-2edf-9d38-1a4c-5ff39906cb28.png "スクリーンショット 2016-01-12 20.33.42.png")

1. Producerがメッセージsendするとdefault.queueに入る。
2. default.queueにConsumerが処理する。このときx-max-retryヘッダにリトライ回数を詰める
3. エラーになるとretry.queue(DLQ)に入る。
4. retry.queueをlistenするConsumerがx-max-retryヘッダとx-current-retryヘッダをつきあわせて、リトライ可能であれば、x-current-retryをインクリメントしてwait.queueにputする。リトライ不可能であればfailure.queueにputして終了。
5. wait.queueに指定されたTTL(インターバル時間)を超過するとDLQとして指定されたdefault.queueに移動。
6. 以降2から繰り返し。

```java
//Spring Boot起動時に自動的にRabbitMQにQueue/Exchange/Bindingの定義が反映される
@Configuration
public class RabbitConfiguration {
	
	@Bean
	public Exchange defaultExchange() {
		return new DirectExchange("default.exchange");
	}
	
	@Bean
	public Exchange errorExchange() {
		return new DirectExchange("error.exchange");
	}
	
	@Bean
	public Queue defaultQueue() {
		Map<String,Object> arguments = new HashMap<>();
		arguments.put("x-dead-letter-exchange","error.exchange");
		arguments.put("x-dead-letter-routing-key","retry.routing-key");
		return new Queue("default.queue", true, false, false, arguments);
	}
	
	@Bean
	public Queue waitQueue() {
		Map<String,Object> arguments = new HashMap<>();
		arguments.put("x-dead-letter-exchange","default.exchange");
		arguments.put("x-message-ttl",2000); //interval 2000msec
		return new Queue("wait.queue", true, false, false, arguments);
	}
	
	@Bean
	public Queue failureQueue() {
		return new Queue("failure.queue");
	}
	
	@Bean
	public Queue retryQueue() {		
		return new Queue("retry.queue");
	}

	@Bean
	@Autowired	
	public Binding defaultQueueBinding(@Qualifier("defaultQueue") Queue queue,
			@Qualifier("defaultExchange") Exchange exchange){
		return BindingBuilder.bind(queue).to(exchange).with("routing-key").noargs();
	}
	
	@Bean
	@Autowired	
	public Binding waitQueueBinding(@Qualifier("waitQueue") Queue queue,
			@Qualifier("errorExchange") Exchange exchange){
		return BindingBuilder.bind(queue).to(exchange).with("routing-key").noargs();
	}
	
	@Bean
	@Autowired	
	public Binding retryQueueBinding(@Qualifier("retryQueue") Queue queue,
			@Qualifier("errorExchange") Exchange exchange){
		return BindingBuilder.bind(queue)
								.to(exchange)
							 	.with("retry.routing-key").noargs();
	}
	
	@Bean
	@Autowired	
	public Binding failureQueueBinding(@Qualifier("failureQueue") Queue queue,
			@Qualifier("errorExchange") Exchange exchange){
		return BindingBuilder.bind(queue)
								.to(exchange)
								.with("failure.routing-key").noargs();
	}
	
	@Bean
	@Autowired	
	public SimpleRabbitListenerContainerFactory requeueRejectContainerFactory(ConnectionFactory cf) {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(cf);
		factory.setDefaultRequeueRejected(false);
		factory.setMaxConcurrentConsumers(3);
		factory.setConcurrentConsumers(3);
		return factory;
	}
}
```

```java
@Component
@RabbitListener(queues="default.queue",containerFactory="requeueRejectContainerFactory")
public class DefaultQueueListener {
	
	@RabbitHandler
	public void process(@Payload String data){
	}

}
```

```java
@Component
@RabbitListener(queues="retry.queue")
public class RetryQueueListener {
	
	@Autowired
	private RabbitMessagingTemplate template;

	@Transactional
	@RabbitHandler
	public void process(@Payload String data ,@Headers Map<String,Object> headers){
		int maxRetry = (int)headers.getOrDefault("x-max-retry", 0);
		int currentRetry = (int)headers.getOrDefault("x-current-retry", 0);
		
		Map<String,Object> newHeaders = new HashMap<>();
		headers.entrySet().stream().filter(e->e.getKey().startsWith("x-")).forEach(e->{
			newHeaders.put(e.getKey(), e.getValue());	
		});
		if(currentRetry < maxRetry){			
			currentRetry++;
			newHeaders.put("x-current-retry", currentRetry);
			GenericMessage<String> message = new GenericMessage<>(data,newHeaders);
			template.send("error.exchange","routing-key",message);
		}else{								
			GenericMessage<String> message = new GenericMessage<>(data,newHeaders);
			template.send("error.exchange","failure.routing-key",message);
		}
	}

}
```


## その他はまったところ

### データベースとRabbitMQへのメッセージ送信の疑似同期
トランザクションについてはRabbitTransactionManagerを利用可能だが、JTAには参加できないため、DBのトランザクションとRabbitMQのトランザクションの同期を取ることはできない。
ベストエフォートでやるなら、RabbitTransactionManagerを利用せず、TransactionSynchronizationManagerを利用してDBコミット後にメッセージを送信するなどの方法が考えられる。この場合メッセージ送信でエラーが発生してもDBはコミットされた状態となる。

```
@Service
public class SampleService {

	@Autowired
	private SampleRepository repository;
	
	@Autowired
	private RabbitMessagingTemplate template;
	
	@Transactional
	public void execute() throws JsonProcessingException{
		
		//DB登録
		repository.insert(new Sample());	
		
		//メッセージ送信
		send("default.exchange", "routing-key", "message to send");					
	}
	
	private <T> void send(String destination, String routingKey, String payload ){
		if( TransactionSynchronizationManager.isActualTransactionActive() ) {
			TransactionSynchronizationManager.registerSynchronization(
			new TransactionSynchronizationAdapter() {
				public void afterCommit(){
					template.convertAndSend(destination,routingKey,payload);					
				}
			});		
		}
	}
}
```

### メッセージの永久受信
デフォルトのrabbitListenerContainerFactoryは、Consumer側で予期しない例外が発生すると、DLQにも移動されずキューに残り続けるので、場合によっては無限に再実行を繰り返す。SimpleRabbitListenerContainerFactoryを自作してdefaultRequeueRejected=falseを設定する必要がある。