小編給大家分享一下SpringBoot中如何基于RabbitMQ實(shí)現(xiàn)消息延遲隊(duì)列，相信大部分人都還不怎么了解，因此分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后大有收獲，下面讓我們一起去了解一下吧！

創(chuàng)新互聯(lián)-云計算及IDC服務(wù)提供商，涵蓋公有云、IDC機(jī)房租用、溫江服務(wù)器租用、等保安全、私有云建設(shè)等企業(yè)級互聯(lián)網(wǎng)基礎(chǔ)服務(wù)，聯(lián)系電話：13518219792

延時隊(duì)列使用場景

>在很多的業(yè)務(wù)場景中，延時隊(duì)列可以實(shí)現(xiàn)很多功能，此類業(yè)務(wù)中，一般上是非實(shí)時的，需要延遲處理的，需要進(jìn)行重試補(bǔ)償?shù)摹?/p>

1. 訂單超時關(guān)閉：在支付場景中，一般上訂單在創(chuàng)建后30分鐘或1小時內(nèi)未支付的，會自動取消訂單。

2. 短信或者郵件通知：在一些注冊或者下單業(yè)務(wù)時，需要在1分鐘或者特定時間后進(jìn)行短信或者郵件發(fā)送相關(guān)資料的。本身此類業(yè)務(wù)于主業(yè)務(wù)是無關(guān)聯(lián)性的，一般上的做法是進(jìn)行異步發(fā)送。

3. 重試場景：比如消息通知，在第一次通知出現(xiàn)異常時，會在隔幾分鐘之后進(jìn)行再次重試發(fā)送。

RabbitMQ實(shí)現(xiàn)延時隊(duì)列

>本身在RabbitMQ中是未直接提供延時隊(duì)列功能的，但可以使用TTL(Time-To-Live，存活時間)和DLX(Dead-Letter-Exchange，死信隊(duì)列交換機(jī))的特性實(shí)現(xiàn)延時隊(duì)列的功能。

存活時間（Time-To-Live 簡稱 TTL）

>RabbitMQ中可以對隊(duì)列和消息分別設(shè)置TTL，TTL表明了一條消息可在隊(duì)列中存活的最大時間。當(dāng)某條消息被設(shè)置了TTL或者當(dāng)某條消息進(jìn)入了設(shè)置了TTL的隊(duì)列時，這條消息會在TTL時間后**死亡成為Dead Letter**。如果既配置了消息的TTL，又配置了隊(duì)列的TTL，那么較小的那個值會被取用。

死信交換（Dead Letter Exchanges 簡稱 DLX）

>上個知識點(diǎn)也提到了，設(shè)置了TTL的消息或隊(duì)列最終會成為Dead Letter，當(dāng)消息在一個隊(duì)列中變成死信之后，它能被重新發(fā)送到另一個交換機(jī)中，這個交換機(jī)就是DLX，綁定此DLX的隊(duì)列就是死信隊(duì)列。

一個消息變成死信一般上是由于以下幾種情況;

1. 消息被拒絕

2. 消息過期

3. 隊(duì)列達(dá)到了最大長度。

所以，通過TTL和DLX的特性可以模擬實(shí)現(xiàn)延時隊(duì)列的功能。當(dāng)隊(duì)列中的消息超時成為死信后，會把消息死信重新發(fā)送到配置好的交換機(jī)中，然后分發(fā)到真實(shí)的消費(fèi)隊(duì)列。故簡單來說，我們可以創(chuàng)建2個隊(duì)列，一個隊(duì)列用于發(fā)送消息，一個隊(duì)列用于消息過期后的轉(zhuǎn)發(fā)的目標(biāo)隊(duì)列。

SpringBoot集成RabbitMQ實(shí)現(xiàn)延時隊(duì)列實(shí)戰(zhàn)

>以下使用SpringBoot集成RabbitMQ進(jìn)行實(shí)戰(zhàn)說明，在進(jìn)行http消息通知時，若通知失敗（地址不可用或者連接超時）時，將此消息轉(zhuǎn)入延時隊(duì)列中，待特定時間后進(jìn)行重新發(fā)送。

0.引入pom依賴

    
    
        org.springframework.boot
        spring-boot-starter-amqp
    
    
    
        cn.hutool
        hutool-http
        4.5.16
    
    
        cn.hutool
        hutool-json
        4.5.16
    
    
        org.springframework.boot
        spring-boot-starter-web
    

1.編寫rabbitmq配置文件（關(guān)鍵配置） RabbitConfig.java

/** 
*
* @ClassName   類名：RabbitConfig 
* @Description 功能說明：
* 

* TODO
*
************************************************************************
* @date        創(chuàng)建日期：2019年7月17日
* @version     版本號：V1.0
*

***************************修訂記錄*************************************
* 
*   2019年7月17日   oKong   創(chuàng)建該類功能。
*
***********************************************************************
*
*/
@Configuration
public class RabbitConfig {
    
    @Autowired
    ConnectionFactory connectionFactory;
    
    /**
     * 消費(fèi)者線程數(shù) 設(shè)置大點(diǎn) 大概率是能通知到的
     */
    @Value("${http.notify.concurrency:50}")
    int concurrency;
    
    /**
     * 延遲隊(duì)列的消費(fèi)者線程數(shù) 可設(shè)置小點(diǎn)
     */
    @Value("${http.notify.delay.concurrency:20}")
    int delayConcurrency;
    
    @Bean
    public RabbitAdmin rabbitAdmin() {
        return new RabbitAdmin(connectionFactory);
    }
    
    @Bean
    public DirectExchange httpMessageNotifyDirectExchange(RabbitAdmin rabbitAdmin) {
        //durable 是否持久化
        //autoDelete 是否自動刪除，即服務(wù)端或者客服端下線后 交換機(jī)自動刪除
        DirectExchange directExchange = new DirectExchange(ApplicationConstant.HTTP_MESSAGE_EXCHANGE,true,false);
        directExchange.setAdminsThatShouldDeclare(rabbitAdmin);
        return directExchange;
    }
    
    //設(shè)置消息隊(duì)列
    @Bean
    public Queue httpMessageStartQueue(RabbitAdmin rabbitAdmin) {
         
       /*
                       創(chuàng)建接收隊(duì)列，4個參數(shù)
         name - 隊(duì)列名稱
         durable - false，不進(jìn)行持有化
         exclusive - true，獨(dú)占性
         autoDelete - true，自動刪除*/
        Queue queue = new Queue(ApplicationConstant.HTTP_MESSAGE_START_QUEUE_NAME, true, false, false);
        queue.setAdminsThatShouldDeclare(rabbitAdmin);
        return queue;
    }
    
    //隊(duì)列綁定交換機(jī)
    @Bean
    public Binding bindingStartQuene(RabbitAdmin rabbitAdmin,DirectExchange httpMessageNotifyDirectExchange, Queue httpMessageStartQueue) {
        Binding binding = BindingBuilder.bind(httpMessageStartQueue).to(httpMessageNotifyDirectExchange).with(ApplicationConstant.HTTP_MESSAGE_START_RK);
        binding.setAdminsThatShouldDeclare(rabbitAdmin);
        return binding;
    }
    
    @Bean
    public Queue httpMessageOneQueue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME, true, false, false);
        queue.setAdminsThatShouldDeclare(rabbitAdmin);
        return queue;
    }
    
    @Bean
    public Binding bindingOneQuene(RabbitAdmin rabbitAdmin,DirectExchange httpMessageNotifyDirectExchange, Queue httpMessageOneQueue) {
        Binding binding = BindingBuilder.bind(httpMessageOneQueue).to(httpMessageNotifyDirectExchange).with(ApplicationConstant.HTTP_MESSAGE_ONE_RK);
        binding.setAdminsThatShouldDeclare(rabbitAdmin);
        return binding;
    }
    
    //-------------設(shè)置延遲隊(duì)列--開始--------------------
    @Bean
    public Queue httpDelayOneQueue() {
        //name - 隊(duì)列名稱
        //durable - true
        //exclusive - false
        //autoDelete - false
        return QueueBuilder.durable("http.message.dlx.one")
                //以下是重點(diǎn)：當(dāng)變成死信隊(duì)列時，會轉(zhuǎn)發(fā)至 路由為x-dead-letter-exchange及x-dead-letter-routing-key的隊(duì)列中
                .withArgument("x-dead-letter-exchange", ApplicationConstant.HTTP_MESSAGE_EXCHANGE)
                .withArgument("x-dead-letter-routing-key", ApplicationConstant.HTTP_MESSAGE_ONE_RK)
                .withArgument("x-message-ttl", 1*60*1000)//1分鐘 過期時間（單位：毫秒），當(dāng)過期后 會變成死信隊(duì)列，之后進(jìn)行轉(zhuǎn)發(fā)
                .build();
    }
    //綁定到交換機(jī)上
    @Bean
    public Binding bindingDelayOneQuene(RabbitAdmin rabbitAdmin, DirectExchange httpMessageNotifyDirectExchange, Queue httpDelayOneQueue) {
        Binding binding = BindingBuilder.bind(httpDelayOneQueue).to(httpMessageNotifyDirectExchange).with("delay.one");
        binding.setAdminsThatShouldDeclare(rabbitAdmin);
        return binding;
    }
    //-------------設(shè)置延遲隊(duì)列--結(jié)束--------------------

    //建議將正常的隊(duì)列和延遲處理的隊(duì)列分開
    //設(shè)置監(jiān)聽容器
    @Bean("notifyListenerContainer")
    public SimpleRabbitListenerContainerFactory httpNotifyListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手動ack
        factory.setConnectionFactory(connectionFactory);
        factory.setPrefetchCount(1);
        factory.setConcurrentConsumers(concurrency);
        return factory;
    }

    // 設(shè)置監(jiān)聽容器
    @Bean("delayNotifyListenerContainer")
    public SimpleRabbitListenerContainerFactory httpDelayNotifyListenerContainer() {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手動ack
        factory.setConnectionFactory(connectionFactory);
        factory.setPrefetchCount(1);
        factory.setConcurrentConsumers(delayConcurrency);
        return factory;
    }
}

ApplicationConstant.java

public class ApplicationConstant {
    
    /**
     * 發(fā)送http通知的 exchange 隊(duì)列
     */
    public static final String HTTP_MESSAGE_EXCHANGE = "http.message.exchange";
    
    /**
     * 配置消息隊(duì)列和路由key值
     */
    public static final String HTTP_MESSAGE_START_QUEUE_NAME = "http.message.start";
    public static final String HTTP_MESSAGE_START_RK = "rk.start";
    
    public static final String HTTP_MESSAGE_ONE_QUEUE_NAME = "http.message.one";
    public static final String HTTP_MESSAGE_ONE_RK = "rk.one";
    
    /**
     * 通知隊(duì)列對應(yīng)的延遲隊(duì)列關(guān)系，即過期隊(duì)列之后發(fā)送到下一個的隊(duì)列信息,可以根據(jù)實(shí)際情況添加，當(dāng)然也可以根據(jù)一定規(guī)則自動生成
     */
    public static final Map delayRefMap = new HashMap() {
        /**
         * 
         */
        private static final long serialVersionUID = -779823216035682493L;

        {
            put(HTTP_MESSAGE_START_QUEUE_NAME, "delay.one");
        }
    };    
}

簡單來說，就是創(chuàng)建一個正常消息發(fā)送隊(duì)列，用于接收http消息請求的參數(shù)，同時進(jìn)行http請求。同時，創(chuàng)建一個延時隊(duì)列，設(shè)置其x-dead-letter-exchange、x-dead-letter-routing-key和x-message-ttl值，將其轉(zhuǎn)發(fā)到正常的隊(duì)列中。使用一個map對象維護(hù)一個關(guān)系，當(dāng)正常消息異常時，需要發(fā)送的延時隊(duì)列的隊(duì)列名稱，當(dāng)然時間場景匯總，根據(jù)需要可以進(jìn)行動態(tài)配置或者根據(jù)一定規(guī)則進(jìn)行動態(tài)映射。

2.創(chuàng)建監(jiān)聽類，用于消息的消費(fèi)操作，此處使用@RabbitListener來消費(fèi)消息（當(dāng)然也可以使用SimpleMessageListenerContainer進(jìn)行消息配置的），創(chuàng)建了一個正常消息監(jiān)聽和延時隊(duì)列監(jiān)聽，由于一般上異常通知是低概率事件，可根據(jù)不同的監(jiān)聽容器進(jìn)行差異化配置。

/** 
*
* @ClassName   類名：HttpMessagerLister 
* @Description 功能說明：http通知消費(fèi)監(jiān)聽接口
* 

* TODO
*
************************************************************************
* @date        創(chuàng)建日期：2019年7月17日
* @version     版本號：V1.0
*

***************************修訂記錄*************************************
* 
*   2019年7月17日   oKong   創(chuàng)建該類功能。
*
***********************************************************************
*
*/
@Component
@Slf4j
public class HttpMessagerLister {
    
    @Autowired
    HttpMessagerService messagerService;
    
    @RabbitListener(id = "httpMessageNotifyConsumer", queues = {ApplicationConstant.HTTP_MESSAGE_START_QUEUE_NAME}, containerFactory = "notifyListenerContainer")
    public void httpMessageNotifyConsumer(Message message, Channel channel) throws Exception {
        doHandler(message, channel);
    }
    
    @RabbitListener(id= "httpDelayMessageNotifyConsumer", queues = {
            ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME,}, containerFactory = "delayNotifyListenerContainer")
    public void httpDelayMessageNotifyConsumer(Message message, Channel channel) throws Exception {
        doHandler(message, channel);
    }
    
    private void doHandler(Message message, Channel channel) throws Exception {
        String body = new String(message.getBody(),"utf-8");
        String queue = message.getMessageProperties().getConsumerQueue();
        log.info("接收到通知請求：{}，隊(duì)列名：{}",body, queue);
        //消息對象轉(zhuǎn)換
        try {
            HttpEntity httpNotifyDto = JSONUtil.toBean(body, HttpEntity.class);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            //發(fā)送通知
            messagerService.notify(queue, httpNotifyDto);
        } catch(Exception e) {
            log.error(e.getMessage());
            //ack
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}

HttpMessagerService.java：消息真正處理的類，此類是關(guān)鍵，這里未進(jìn)行日志記錄，真實(shí)場景中，強(qiáng)烈建議進(jìn)行消息通知的日志存儲，防止日后信息的查看，同時也能通過發(fā)送狀態(tài)，在重試次數(shù)都失敗后，進(jìn)行定時再次發(fā)送功能，同時也有據(jù)可查。

@Component
@Slf4j
public class HttpMessagerService {
    
    @Autowired
    AmqpTemplate mqTemplate;    
    
    public void notify(String queue,HttpEntity httpEntity) {
        //發(fā)起請求
        log.info("開始發(fā)起http請求:{}", httpEntity);
        try {
            switch(httpEntity.getMethod().toLowerCase()) {
            case "POST":
                  HttpUtil.post(httpEntity.getUrl(), httpEntity.getParams());
                  break;
            case "GET":
            default:
                HttpUtil.get(httpEntity.getUrl(), httpEntity.getParams());
            }
        } catch (Exception e) {
            //發(fā)生異常，放入延遲隊(duì)列中
            String nextRk = ApplicationConstant.delayRefMap.get(queue);
            if(ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME.equals(queue)) {
                //若已經(jīng)是最后一個延遲隊(duì)列的消息隊(duì)列了，則后續(xù)可直接放入數(shù)據(jù)庫中 待后續(xù)定時策略進(jìn)行再次發(fā)送
                log.warn("http通知已經(jīng)通知N次失敗，進(jìn)入定時進(jìn)行發(fā)起通知,url={}", httpEntity.getUrl());
            } else {
               log.warn("http重新發(fā)送通知：{}, 通知隊(duì)列rk為：{}, 原隊(duì)列：{}", httpEntity.getUrl(), nextRk, queue);
               mqTemplate.convertAndSend(ApplicationConstant.HTTP_MESSAGE_EXCHANGE, nextRk, cn.hutool.json.JSONUtil.toJsonStr(httpEntity));
            }
        }
    }
}

3.創(chuàng)建控制層服務(wù)（真實(shí)場景中，如SpringCloud微服務(wù)中，一般上是創(chuàng)建個api接口，供其他服務(wù)進(jìn)行調(diào)用）

@Slf4j
@RestController
@Api(tags = "http測試接口")
public class HttpDemoController {

    @Autowired
    AmqpTemplate mqTemplate;
    
    @PostMapping("/send")
    @ApiOperation(value="send",notes = "發(fā)送http測試")
    public String sendHttp(@RequestBody HttpEntity httpEntity) {
        //發(fā)送http請求
        log.info("開始發(fā)起http請求，發(fā)布異步消息：{}", httpEntity);
        mqTemplate.convertAndSend(ApplicationConstant.HTTP_MESSAGE_EXCHANGE, ApplicationConstant.HTTP_MESSAGE_START_RK, cn.hutool.json.JSONUtil.toJsonStr(httpEntity));
        return "發(fā)送成功：url=" + httpEntity.getUrl();        
    }
}

4.配置文件添加RabbitMQ相關(guān)配置信息

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

# 通知-消費(fèi)者線程數(shù) 設(shè)置大點(diǎn) 大概率是能通知到的
http.notify.concurrency=150
# 延遲隊(duì)列的消費(fèi)者線程數(shù) 可設(shè)置小點(diǎn) 
http.notify.delay.concurrency=10

5.編寫啟動類。

@SpringBootApplication
@Slf4j
public class DelayQueueApplication {

    public static void main(String[] args) throws Exception {
        SpringApplication.run(DelayQueueApplication.class, args);
        log.info("spring-boot-rabbitmq-delay-queue-chapter38服務(wù)啟動!");
    }
}

6.啟動服務(wù)。使用swagger進(jìn)行簡單調(diào)用測試。

• 正常通知：

2019-07-20 23:52:23.792  INFO 65216 --- [nio-8080-exec-1] c.l.l.s.c.controller.HttpDemoController  : 開始發(fā)起http請求，發(fā)布異步消息：HttpEntity(url=www.baidu.com, params={a=1}, method=get)
2019-07-20 23:52:23.794  INFO 65216 --- [TaskExecutor-97] c.l.l.s.chapter38.mq.HttpMessagerLister  : 接收到通知請求：{"method":"get","params":{"a":1},"url":"www.baidu.com"}，隊(duì)列名：http.message.start
2019-07-20 23:52:23.794  INFO 65216 --- [TaskExecutor-97] c.l.l.s.c.service.HttpMessagerService    : 開始發(fā)起http請求:HttpEntity(url=www.baidu.com, params={a=1}, method=get)

• 異常通知：訪問一個不存在的地址

2019-07-20 23:53:14.699  INFO 65216 --- [nio-8080-exec-4] c.l.l.s.c.controller.HttpDemoController  : 開始發(fā)起http請求，發(fā)布異步消息：HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:53:14.705  INFO 65216 --- [TaskExecutor-84] c.l.l.s.chapter38.mq.HttpMessagerLister  : 接收到通知請求：{"method":"get","params":{"a":1},"url":"www.baidu.com1"}，隊(duì)列名：http.message.start
2019-07-20 23:53:14.705  INFO 65216 --- [TaskExecutor-84] c.l.l.s.c.service.HttpMessagerService    : 開始發(fā)起http請求:HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:53:14.706  WARN 65216 --- [TaskExecutor-84] c.l.l.s.c.service.HttpMessagerService    : http重新發(fā)送通知：www.baidu.com1, 通知隊(duì)列rk為：delay.one, 原隊(duì)列：http.message.start

在RabbitMQ后臺中，可以看見http.message.dlx.one隊(duì)列中存在這需要延時處理的消息，在一分鐘后會轉(zhuǎn)發(fā)至http.message.one隊(duì)列中。

在一分鐘后，可以看見消息本再次消費(fèi)了。

2019-07-20 23:54:14.722  INFO 65216 --- [TaskExecutor-16] c.l.l.s.chapter38.mq.HttpMessagerLister  : 接收到通知請求：{"method":"get","params":{"a":1},"url":"www.baidu.com1"}，隊(duì)列名：http.message.one
2019-07-20 23:54:14.723  INFO 65216 --- [TaskExecutor-16] c.l.l.s.c.service.HttpMessagerService    : 開始發(fā)起http請求:HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:54:14.723  WARN 65216 --- [TaskExecutor-16] c.l.l.s.c.service.HttpMessagerService    : http通知已經(jīng)通知N次失敗，進(jìn)入定時進(jìn)行發(fā)起通知,url=www.baidu.com1

一些最佳實(shí)踐

>在正式場景中，一般上補(bǔ)償或者重試機(jī)制大概率是不會發(fā)送的，倘若發(fā)生時，一般上是第三方業(yè)務(wù)系統(tǒng)出現(xiàn)了問題，故一般上在進(jìn)行補(bǔ)充時，應(yīng)該在非高峰期進(jìn)行操作，故應(yīng)該對延時監(jiān)聽器，應(yīng)該在高峰期時停止消費(fèi)，在非高峰期時進(jìn)行消費(fèi)。同時，還可以根據(jù)不同的通知類型，放入不一樣的延時隊(duì)列中，保障業(yè)務(wù)的正常。這里簡單說明下，動態(tài)停止或者啟動演示監(jiān)聽器的方式。一般上是使用RabbitListenerEndpointRegistry對象獲取延時監(jiān)聽器，之后進(jìn)行動態(tài)停止或者啟用。可設(shè)置@RabbitListener的id屬性，直接進(jìn)行獲取，當(dāng)然也可以直接獲取所有的監(jiān)聽器，進(jìn)行自定義判斷了。

    @Autowired
    RabbitListenerEndpointRegistry registry;
    
   @GetMapping("/set")
    @ApiOperation(value = "set", notes = "設(shè)置消息監(jiān)聽器的狀態(tài)")
    public String setSimpleMessageListenerContainer(String status) {
        if("1".equals(status)) {
            registry.getListenerContainer("httpDelayMessageNotifyConsumer").start();
        } else {
            registry.getListenerContainer("httpDelayMessageNotifyConsumer").stop();
        }
        return status;
    }

這里，只是簡單進(jìn)行演示說明，在真實(shí)場景下，可以使用定時器，判斷當(dāng)前是否為高峰期，進(jìn)而進(jìn)行動態(tài)設(shè)置監(jiān)聽器的狀態(tài)。

以上是“SpringBoot中如何基于RabbitMQ實(shí)現(xiàn)消息延遲隊(duì)列”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家都有了一定的了解，希望分享的內(nèi)容對大家有所幫助，如果還想學(xué)習(xí)更多知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道！