SpringBoot が RabbitMQ を統合してデッドレターキューと遅延キューを処理する方法

はじめに

RabbitMQ メッセージの概要

RabbitMQ メッセージは、デフォルトではタイムアウトしません。

デッドレターキューとは何ですか?遅延キューとは何ですか?

デッド レター キュー:

DLX、正式名は Dead-Letter-Exchange で、デッド レター エクスチェンジャーと呼ばれることもあり、デッド レター メールボックスと呼ぶ人もいます。メッセージがキュー内でデッド レターになった場合、別の交換に再送信できます。この交換は DLX です。DLX にバインドされたキューはデッド レター キューと呼ばれます。

次の状況では、メッセージがデッドレターになります:

• メッセージは拒否され (Basic.Reject/Basic.Nack)、requeue パラメーターはfalse に設定;

• メッセージの有効期限が切れました;

• キューが最大長に達しました。

遅延キュー:

遅延キューは、遅延メッセージを保存するために使用されます。遅延メッセージ: メッセージが送信された後、コンシューマーがすぐにメッセージを取得することを望まないが、コンシューマーが消費するメッセージを取得できるようになるまで、特定の期間待機することを意味します。

関連URL

RabbitMQにおけるデッドレターキューと遅延キューの使い方の詳しい説明

インスタンスコード

ルーティング設定

package com.example.config;
 
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class RabbitRouterConfig {
    public static final String EXCHANGE_TOPIC_WELCOME   = "Exchange@topic.welcome";
    public static final String EXCHANGE_FANOUT_UNROUTE  = "Exchange@fanout.unroute";
    public static final String EXCHANGE_TOPIC_DELAY     = "Exchange@topic.delay";
 
    public static final String ROUTINGKEY_HELLOS        = "hello.#";
    public static final String ROUTINGKEY_DELAY         = "delay.#";
 
    public static final String QUEUE_HELLO              = "Queue@hello";
    public static final String QUEUE_HI                 = "Queue@hi";
    public static final String QUEUE_UNROUTE            = "Queue@unroute";
    public static final String QUEUE_DELAY              = "Queue@delay";
 
    public static final Integer TTL_QUEUE_MESSAGE       = 5000;
 
    @Autowired
    AmqpAdmin amqpAdmin;
 
    @Bean
    Object initBindingTest() {
        amqpAdmin.declareExchange(ExchangeBuilder.fanoutExchange(EXCHANGE_FANOUT_UNROUTE).durable(true).autoDelete().build());
        amqpAdmin.declareExchange(ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_DELAY).durable(true).autoDelete().build());
        amqpAdmin.declareExchange(ExchangeBuilder.topicExchange(EXCHANGE_TOPIC_WELCOME)
                .durable(true)
                .autoDelete()
                .withArgument("alternate-exchange", EXCHANGE_FANOUT_UNROUTE)
 
                .build());
 
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_HI).build());
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_HELLO)
                .withArgument("x-dead-letter-exchange", EXCHANGE_TOPIC_DELAY)
                .withArgument("x-dead-letter-routing-key", ROUTINGKEY_DELAY)
                .withArgument("x-message-ttl", TTL_QUEUE_MESSAGE)
                .build());
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_UNROUTE).build());
        amqpAdmin.declareQueue(QueueBuilder.durable(QUEUE_DELAY).build());
 
        amqpAdmin.declareBinding(new Binding(QUEUE_HELLO, Binding.DestinationType.QUEUE,
                EXCHANGE_TOPIC_WELCOME, ROUTINGKEY_HELLOS, null));
        amqpAdmin.declareBinding(new Binding(QUEUE_UNROUTE, Binding.DestinationType.QUEUE,
                EXCHANGE_FANOUT_UNROUTE, "", null));
        amqpAdmin.declareBinding(new Binding(QUEUE_DELAY, Binding.DestinationType.QUEUE,
                EXCHANGE_TOPIC_DELAY, ROUTINGKEY_DELAY, null));
 
        return new Object();
    }
}

コントローラ

package com.example.controller;
 
import com.example.config.RabbitRouterConfig;
import com.example.mq.Sender;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;
 
import java.time.LocalDateTime;
 
@RestController
public class HelloController {
    @Autowired
    private Sender sender;
 
    @PostMapping("/hi")
    public void hi() {
        sender.send(RabbitRouterConfig.QUEUE_HI, "hi1 message:" + LocalDateTime.now());
    }
 
    @PostMapping("/hello1")
    public void hello1() {
        sender.send("hello.a", "hello1 message:" + LocalDateTime.now());
    }
 
    @PostMapping("/hello2")
    public void hello2() {
        sender.send(RabbitRouterConfig.EXCHANGE_TOPIC_WELCOME, "hello.b", "hello2 message:" + LocalDateTime.now());
    }
 
    @PostMapping("/ae")
    public void aeTest() {
        sender.send(RabbitRouterConfig.EXCHANGE_TOPIC_WELCOME, "nonono", "ae message:" + LocalDateTime.now());
    }
}

送信者

package com.example.mq;
 
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
 
import java.util.Date;
 
@Component
public class Sender {
    @Autowired
    private AmqpTemplate rabbitTemplate;
 
    public void send(String routingKey, String message) {
        this.rabbitTemplate.convertAndSend(routingKey, message);
    }
 
    public void send(String exchange, String routingKey, String message) {
        this.rabbitTemplate.convertAndSend(exchange, routingKey, message);
    }
}

受信者

package com.example.mq;
 
import com.example.config.RabbitRouterConfig;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
public class Receiver {
    @RabbitListener(queues = RabbitRouterConfig.QUEUE_HI)
    public void hi(String payload) {
        System.out.println ("Receiver(hi) : "  + payload);
    }
 
    // @RabbitListener(queues = RabbitRouterConfig.QUEUE_HELLO)
    // public void hello(String hello) throws InterruptedException {
    //     System.out.println ("Receiver(hello) : "  + hello);
    //     Thread.sleep(5 * 1000);
    //     System.out.println("(hello):sleep over");
    // }
    //
    // @RabbitListener(queues = RabbitRouterConfig.QUEUE_UNROUTE)
    // public void unroute(String hello) throws InterruptedException {
    //     System.out.println ("Receiver(unroute) : "  + hello);
    //     Thread.sleep(5 * 1000);
    //     System.out.println("(unroute):sleep over");
    // }
 
    @RabbitListener(queues = RabbitRouterConfig.QUEUE_DELAY)
    public void delay(String hello) throws InterruptedException {
        System.out.println ("Receiver(delay) : "  + hello);
        Thread.sleep(5 * 1000);
        System.out.println("(delay):sleep over");
    }
}

application.yml

server:
#  port: 9100
  port: 9101
spring:
  application:
#    name: demo-rabbitmq-sender
    name: demo-rabbitmq-receiver
  rabbitmq:
    host: localhost
    port: 5672
    username: admin
    password: 123456
#    virtualHost: /
    publisher-confirms: true
    publisher-returns: true
#    listener:
#      simple:
#        acknowledge-mode: manual
#      direct:
#        acknowledge-mode: manual

インスタンステスト

開始送信者と受信者を別々にします。

アクセス: http://localhost:9100/hello2

5 秒後の出力:

受信者(遅延): hello2 メッセージ:2020-11- 27T09 :30:51.548
(遅延):寝泊り

以上がSpringBoot が RabbitMQ を統合してデッドレターキューと遅延キューを処理する方法の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。