快捷導(dǎo)航

springboot整合RabbitMQ 中的 TTL實(shí)例代碼

更新時(shí)間：2022年09月28日 15:23:24 作者：笑霸final

TTL 是 RabbitMQ 中一個(gè)消息或者隊(duì)列的屬性，表明一條消息或者該隊(duì)列中的所有消息的最大存活時(shí)間，單位是毫秒,這篇文章主要介紹了springboot整合RabbitMQ 中的 TTL,需要的朋友可以參考下

TTL簡(jiǎn)介

TTL 是什么呢？TTL 是 RabbitMQ 中一個(gè)消息或者隊(duì)列的屬性，表明一條消息或者該隊(duì)列中的所有消息的最大存活時(shí)間，單位是毫秒。換句話說(shuō)，如果一條消息設(shè)置了 TTL 屬性或者進(jìn)入了設(shè)置 TTL 屬性的隊(duì)列，那么這條消息如果在 TTL 設(shè)置的時(shí)間內(nèi)沒(méi)有被消費(fèi)，則會(huì)成為"死信"。
下面就根據(jù)這個(gè)圖片來(lái)驗(yàn)證代碼

配置類代碼

這里寫(xiě)一些配置，比如創(chuàng)建隊(duì)列交換機(jī) 和它們之間的綁定關(guān)系

@Qualifier 注解與我們想要使用的特定 Spring bean 的名稱一起進(jìn)行裝配，Spring 框架就能從多個(gè)相同類型并滿足裝配要求的 bean 中找到我們想要的，避免讓Spring腦裂。我們需要做的是@Component或者@Bean注解中聲明的value屬性以確定名稱

注意包別導(dǎo)錯(cuò)了

package com.xbfinal.springbootrabbitmq.config;


import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;

/**
 * ttl隊(duì)列 配置文件類
 *
 */
@Configuration
public class TtlQueueConfig {

    //普通交換機(jī)名稱
    public static final String X_EXCHANGE="X";
    //死信交換機(jī)名稱
    public static final String Y_DEAD_LETTER_EXCHANGE="Y";
    //普通隊(duì)列名稱
    public static final String QUEUE_A="QA";
    public static final String QUEUE_B="QB";
    //死信隊(duì)列名稱
    public static final String DEAD_LETTER_QUEUE_D="QD";

    /**
     * 聲明x交換機(jī)
     * @return
     */
    @Bean("xExchange")//別名和方法名取一樣
    public DirectExchange xExchange(){
        return new DirectExchange(X_EXCHANGE);
    }

    /**
     * 聲明y交換機(jī)
     * @return
     */
    @Bean("yExchange")//別名和方法名取一樣
    public DirectExchange yExchange(){
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }

    //聲明隊(duì)列A
    @Bean("queueA")
    public Queue queueA(){
        final HashMap<String, Object> arguments
                = new HashMap<>();
        //設(shè)置死信交換機(jī)
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //設(shè)置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","YD");
        //設(shè)置TTL設(shè)置10秒過(guò)期
        arguments.put("x-message-ttl",10000);

        return QueueBuilder.durable(QUEUE_A)
                .withArguments(arguments)
                .build();
    }

    //聲明隊(duì)列B
    @Bean("queueB")
    public Queue queueB(){
         HashMap<String, Object> arguments
                = new HashMap<>();
        //設(shè)置死信交換機(jī)
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //設(shè)置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","YD");
        //設(shè)置TTL設(shè)置40秒過(guò)期
        arguments.put("x-message-ttl",40000);

        return QueueBuilder.durable(QUEUE_B)
                .withArguments(arguments)
                .build();
    }

    @Bean("queueD")
    public Queue queueD(){
        return QueueBuilder.durable(DEAD_LETTER_QUEUE_D)
                .build();
    }

    /**
     * A隊(duì)列綁定X交換機(jī)
     * @param queueA
     * @return
     */
    @Bean
    public Binding queueABindingX(@Qualifier("queueA")Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    /**
     * B隊(duì)列綁定X交換機(jī)
     * @param queueB
     * @param xExchange
     * @return
     */
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB")Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    }

    /**
     * D隊(duì)列綁定死信y交換機(jī)
     * @param queueD
     * @param yExchange
     * @return
     */
    @Bean
    public Binding queueDBindingX(@Qualifier("queueD")Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange){
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }

}

生產(chǎn)者代碼

我們用Controller寫(xiě)，通過(guò)網(wǎng)頁(yè)提交的方式生產(chǎn)消息
url：http://localhost:8080/ttl/sendMsg/message

package com.xbfinal.springbootrabbitmq.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;

/**
 * 生產(chǎn)者
 * 發(fā)送延遲消息
 */
@Slf4j
@RestController
@RequestMapping("/ttl")
public class SendMsgController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMsg/{message}")
    public void sendMsg(@PathVariable String message){
        log.info("當(dāng)前時(shí)間:{},發(fā)送了一條消息({})給兩個(gè)隊(duì)列", new Date().toString(),message);
        //發(fā)送消息
        rabbitTemplate.convertAndSend("X","XA","10秒"+message);
        rabbitTemplate.convertAndSend("X","XB","40秒"+message);
    }
}

消息消費(fèi)者代碼

注意@RabbitListener注解
@RabbitListener注解指定目標(biāo)方法來(lái)作為消費(fèi)消息的方法，通過(guò)注解參數(shù)指定所監(jiān)聽(tīng)的隊(duì)列或者Binding。使用@RabbitListener可以設(shè)置一個(gè)自己明確默認(rèn)值的RabbitListenerContainerFactory對(duì)象。

@RabbitListener標(biāo)注在方法上，直接監(jiān)聽(tīng)指定的隊(duì)列，此時(shí)接收的參數(shù)需要與發(fā)送市類型一致
3.@RabbitListener 可以標(biāo)注在類上面，需配合 @RabbitHandler 注解一起使用

@RabbitListener 標(biāo)注在類上面表示當(dāng)有收到消息的時(shí)候，就交給 @RabbitHandler 的方法處理，根據(jù)接受的參數(shù)類型進(jìn)入具體的方法中。

package com.xbfinal.springbootrabbitmq.consumer;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.util.Date;

/**
 * 隊(duì)列TTL的消費(fèi)者
 */
@Slf4j
@Component
public class DeadLetterQueueConsumer {

    //接收消息
    @RabbitListener(queues = "QD")
    public void receivedD(Message message, Channel channel)throws Exception{
        String msg=new String(message.getBody());
        log.info("當(dāng)前時(shí)間：{}，收到消息：{}",new Date().toString(),msg);
    }
}