Java Spring FrameworkのRabbitMQ遅延メッセージ交換の例

この投稿では、RabbitMQで保留中のメッセージを使用する方法を示します。 遅延キューを使用すると便利なタスクの例として、ポストバックメカニズム（ s2s ping back 、 s2s pixel ）を取り上げます 。

ポストバックメカニズムの概要：

1.イベントがあります

2.アプリケーションは、このイベントについてサードパーティのサービスに通知する必要があります。

3.サードパーティのサービスが利用できなかった場合は、数分後に再度通知を繰り返す必要があります



再通知するために、遅延キューを使用します。



デフォルトでは、RabbitMQはメッセージの遅延方法を認識せず、発行後すぐに配信されます。 遅延配信機能は、 rabbitmq-delayed-message-exchangeプラグインとして利用できます。



プラグインは実験的なものです。 一般的に非常に安定しているという事実にもかかわらず、それはあなた自身の危険とリスクで本番で使用されるべきです。

RMQとプラグインを使用してDockerコンテナーを構築する

基礎として、テストに役立つ管理プラグインを使用して公式イメージを取得します。



Dockerfile：

FROM rabbitmq:3.6-management RUN apt-get update && apt-get install -y curl RUN curl http://www.rabbitmq.com/community-plugins/v3.6.x/rabbitmq_delayed_message_exchange-0.0.1.ez > $RABBITMQ_HOME/plugins/rabbitmq_delayed_message_exchange-0.0.1.ez RUN rabbitmq-plugins enable --offline rabbitmq_delayed_message_exchange
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

春のAMQP

Spring Frameworkは、 pring-rabbit



プロジェクトのプラグインを完全にサポートしています。 バージョン1.6.4以降では、xml / bean構成と注釈の両方を使用できます。



Spring Boot Amqp Starterを使用します。

注釈による構成

ブートストラップと注釈により、Springは作業の大部分を担います。 書くだけ：

 @RabbitListener(bindings = @QueueBinding(value = @Queue(value = DELAY_QUEUE_NAME), exchange = @Exchange(value = DELAY_EXCHANGE_NAME, delayed = "true"), key = DELAY_QUEUE_NAME)) public void onMessage(Message<?> message) { //... }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

また、アプリケーションが起動すると、RabbitAdminはdelayed exchange



、 queue



、イベントハンドラーの作成を自動的に宣言し、アノテーション付きメソッドにアタッチします。



メッセージを処理するためにより多くのスレッドが必要ですか？ これは、外部構成ファイル（ spring.rabbitmq.listener.concurrencyプロパティー）またはアノテーションのcontainerFactoryパラメーターを介して構成されます。





RabbitTemplateを使用して保留中のメッセージを送信すると便利です。

 rabbitTemplate.send( DELAY_EXCHANGE_NAME, DELAY_QUEUE_NAME, MessageBuilder .withBody(data) .setHeader("x-delay", MESSAGE_DELAY_MS).build() );
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

すぐに送信されますが、 x-delay



ヘッダーで指定された遅延で配信されます。 最大許容遅延時間（2 ^ 32-1）ms。



サンプルアプリケーションの準備：

 @SpringBootApplication public class Application { private final Logger log = LoggerFactory.getLogger(Application.class); private final static String DELAY_QUEUE_NAME = "delayed.queue"; private final static String DELAY_EXCHANGE_NAME = "delayed.exchange"; private final static String DELAY_HEADER = "x-delay"; private final static String NUM_ATTEMPT_HEADER = "x-num-attempt"; private final static long RETRY_BACKOFF = 5000; @Autowired private RabbitTemplate rabbitTemplate; @RabbitListener(bindings = @QueueBinding(value = @Queue(value = DELAY_QUEUE_NAME), exchange = @Exchange(value = DELAY_EXCHANGE_NAME, delayed = "true"), key = DELAY_QUEUE_NAME)) public void onMessage(Message<byte[]> message) { String endpointUrl = new String(message.getPayload()); Long numAttempt = (Long) message.getHeaders().getOrDefault(NUM_ATTEMPT_HEADER, 1L); log.info("Message received, url={}, attempt={}", endpointUrl, numAttempt); if (!doNotifyEndpoint(endpointUrl) && numAttempt < 3) { rabbitTemplate.send( DELAY_EXCHANGE_NAME, DELAY_QUEUE_NAME, MessageBuilder .withBody(message.getPayload()) .setHeader(DELAY_HEADER, numAttempt * RETRY_BACKOFF) .setHeader(NUM_ATTEMPT_HEADER, numAttempt + 1) .build() ); } } private boolean doNotifyEndpoint(String url) { try { HttpURLConnection connection = (HttpURLConnection) new URL(url).openConnection(); /* @todo: set up connection timeouts */ return (connection.getResponseCode() == 200); } catch (Exception e) { log.error(e.getMessage()); return false; } } public static void main(String[] args) { SpringApplication.run(Application.class, args); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

遅延配信はコントロールパネル（rmq-management）で確認します。ポート15672で利用可能です：







ログ：

 2016-12-21 14:27:25.927: Message received, url=http://localhost:1234, attempt=1 2016-12-21 14:27:25.941: Connection refused (Connection refused) 2016-12-21 14:27:30.946: Message received, url=http://localhost:1234, attempt=2 2016-12-21 14:27:30.951: Connection refused (Connection refused) 2016-12-21 14:27:40.954: Message received, url=http://localhost:1234, attempt=3
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

XML設定

XML構成を使用するtrue



、エクスチェンジBeanでdelayed



プロパティをtrue



に設定するだけで、RabbitAdminが残りを自動的に実行します。

便利なリンク

• 公式文書
• プラグインページ