Fanout 交換機の基本原理

RabbitMQ の Fanout 型交換機は、ルーティングキーの評価処理をスキップし、自身が持つバインドリストに登録されている全てのメッセージキューへ同一ペイロードを複製して転送する設計パターンです。この特性を活用することで、イベント発行・通知展開やログ分散処理など、ワンソースからマルチターゲットへの同期通信が効率的に実装できます。

1. 依存関係の取得

Spring Boot の自動構成機能を使用するため、プロジェクトビルドファイルに AMQP Starter を追加します。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2. インフラストラクチャーのプログラム的定義

GUIツールを用いた手作業登録を排除し、起動タイミングで `AmqpAdmin` インタフェースを通じて交換機・キュー・バインディング関係を自動生成します。以下の構成クラスは `ApplicationReadyEvent` を契機に基盤構築を行います。

@Component
public class BrochureInfrastructure {

    private final AmqpAdmin adminClient;

    public BrochureInfrastructure(AmqpAdmin adminClient) {
        this.adminClient = adminClient;
    }

    @EventListener(ApplicationReadyEvent.class)
    public void buildMessagingTopology() {
        String targetExchange = "broadcast_domain_exchange";

        adminClient.declareExchange(new FanoutExchange(targetExchange, true, false));

        Queue emailTarget = new Queue("dest_queue_email", true);
        Queue smsTarget = new Queue("dest_queue_sms", true);
        
        adminClient.declareQueue(emailTarget);
        adminClient.declareQueue(smsTarget);

        adminClient.declareBinding(BindingBuilder.bind(emailTarget)
                .to(new FanoutExchange(targetExchange)));
        adminClient.declareBinding(BindingBuilder.bind(smsTarget)
                .to(new FanoutExchange(targetExchange)));

        System.out.println("メッセージ配信用インフラの初期化が完了しました。");
    }
}

3. 接続パラメータの設定

YAML形式の設定ファイルに Broker エンドポイント情報を記載します。仮想パスは規定値との兼ね合いで省略可能ですが、明示的に定義することで環境分離を明確にします。

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

4. JSONシリアライズルールの適用

デフォルトの状態ではオブジェクト転送時に制限がかかるため、標準的な JSON フォーマットへの変換ルールを Bean レジストリに登録します。

@Configuration
public class DataTransformationRules {

    @Bean
    public MessageConverter jsonSerializer() {
        return new Jackson2JsonMessageConverter();
    }
}

送信対象となる DTO クラス定義：

@Data
@AllArgsConstructor
public class DistributionRecord {
    private Long traceId;
    private String eventType;
    private int priorityIndex;
}

5. メッセージ送信プロセスの実装

`RabbitTemplate` を介して交換機へデータを流し込みます。Fanout ルーティング則により、第3引数のルーティングキー部分には任意の値を渡すか空白文字列を指定します。

@Service
public class EventEmitterService {

    private final RabbitTemplate dispatcher;

    public EventEmitterService(RabbitTemplate dispatcher) {
        this.dispatcher = dispatcher;
    }

    public void emitBatchRecords(int loopCount) {
        for (int i = 0; i < loopCount; i++) {
            DistributionRecord payload = new DistributionRecord(
                (long)(i + 8000L),
                "SYSTEM_HEALTH_CHECK",
                3
            );
            
            dispatcher.convertAndSend("broadcast_domain_exchange", "", payload);
        }
        System.out.println(loopCount + "件のレコードが交換機に投入されました。");
    }
}

6. コンシューマー側リスナーの実装

`@RabbitListener` 属性により特定キューのポーリング制御を行います。前述のコンバーターが有効なため、メソッドの引数に直結した DTO 型を宣言すると自動的に JSON から Java オブジェクトへ復元されます。

@Component
public class BatchIngestionHandler {

    private static final java.util.logging.Logger log = java.util.logging.Logger.getLogger(BatchIngestionHandler.class.getName());

    private long emailProcessingCount = 0;
    private long smsProcessingCount = 0;

    @RabbitListener(queues = "dest_queue_email")
    public void processForEmail(DistributionRecord record) {
        emailProcessingCount++;
        log.info("[EMAIL CH] 処理番号:{}. ID={} | イベント種別={}",
                 emailProcessingCount, record.getTraceId(), record.getEventType());
    }

    @RabbitListener(queues = "dest_queue_sms")
    public void processForSms(DistributionRecord record) {
        smsProcessingCount++;
        log.info("[SMS CH]   処理番号:{}. ID={} | イベント種別={}",
                 smsProcessingCount, record.getTraceId(), record.getEventType());
    }
}

上記フローに従ってアプリケーションを実行すると、交換機を経由した発行パケットが自動的に重複展開され、定義された各キューで個別のスレッドが割り当てられて処理を進めます。 Broker 管理画面ではメッセージの滞留数や受領実績をリアルタイムで確認できるほか、エラー発生時の DLQ（死信キュー）連携も同様の宣言的スタイルで拡張可能です。