快捷導(dǎo)航

springboot整合RabbitMQ 中的 TTL實例代碼

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

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

TTL簡介

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

配置類代碼

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

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

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

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隊列 配置文件類
 *
 */
@Configuration
public class TtlQueueConfig {

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

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

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

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

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

    //聲明隊列B
    @Bean("queueB")
    public Queue queueB(){
         HashMap<String, Object> arguments
                = new HashMap<>();
        //設(shè)置死信交換機
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //設(shè)置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","YD");
        //設(shè)置TTL設(shè)置40秒過期
        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隊列綁定X交換機
     * @param queueA
     * @return
     */
    @Bean
    public Binding queueABindingX(@Qualifier("queueA")Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    /**
     * B隊列綁定X交換機
     * @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隊列綁定死信y交換機
     * @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寫，通過網(wǎng)頁提交的方式生產(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)前時間:{},發(fā)送了一條消息({})給兩個隊列", new Date().toString(),message);
        //發(fā)送消息
        rabbitTemplate.convertAndSend("X","XA","10秒"+message);
        rabbitTemplate.convertAndSend("X","XB","40秒"+message);
    }
}

消息消費者代碼

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

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

@RabbitListener 標(biāo)注在類上面表示當(dāng)有收到消息的時候，就交給 @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;

/**
 * 隊列TTL的消費者
 */
@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)前時間：{}，收到消息：{}",new Date().toString(),msg);
    }
}