Rabbitmq之整合Springboot，普通队列以及死信队列demo实例，队列优化以及插件实现延迟队列

Posted 2022-05-23 爱上口袋的天空

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Rabbitmq之整合Springboot，普通队列以及死信队列demo实例，队列优化以及插件实现延迟队列相关的知识，希望对你有一定的参考价值。

1、引入pom.xml依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.kgf.rabbitmq</groupId>
    <artifactId>springboot-rabbitmq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <!--继承一个父模块，然后再引入相应的依赖  -->
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.12.RELEASE</version>
        <!--relativePath是可选的,maven会首先搜索这个地址,在搜索本地远程repositories之前  -->
        <relativePath />
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.73</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>

        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger2</artifactId>
            <version>2.9.2</version>
        </dependency>

        <dependency>
            <groupId>io.springfox</groupId>
            <artifactId>springfox-swagger-ui</artifactId>
            <version>2.9.2</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-jar-plugin</artifactId>
                <version>3.2.0</version>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <version>2.2.11.RELEASE</version>
            </plugin>
            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

2、配置文件信息

spring:
  application:
    name: springboot-rabbit-mq
  rabbitmq:
    host: 192.168.56.20
    port: 5672
    username: admin
    password: 123456

server:
  port: 8080

3、添加Swagger配置类

package com.kgf.rabbitmq.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import springfox.documentation.builders.ApiInfoBuilder;
import springfox.documentation.service.ApiInfo;
import springfox.documentation.service.Contact;
import springfox.documentation.spi.DocumentationType;
import springfox.documentation.spring.web.plugins.Docket;
import springfox.documentation.swagger2.annotations.EnableSwagger2;

@Configuration
 @EnableSwagger2
 public class SwaggerConfig 
     @Bean
     public Docket webApiConfig()
         return new Docket(DocumentationType.SWAGGER_2)
                 .groupName("webApi")
                 .apiInfo(webApiInfo())
                 .select()
                 .build();
     

 private ApiInfo webApiInfo()
     return new ApiInfoBuilder()
             .title("rabbitmq接口文档")
             .description("本文档描述了rabbitmq微服务接口定义")
             .version("1.0")
             .contact(new Contact("enjoy6288","http://test.com","123456@qq.com"))
             .build();

4、队列实现

4.1、代码架构

创建两个队列QA和QB，两者队列TTL分别设置为10S和40S，然后在创建一个交换机X和死信交换机Y，它们的类型都是direct，创建一个死信队列QD，它们的绑定关系如下：

5、代码实现

5.1、配置类代码

package com.kgf.rabbitmq.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;
import java.util.Map;

/*
* 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_LATTER_QUEUE = "QD";

    //声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange()
        return new DirectExchange(X_EXCHANGE);
    

    //声明yExchange
    @Bean("yExchange")
    public DirectExchange yExchange()
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    

    //声明队列
    @Bean("queueA")
    public Queue queueA()
        Map<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //设置死信Routing-key
        arguments.put("x-dead-letter-routing-key","YD");
        //设置TTL 单位是ms
        arguments.put("x-message-ttl",10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    

    //声明普通队列 TTL为40s
    @Bean("queueB")
    public Queue queueB()
        Map<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //设置死信Routing-key
        arguments.put("x-dead-letter-routing-key","YD");
        //设置TTL 单位是ms
        arguments.put("x-message-ttl",40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    

    //死信队列
    @Bean("queueD")
    public Queue queueD()
        return QueueBuilder.durable(DEAD_LATTER_QUEUE).build();
    

    //绑定
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange)
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    

    //绑定
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange)
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    

    //绑定
    @Bean
    public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange)
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");

5.2、消费者代码

package com.kgf.rabbitmq.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 receiveD(Message message, Channel channel) throws Exception 
         String msg = new String(message.getBody());
         log.info("当前时间：，收到死信队列的消息：",new Date().toString(),msg);

5.3、生产者代码

package com.kgf.rabbitmq.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;

/*
 * 发送延迟消息
 * */
 @Slf4j
 @RestController
 @RequestMapping("/ttl")
 public class SendMsgController 
 
     @Autowired
     private RabbitTemplate rabbitTemplate;
 
     //开始发消息
     @GetMapping("/sendMsg/message")
     public void sendMsg(@PathVariable String message)
       log.info("当前时间：，发送一条信息给两个TTL队列：",new Date().toString(),message);
 
       rabbitTemplate.convertAndSend("X","XA","消息来自TTL为10s的队列：" + message);
       rabbitTemplate.convertAndSend("X","XB","消息来自TTL为40s的队列：" + message);

5.4、效果

6、队列优化

6.1、问题

第一条消息在10S后变成了死信消息，然后被消费者消费掉，第二条消息在40S之后变成了死信消息，然后被消费掉，这样一个延时队列就打造完成了。

不过，如果这样使用的话，岂不是每增加一个新的时间需求，就要新增一个队列，这里只有10S和40S两个时间选项，如果需要一个小时后处理，那么就需要增加TTL为一个小时的队列，如果是预定会议室然后提前通知这样的场景，岂不是要增加无数个队列才能满足需求？

6.2、代码架构图

6.3、实现

6.3.1、配置文件类

package com.kgf.rabbitmq.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;
import java.util.Map;

/*
* 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 QUEUE_C = "QC";
    //死信队列的名称
    public static final String DEAD_LATTER_QUEUE = "QD";

    //声明QC队列
    @Bean("queueC")
    public Queue queueC()
        Map<String, Object> arguments = new HashMap<>();
        //设置死信交换机
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","YD");
        return QueueBuilder.durable().withArguments(arguments).build();
    
    @Bean
    public Binding queueCBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange)
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    

    //声明xExchange
    @Bean("xExchange")
    public DirectExchange xExchange()
        return new DirectExchange(X_EXCHANGE);
    

    //声明yExchange
    @Bean("yExchange")
    public DirectExchange yExchange()
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    

    //声明队列
    @Bean("queueA")
    public Queue queueA()
        Map<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //设置死信Routing-key
        arguments.put("x-dead-letter-routing-key","YD");
        //设置TTL 单位是ms
        arguments.put("x-message-ttl",10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(arguments).build();
    

    //声明普通队列 TTL为40s
    @Bean("queueB")
    public Queue queueB()
        Map<String, Object> arguments = new HashMap<>(3);
        //设置死信交换机
        arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_EXCHANGE);
        //设置死信Routing-key
        arguments.put("x-dead-letter-routing-key","YD");
        //设置TTL 单位是ms
        arguments.put("x-message-ttl",40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
    

    //死信队列
    @Bean("queueD")
    public Queue queueD()
        return QueueBuilder.durable(DEAD_LATTER_QUEUE).build();
    

    //绑定
    @Bean
    public Binding queueABindingX(@Qualifier("queueA") Queue queueA,
                                  @Qualifier("xExchange") DirectExchange xExchange)
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    

    //绑定
    @Bean
    public Binding queueBBindingX(@Qualifier("queueB") Queue queueB,
                                  @Qualifier("xExchange") DirectExchange xExchange)
        return BindingBuilder.bind(queueB).to(xExchange).with("XB");
    

    //绑定
    @Bean
    public Binding queueDBindingX(@Qualifier("queueD") Queue queueD,
                                  @Qualifier("yExchange") DirectExchange yExchange)
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");

6.3.2、生产者

package com.kgf.rabbitmq.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;

/*
 * 发送延迟消息
 * */
 @Slf4j
 @RestController
 @RequestMapping("/ttl")
 public class SendMsgController 
 
     @Autowired
     private RabbitTemplate rabbitTemplate;
 
     //开始发消息
     @GetMapping("/sendMsg/message")
     public void sendMsg(@PathVariable String message)
       log.info("当前时间：，发送一条信息给两个TTL队列：",new Date().toString(),message);
 
       rabbitTemplate.convertAndSend("X","XA","消息来自TTL为10s的队列：" + message);
       rabbitTemplate.convertAndSend("X","XB","消息来自TTL为40s的队列：" + message);
 
     

    //开始发消息
    @GetMapping("sendExpirationMsg/message/ttlTime")
    public void sendMsg(@PathVariable String message,@PathVariable String ttlTime)
        log.info("当前时间：，发送一条时长毫秒TTL信息给队列QC：",
                new Date().toString(),ttlTime,message);
        rabbitTemplate.convertAndSend("X","XC",message,msg->
            //发送消息的时候 延迟时长
            msg.getMessageProperties().setExpiration(ttlTime);
            return msg;
        );

6.3.3、消费者代码无需改变

6.3.4、效果及问题：按理来说应该是你好2先收到

因为RabbitMQ智慧检查第一个消息是否过期，如果过期则丢到死信队列，如果第一个消息的延时时长很长，而第二个消息的延时时长很短，第二个消息并不会优先得到执行！

7、插件实现延迟队列

7.1、下载延迟插件

https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.8.9/rabbitmq_delayed_message_exchange-3.8.9-0199d11c.ez

7.2、将延迟插件放到RabbitMQ的插件目录下

7.3、安装插件并重启服务

 rabbitmq-plugins enable rabbitmq_delayed_message_exchange
 systemctl restart rabbitmq-server

7.4、代码架构图

7.4.1、配置文件类

package com.kgf.rabbitmq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
 public class DelayedQueueConfig 

 //队列
 public static final String DELAYED_QUEUE_NAME = "delayed.queue";
 //交换机
 public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";
 //routingKey
 public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";

 //声明队列
 @Bean
 public Queue delayedQueue()
   return new Queue(DELAYED_QUEUE_NAME);
 ;



 //声明交换机
 @Bean
 public CustomExchange delayedExchange()

     Map<String, Object> arguments = new HashMap<>();
     arguments.put("x-delayed-type","direct");

     return new CustomExchange(DELAYED_EXCHANGE_NAME,"x-delayed-message",
             true,false,arguments);
 
 //绑定
 @Bean
 public Binding delayedQueueBindingDelayedExchange(@Qualifier("delayedQueue") Queue delayedQueue,
                                                   @Qualifier("delayedExchange") CustomExchange delayedExchange)
     return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();

7.4.2、消费者

package com.kgf.rabbitmq.consumer;

import com.kgf.rabbitmq.config.DelayedQueueConfig;
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;

// 消费者代码 基于插件的延迟消息
 @Slf4j
 @Component
 public class DelayQueueConsumer 

 //监听消息
 @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)
 public void recieveDelayQueue(Message message)
     String msg = new String(message.getBody());
     log.info("当前时间：，收到延迟队列的消息：",new Date().toString(),msg);

7.4.3、生产者

 /*
 * 发送延迟消息
 * */
 @Slf4j
 @RestController
 @RequestMapping("/ttl")
 public class SendMsgController 
 
     @Autowired
     private RabbitTemplate rabbitTemplate;
 
     //开始发消息 基于插件的 消息 及 延迟的时间
     @GetMapping("/sendDelayMsg/message/delayTime")
     public void sendMsg(@PathVariable String message,@PathVariable Integer delayTime)
         log.info("当前时间：，发送一条时长毫秒信息给延迟队列delayed.queue：",
                 new Date().toString(),delayTime,message);
         rabbitTemplate.convertAndSend(DelayedQueueConfig.DELAYED_EXCHANGE_NAME
                 ,DelayedQueueConfig.DELAYED_ROUTING_KEY,message,msg -> 
             // 发送消息的时候 延迟时长 单位ms
             msg.getMessageProperties().setDelay(delayTime);
             return msg;
                 );

7.4.4、效果

访问：

http://localhost:8080/ttl/sendDelayMsg/come on baby1/20000,

http://localhost:8080/ttl/sendDelayMsg/come on baby2/2000

8、总结

延时队列在需要延时处理的场景下非常有用，使用RabbitMQ来实现延时队列可以很好的利用
RabbitMQ.的特性，如:消息可靠发送、消息可靠投递、死信队列来保障消息至少被消费一次以及未被正确处理的消息不会被丢弃。另外，通过RabbitMQ集群的特性，可以很好的解决单点故障问题，不会因为单个节点挂掉导致延时队列不可用或者消息丢失。
当然，延时队列还有很多其它选择，比如利用Java的DelayQueue，利用Redis.的zsset，利用Quartz或者利用kafka的时间轮，这些方式各有特点,看需要适用的场景

以上是关于Rabbitmq之整合Springboot，普通队列以及死信队列demo实例，队列优化以及插件实现延迟队列的主要内容，如果未能解决你的问题，请参考以下文章

springboot整合rabbitmq

SpringBoot整合RabbitMQ之发送接收消息实战

RabbitMQ笔记SpringBoot整合RabbitMQ之simple容器（生产者）

SpringBoot整合RabbitMQ之整合配置篇

SpringBoot2.0之整合RabbitMQ

SpringBoot整合RabbitMQ之Spring事件驱动模型