メッセージングフレームワーク

概要

https://spring.io/projects/spring-cloud-stream#overview

https://docs.spring.io/spring-cloud-stream/docs/current/reference/html/

メッセージングの課題

複数のメッセージキューイングシステムを使用する場合、各システムのAPI差異により開発が複雑化します。Spring Cloud Streamはこれらの差異を抽象化し、統一的なプログラミングモデルを提供します。

機能概要

Spring Cloud Streamは、イベント駆動型マイクロサービスを構築するためのフレームワークです。アプリケーションは入力/出力チャネルを通じてバインダーと通信し、設定によりメッセージングシステムと接続します。現在サポートされているメッセージングシステムはRabbitMQとKafkaです。

デザインコンセプト

一般的なメッセージング構造

プロデューサーとコンシューマーはMessageオブジェクトを通じて通信します。メッセージはMessageChannelを介して送信され、サブスクライバーが受信します。

なぜSpring Cloud Streamを使うのか

RabbitMQとKafkaのアーキテクチャ差異（例: エクスチェンジ vs トピック/パーティション）により、メッセージングシステムの切り替えが困難です。Spring Cloud Streamはこの課題を解消するための抽象化レイヤーを提供します。

実装方法

バインダーを介してアプリケーションとメッセージングシステムを分離します。統一されたチャネルインターフェースを通じて、メッセージの送受信を実現します。

メッセージ処理フロー

バインダー： メッセージングシステムとの接続を管理し、差異を隠蔽します。

チャネル： ソースとシンクの役割を分担します。

プログラミングAPIと注釈

メッセージプロデューサー

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>


spring:
  application:
    name: msg-producer

  cloud:
    stream:
      binders:  
        mqConfig:  
          type: rabbit  
          environment:  
            spring:
              rabbit:
                host: localhost
                port: 5672
                username: guest
                password: guest

      bindings: 
        output: 
          destination: msgExchange  
          content-type: application/json  
          binder: mqConfig


@EnableBinding(Source.class)
public class Producer {

    @Autowired
    private MessageChannel output;

    public String sendMessage(){
        output.send(MessageBuilder.withPayload("test").build());
        return UUID.randomUUID().toString();
    }

}


@RestController
public class MsgController {

    @Autowired
    private Producer producer;

    @GetMapping(value = "/send")
    public String send(){
        return producer.sendMessage();
    }

}


@SpringBootApplication
public class MsgProducerApp {

    public static void main(String[] args) {
        SpringApplication.run(MsgProducerApp.class, args);
    }

}

http://localhost:8080/send

メッセージコンシューマー

<dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
        </dependency>


spring:
  application:
    name: msg-consumer

  cloud:
    stream:
      binders:  
        mqConfig:  
          type: rabbit  
          environment:  
            spring:
              rabbit:
                host: localhost
                port: 5672
                username: guest
                password: guest

      bindings: 
        input: 
          destination: msgExchange  
          content-type: application/json  
          binder: mqConfig


@Component
@EnableBinding(Sink.class)
public class Consumer {

    @StreamListener(Sink.INPUT)
    public void processMessage(String message){
        System.out.println("受信データ: " + message);
    }

}


@SpringBootApplication
public class MsgConsumerApp {

    public static void main(String[] args) {
        SpringApplication.run(MsgConsumerApp.class, args);
    }

}

グループ化と永続化

クラスタ環境での課題

同じ機能のマイクロサービスが複数稼働する場合、メッセージの重複処理が発生します。

グループ設定による解決

Spring Cloud Streamのグループ機能により、同一グループ内のコンシューマーは競合関係となり、メッセージは1つのサービスにのみ届きます。

spring:
  cloud:
    stream:
      bindings:
        input:
          group: msgGroup

永続化の仕組み

グループを明示的に設定することで、アプリケーション再起動時も未処理メッセージを再取得できます。グループ未設定の場合、障害発生時にメッセージが失われる可能性があります。