spring-amqp

Spring AMQP (RabbitMQ) - クイックリファレンス

完全なリファレンス：交換タイプ（topic、fanout）、コンシューマーパターン（複数メソッド、要求と応答）、エラー処理、DLQ 処理、テストについては、advanced.md を参照してください。

詳細な知識：包括的なドキュメントは、テクノロジーに rabbitmq を指定して mcp__documentation__fetch_docs を使用してください。

依存関係

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

設定

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    listener:
      simple:
        acknowledge-mode: manual
        concurrency: 3
        max-concurrency: 10
        prefetch: 10

交換機、キュー、バインディングの設定

@Configuration
public class RabbitMQConfig {

    public static final String EXCHANGE = "orders.exchange";
    public static final String QUEUE = "orders.queue";
    public static final String ROUTING_KEY = "orders.created";

    @Bean
    public DirectExchange ordersExchange() {
        return new DirectExchange(EXCHANGE);
    }

    @Bean
    public Queue ordersQueue() {
        return QueueBuilder.durable(QUEUE)
            .withArgument("x-dead-letter-exchange", "orders.dlx")
            .withArgument("x-dead-letter-routing-key", "dead")
            .build();
    }

    @Bean
    public Binding ordersBinding() {
        return BindingBuilder.bind(ordersQueue()).to(ordersExchange()).with(ROUTING_KEY);
    }

    @Bean
    public Jackson2JsonMessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }
}

プロデューサー

@Service
@RequiredArgsConstructor
public class OrderProducer {

    private final RabbitTemplate rabbitTemplate;

    public void sendOrder(OrderEvent event) {
        rabbitTemplate.convertAndSend(EXCHANGE, ROUTING_KEY, event);
    }

    public void sendWithHeaders(OrderEvent event, String correlationId) {
        rabbitTemplate.convertAndSend(EXCHANGE, ROUTING_KEY, event, message -> {
            message.getMessageProperties().setCorrelationId(correlationId);
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            return message;
        });
    }
}

コンシューマー

@Service
public class OrderConsumer {

    @RabbitListener(queues = "orders.queue")
    public void consume(
            @Payload OrderEvent event,
            @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
            @Header(AmqpHeaders.CORRELATION_ID) String correlationId,
            Channel channel) throws IOException {

        try {
            processOrder(event);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            channel.basicNack(deliveryTag, false, false); // Send to DLQ
        }
    }
}

ベストプラクティス

推奨事項	非推奨事項
手動確認応答を使用する	処理せずに自動確認応答する
DLX/DLQ を設定する	失敗したメッセージを通知なく失う
メッセージ TTL を設定する	メッセージを永久に保持する
永続配信を使用する	重要なメッセージに一時配信を使用する
プリフェッチを設定する	プリフェッチを無制限にする

このスキルを使用すべきでない場合

• Raw RabbitMQ - サーバー設定には rabbitmq スキルを使用してください
• Kafka との統合 - spring-kafka スキルを使用してください
• 汎用的なメッセージングパターン - messaging-expert を参照してください

アンチパターン

アンチパターン	問題	解決策
処理せずに自動確認応答する	メッセージの損失	手動確認応答を使用する
DLQ を設定しない	失敗したメッセージの損失	デッドレター交換機を設定する
プリフェッチを無制限にする	メモリの問題	適切なプリフェッチ数を設定する
リスナー内でブロックする	スレッドの枯渇	非同期処理を使用する
再試行ポリシーを設定しない	一時的な障害による損失	バックオフ付きの再試行を設定する

トラブルシューティングのクイックガイド

問題	診断	解決方法
接続が拒否される	RabbitMQ が実行中か確認する	ホスト／ポートを確認する
メッセージが消費されない	キューのバインディングを確認する	ルーティングキーを確認する
メッセージが DLQ にある	x-death ヘッダーを確認する	処理エラーを修正する
コンシューマーが起動しない	@RabbitListener を確認する	キューが存在することを確認する
シリアライズエラー	メッセージタイプを確認する	Jackson コンバーターを設定する

本番環境のチェックリスト

• 接続ファクトリーを設定済み
• 手動確認応答モードを設定済み
• DLX/DLQ を設定済み
• 再試行ポリシーを設定済み
• メッセージコンバーターを設定済み
• パブリッシャー確認応答を有効化済み
• プリフェッチ数を最適化済み
• エラーハンドラーを実装済み
• メッセージ TTL を設定済み
• モニタリングを有効化済み

リファレンスドキュメント

• Spring AMQP リファレンス