🔗 Spring Framework 講座【第47回】マイクロサービス連携の要！〜メッセージキューとSpring Cloud Streamの基本〜

docs

前回、イベント駆動型アーキテクチャ（EDA）の基本として、単一のSpringアプリケーション内でアプリケーションイベントを利用する方法を学びました。

しかし、システムがマイクロサービスなど複数の独立したアプリケーションに分割されている場合、異なるアプリケーション間で非同期かつ確実にイベントを連携させる仕組みが必要です。その役割を担うのが、**メッセージキュー（Message Queue）またはメッセージブローカー（Message Broker）**です。

今回は、メッセージキューの基本概念と、Spring BootからKafkaやRabbitMQといった主要なメッセージブローカーと連携するためのフレームワーク Spring Cloud Streamの概要を学びます。

1. メッセージキューとは？

メッセージキューは、データを送信する側（プロデューサー）と、データを受信する側（コンシューマー）を間に立って仲介するシステムです。

  • 機能: プロデューサーがメッセージ（データ）をキューに格納し、コンシューマーはキューからメッセージを取り出して処理します。
  • 非同期性: メッセージがキューに格納された時点で、プロデューサーは処理を完了できます。コンシューマーは自分のペースでメッセージを取り出して処理するため、送信側と受信側の処理速度の違い（負荷）を吸収できます。
  • 永続性: メッセージはキューに保存されるため、コンシューマー側がダウンしていても、メッセージが失われることはありません（再起動後に処理が再開できます）。

1-1. 代表的なメッセージブローカー

ブローカー名特徴用途
RabbitMQ信頼性重視。古くからあり、比較的軽量。確実なメッセージ配信が必要な場合に強み。支払い処理、重要通知
Apache Kafkaスループット・スケール重視。分散ログシステムとして設計されており、大量のメッセージの高速処理と履歴保持に強み。ログ収集、リアルタイムデータ処理

2. Spring Cloud Streamの役割

メッセージキューを利用する際、アプリケーション側は、ブローカー固有のAPI（RabbitMQならAMQP、KafkaならKafka Clientなど）を直接扱う必要があります。これは、ブローカーを切り替える際にコードの大幅な変更が必要になることを意味します。

Spring Cloud Streamは、このブローカー固有のAPIを抽象化し、Springアプリケーションから統一された方法でメッセージキューを扱えるようにするフレームワークです。

  • 抽象化: 開発者はブローカーの種類（KafkaかRabbitMQか）を意識せず、インターフェースベースでメッセージの送受信ロジックを記述できます。
  • 設定のみ: ブローカーの接続先やトピック（キュー名）は、application.yml の設定で簡単に切り替えられます。

3. Spring Cloud Streamの基本的な使い方

Spring Cloud Streamでは、「メッセージの入力口（Input）」と「メッセージの出力口（Output）」を定義して利用します。

3-1. プロデューサー（メッセージ送信）の実装

メッセージを送信する側は、@EnableBinding アノテーションと StreamBridge を使用します。

  1. 依存関係: spring-cloud-starter-stream-kafka や spring-cloud-starter-stream-rabbit などを追加します。
  2. 設定ファイル: どのメッセージチャネル（トピック/キュー）を使うかを設定します。

YAML

# application.yml の設定例 (Kafkaを使用する場合)
spring:
  cloud:
    stream:
      bindings:
        # 'order-output' という名前の出力チャネルを定義
        order-output: 
          destination: orders-topic # 接続先のKafkaトピック名
          content-type: application/json
  1. メッセージ送信コード:

Java

import org.springframework.cloud.stream.function.StreamBridge;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class OrderProducerService {
    
    private final StreamBridge streamBridge; // DIで注入

    public void publishOrderCompleted(Long orderId) {
        OrderCompletedEvent event = new OrderCompletedEvent(orderId, "customer@example.com");
        
        // order-outputというチャネルに、イベントオブジェクトを送信
        boolean result = streamBridge.send("order-output", event);
        
        if (result) {
            System.out.println("Kafkaトピック 'orders-topic' へメッセージを送信しました。");
        }
    }
}

3-2. コンシューマー（メッセージ受信）の実装

メッセージを受信する側は、@EnableBinding の代わりに @Configuration と @Bean を使って、Consumer<T> または Function<T, R> を定義します。

Java

import java.util.function.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class NotificationConsumer {
    
    // Spring Cloud Streamが、設定に基づいてこのBeanをコンシューマーとして登録する
    @Bean
    public Consumer<OrderCompletedEvent> handleOrderEvent() {
        return event -> {
            System.out.println("メッセージキューからイベントを受信しました。注文ID: " + event.orderId());
            // 外部サービスへのメール送信処理などを実行
        };
    }
}

この handleOrderEvent の関数名が、application.yml で指定された入力チャネル名（例： handleOrderEvent-in-0）と自動で関連付けられ、メッセージキューからのデータがこのメソッドに渡されます。

4. まとめ：分散システムの基盤

メッセージキューは、マイクロサービスを連携させる上での基盤であり、システム全体の耐障害性とスケーラビリティを飛躍的に向上させます。Spring Cloud Streamを使うことで、この強力な仕組みをSpring Bootのコードに統一的に、かつ容易に組み込むことができます。

✅ 本日のまとめ

  • メッセージキュー（Message Queue）は、プロデューサーとコンシューマーの間でメッセージを仲介し、非同期連携と負荷吸収を実現する。
  • RabbitMQは信頼性重視、Kafkaは高スループットと履歴保持に優れる。
  • Spring Cloud Streamは、ブローカー固有のAPIを抽象化し、KafkaやRabbitMQといったメッセージキューへのアクセスを統一化するフレームワークである。
  • メッセージ送信（プロデューサー）には StreamBridge を、メッセージ受信（コンシューマー）には Consumer<T> の Bean 定義を使用する。
  • ブローカーの接続設定やチャネル名は application.yml で管理され、コードを変更せずにブローカーの切り替えが可能になる。

🔔 次回予告

分散システムと非同期連携の知識が深まったところで、次はSpringがリアクティブプログラミングの世界で進化させた非同期通信の根幹、WebFluxの理論的な背景と、非同期処理を扱うためのライブラリ Reactor の基本に入ります。

次回：【第48回】ノンブロッキングの極意！〜リアクティブプログラミングの基本（Reactor）〜 にご期待ください！

コメント

タイトルとURLをコピーしました