SpringBootではじめるメッセージ連携

# 目次
1. メッセージ連携とは
1. 記事の趣旨
1. 前提環境
1. メッセージング基盤を構築する
1. Sourceを作る
1. Sinkを作る
1. Processorを作る
1. 動作確認
1. まとめ

# メッセージ連携とは

メッセージ連携とは、アプリケーションプログラム同士で
データや処理要求など「メッセージ」を交換することで、
複数アプリケーションを非同期に連携させるための仕組み。

マイクロサービス設計においては
メッセージを一時的に保持し、この受け渡しを仲介するミドルウェアを導入することで
各サービスを疎に保ち、スケーラビリティ向上を図る方針が採られている。

みんな大好きSpringFrameworkおよびSpringBootでは、
メッセージ連携を担うサービスの軽量開発を実現する
Spring Cloud Streamプロジェクトが注目されている。

[Spring Cloud Stream](https://spring.io/projects/spring-cloud-stream)

# 記事の趣旨
Spring Cloud Streamを使ってメッセージ連携に触れてみましょう。

完成したコードは[こちら](https://gitlab.com/kyosuke_yamagata/hello-messaging)。
記事では解説しないがテストクラス付き。

構成図は以下。読み進める中で、判らなくなったら見返してください。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/32882/ead84958-445b-798c-be23-05af00ec74d2.png)

動作イメージ知りたければ、先にページ下部の章「動作確認」を読んでください。

# 前提環境
+ Spring Tool Suiteが導入されていること。
+ JRE 8以上が導入されていること。
+ Dockerが導入されていること。

筆者の環境は以下。

```
Spring Tool Suite 3

Version: 3.9.9.RELEASE
Build Id: 201906181741
Platform: Eclipse 2019-06 (4.12.0)
```

```
$ docker version
Client: Docker Engine - Community
 Version:           19.03.2
 API version:       1.40
 Go version:        go1.12.8
(中略)
Server: Docker Engine - Community
 Engine:
  Version:          19.03.2
  API version:      1.40 (minimum version 1.12)
  Go version:       go1.12.8
(中略)
 containerd:
  Version:          v1.2.6
(中略)
 runc:
  Version:          1.0.0-rc8
(中略)
 docker-init:
  Version:          0.18.0
(中略)
$ 
```

```
$ java --version
java 9
Java(TM) SE Runtime Environment (build 9+181)
Java HotSpot(TM) 64-Bit Server VM (build 9+181, mixed mode)
$ 
```
# メッセージング基盤を構築する
導入が容易であることから、RabbitMQを採用する。

DockerHubからイメージをpullし、runでコンテナを起動する。
RabbitMQコンテナのマニュアルは[こちら](https://hub.docker.com/_/rabbitmq)。

```
$ docker pull rabbitmq
//起動コマンド
$ docker run -p 15672:15672 -p 5672:5672 rabbitmq:3-management
```

起動が完了すると、管理コンソールにアクセスできる。URLは以下。
http://localhost:15672/

IDおよびPWは、初期設定ではguest:guestとなる。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/32882/b6f98250-099f-8bff-201f-651abd5db6b8.png)


# Sourceを作る
Spring Cloud Streamではサービス外、例えばREST API等からデータを受け取って
メッセージング基盤へ連携(OUTPUT)するサービスをSourceと呼ぶ。

コード全量は[こちら](https://gitlab.com/kyosuke_yamagata/hello-messaging/blob/master/hello-source/src/main/java/com/example/hellosource/HelloSourceApplication.java)。

まずメッセージング基盤の対象キュー名を、application.propertiesに定義する。

```
spring.cloud.stream.bindings.output.destination=hello-processor
```

次に@EnableBindingアノテーションで
POJOなクラスと、メッセージング基盤の対象キューを紐づける。

加えてSourceオブジェクトをDIし
Source.output.send()メソッドを呼び出してメッセージを送信する。

```java
（前略）	
@EnableBinding(Source.class)
public class HelloSourceApplication {
	private final Source source;
	
	public HelloSourceApplication(Source source) {
		this.source=source;
	}
（中略）	
	@PostMapping
	public void tweet(@RequestBody Tweet tweet) {
		source.output().send(MessageBuilder.withPayload(tweet).build());
	}
	
	public static class Tweet {
		public String tweet;
	}

}
```

# Sinkを作る
Sourceとは逆に、メッセージング基盤から連携(INPUT)を受けて
バックエンドにデータを連携するサービスを
Spring Cloud StreamではSinkと呼ぶ。

コード全量は[こちら](https://gitlab.com/kyosuke_yamagata/hello-messaging/blob/master/hello-sink/src/main/java/com/example/hellosink/HelloSinkApplication.java)。


まずメッセージング基盤の対象キュー名を、application.propertiesに定義する。

```
spring.cloud.stream.bindings.input.destination=hello-sink
```

次に@EnableBindingアノテーションで
POJOなクラスと、メッセージング基盤の対象キューを紐づける。
また@StreamListenerアノテーションにより、
キューから受け取ったメッセージを直接、DTOと紐づける。

今回は、受け取ったDTOを標準出力に出して終了する。

```java
（前略）	
@EnableBinding(Sink.class)
public class HelloSinkApplication {
（中略）	
	@StreamListener(Sink.INPUT)
	public void print(Tweet tweet) {
		System.out.println("Received " + tweet.tweet);
	}

	public static class Tweet {
		public String tweet;
	}
}
```

# Processorを作る
メッセージング基盤から連携(INPUT)を受けて
メッセージング基盤へ連携(OUTPUT)するサービスをProcessorと呼ぶ。

コード全量は[こちら](https://gitlab.com/kyosuke_yamagata/hello-messaging/blob/master/hello-processor/src/main/java/com/example/helloprocessor/HelloProcessorApplication.java)


まずメッセージング基盤の対象キュー名をapplication.propertiesに定義する。
今回の例では、OUTPUTにSinkサービスが購読する対象キュー名を定義している。
またINPUTとして、Sourceがメッセージを送る先の対象キュー名を定義している。

```
spring.cloud.stream.bindings.output.destination=hello-sink

spring.cloud.stream.bindings.input.destination=hello-processor
```

次に@EnableBindingアノテーションで
POJOなクラスと、メッセージング基盤の対象キューを紐づける。

更に、@StreamListenerアノテーションを用いて、受け取ったメッセージをDTOに紐づけ。
加えて@SendToアノテーションで
メッセージを送る対象キューおよびメッセージ本文を、メソッド戻り値であるDTOに紐づけ。

結果として、メッセージを受信したら
その値を加工(後ろに" processing!"を付与)して、メッセージを送信する挙動となる。

```java
（前略）	
@EnableBinding(Processor.class)
public class HelloProcessorApplication {
（中略）	
	@StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
	public Tweet transform(Tweet tweet) {
		tweet.tweet += " processing!";
		return tweet;
	}
	
	public static class Tweet {
		public String tweet;
	}
}

```

# 動作確認
それぞれのプロジェクトについてmaven clean packageを実行。
target配下に生成された各Jarを以下のとおり起動する。

```
$ java -jar target/hello-source-0.0.1-SNAPSHOT.jar --server.port=8080
$ java -jar target/hello-sink-0.0.1-SNAPSHOT.jar --server.port=8082
$ java -jar target/hello-processor-0.0.1-SNAPSHOT.jar --server.port=8090
```

curlで、Sourceに対してPOSTリクエストを送る。

```
$ curl -v localhost:8080 -d '{"tweet":"Hello"}' -H 'Content-Type: application/json'
```

SinkのJarを起動したコンソールに、メッセージが出力される。

```
Received Hello processing!
```

# まとめ
SpringBootならびにRabbitMQ、Spring Cloud Streamを用いて
マイクロサービスなメッセージ連携の設計／実装方法を学んだ。

是非それぞれのサービスを沢山、起動して遊んで頂きたい。

# 参考文献
[Stream Processing with RabbitMQ](https://dataflow.spring.io/docs/stream-developer-guides/streams/standalone-stream-rabbitmq/)

[Spring Cloud Streamでマイクロサービス間メッセージング](https://blog.ik.am/entries/369)

[Spring Cloud Stream Tutorial](https://github.com/Pivotal-Japan/spring-cloud-stream-tutorial)