程序人生网站首页 程序人生

尚硅谷RabbitMQ视频-029-进阶篇-消息超时

发布时间：2026-11-08 02:34:32编辑：雪饮阅读（）

相关说明

给消息设定一个过期时间，超过这个时间没有被取走的消息就会被删除
我们可以从两个层面来给消息设定过期时间：
- 队列层面：在队列层面设定消息的过期时间，并不是队列的过期时间。意思是这个队列中的消息全部使用同一个过期时间。
- 消息本身：给具体的某个消息设定过期时间

直接创建一个交换机exchange.test.timeout

再增加一个队列queue.test.timeout，且5000毫秒超时（Arguments中新增x-message-ttl参数，参数值类型是number，参数值为5000）

进入队列queue.test.timeout为队列queue.test.timeout绑定交换机exchange.test.timeout

From exchange：exchange.test.timeout

Routing key：routing.key.test.timeout

只为了测试队列超时，所以我们直接不写消费端，那就是写生产端咯

仍旧用新模块来写

D:\rabbitMqLearning \timeout\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">

    <parent>

        <artifactId>rabbitMqLearning</artifactId>

        <groupId>org.example</groupId>

        <version>1.0-SNAPSHOT</version>

    </parent>

    <modelVersion>4.0.0</modelVersion>



    <artifactId>timeout</artifactId>



    <properties>

        <maven.compiler.source>17</maven.compiler.source>

        <maven.compiler.target>17</maven.compiler.target>

    </properties>





    <!--普通项目转换为spring boot项目-->

    <dependencyManagement>

        <dependencies>

            <dependency>

                <groupId>org.springframework.boot</groupId>

                <artifactId>spring-boot-dependencies</artifactId>

                <version>3.1.5</version>

                <type>pom</type>

                <scope>import</scope>

            </dependency>

        </dependencies>

    </dependencyManagement>






    <dependencies>

        <dependency>

            <groupId>com.rabbitmq</groupId>

            <artifactId>amqp-client</artifactId>

            <version>5.16.0</version>

        </dependency>



        <!-- spring-boot rabbitmq amqp 依赖 -->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-amqp</artifactId>

        </dependency>

        <!-- spring-boot web依赖 -->

        <!-- (注意默认会占用本地8080端口，若已经被占用，则可能无法启动) -->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-web</artifactId>

        </dependency>

        <!-- spring-boot lombok -->

        <dependency>

            <groupId>org.projectlombok</groupId>

            <artifactId>lombok</artifactId>

        </dependency>






        <!--用测试依赖来实现实现生产者-->

        <dependency>

            <groupId>org.springframework.boot</groupId>

            <artifactId>spring-boot-starter-test</artifactId>

        </dependency>

    </dependencies>

</project>

D:\rabbitMqLearning\timeout\src\main\resources\application.yml:

spring:

  rabbitmq:

    host: localhost

    port: 5672

    username: guest

    password: 123456

    virtual-host: /

    listener:

      simple:

        acknowledge-mode: manual # 把消息确认模式改为手动确认,因为默认都是按成功处理，默认都是ack

        prefetch: 1 # 限流闸门：每个消费者最多同时持有 1 条未确认(unacked)消息；处理完并手动 ack 后，broker 才推下一条，防止消息洪流瞬间压垮消费者

    publisher-confirm-type: correlated #交换机的确认

    publisher-returns: true #队列的确认

D:\rabbitMqLearning\timeout\src\test\java\org\example\RabbitMQTest.java:

package org.example;



import org.junit.jupiter.api.Test;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;



@SpringBootTest

public class RabbitMQTest {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";

    public static final String ROUTING_KEY = "order";



    public static final String EXCHANGE_DIRECT_TIMEOUT = "exchange.test.timeout";

    public static final String ROUTING_KEY_TIMEOUT = "routing.key.test.timeout";



    @Autowired

    private RabbitTemplate rabbitTemplate;



    @Test

    public void test03SendMessage() {

        for (int i = 0; i < 100; i++) {

            rabbitTemplate.convertAndSend(EXCHANGE_DIRECT_TIMEOUT, ROUTING_KEY_TIMEOUT, "Test Timeout " + i);

        }

    }

}

D:\rabbitMqLearning\timeout\src\main\java\org\example\listener\MyMessageListener.java:

package org.example.listener;



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.io.IOException;

import java.util.concurrent.TimeUnit;



@Component

@Slf4j

public class MyMessageListener {

    public static final String QUEUE_NAME = "queue.order";



    //@RabbitListener(queues = {QUEUE_NAME})

    public void processMessage(String dataString, Message message, Channel channel) throws IOException {

        long deliveryTag=message.getMessageProperties().getDeliveryTag();

        try {

            // 核心操作 处理消息

            log.info("org.example.confirmConsumer.listener 消费端 消息内容: " + dataString);





            /*

            核心操作成功: 返回 ACK 信息，

            multiple参数为：true为批量操作，false为非批量操作,

            basicAck作用：**成功确认**，Broker 收到 ack 之后，直接删除这条消息

            */

            channel.basicAck(deliveryTag, false);

        } catch (Exception e) {





            /*

            获取当前消息是否是重复投递的

            redelivered 为 true：说明当前消息已经重复投递过一次了

            redelivered 为 false：说明当前消息是第一次投递

            */

            Boolean redelivered = message.getMessageProperties().getRedelivered();



            /*

            核心操作失败: 返回 NACK 信息

            requeue 参数: 控制消息是否重新放回队列

            取值为 true: 重新放回队列, broker 会重新投递这个消息

            取值为 false: 不重新放回队列, broker 会丢弃这个消息

            basicNack作用：**失败拒绝**

            */



            //已是重投消息，则拒绝重投

            if(redelivered){

                channel.basicNack(deliveryTag, false, false);

            }

            //非重投消息，则重新投递

            if(!redelivered){

                channel.basicNack(deliveryTag, false, true);

            }

        }

    }

    @RabbitListener(queues = {QUEUE_NAME})

    public void processMessageTestPrefetch(String dataString, Message message, Channel channel) throws IOException, InterruptedException {

        log.info("消费端 消息内容: " + dataString);



        //这里是模拟实际业务逻辑，因为实际业务逻辑处理一般情况是会比较慢一点

        TimeUnit.SECONDS.sleep(1);



        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

    }



}

D:\rabbitMqLearning\timeout\src\main\java\org\example\RabbitMQConsumerMainType.java:

package org.example;



import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;



@SpringBootApplication

public class RabbitMQConsumerMainType {

    public static void main(String[] args) {

        SpringApplication.run(RabbitMQConsumerMainType.class, args);

    }

}

然后运行test03SendMessage产生消息挤压到队列，因为目前没有消费者

然后运行后在rabbitmq的管理端进入queue.test.timeout可以看到ready和total大概5秒后消息没有处理则自动消失了，变成0了

刚才并不是我们的代码生效了，而是刚才建立队列的时候队列的超时处理消息的时间是5000毫秒。

所以接下来我们删除队列并重建同名队列，但不填写在处理队列内部消息的超时时间，然后重新绑定的原来的exchange.test.timeout交换机。

那么其实刚才的5秒，在配置低的电脑上，会发现还是很快删除，为了更明显，你可以删除哪个队列重新配置队列内消息处理超时未10秒，你再次运行的时候就会停留更久一些，那么此时你也可以手动删除消息

在队列queue.test.timeout里面的Get messages中ack mode中下拉选择选中Automatic ack，然后点击下面的Get Message(s)就可以主动ack一条消息，也即可以理解为删除一条消息。

那么接下来我们为生产者生产消息这里新增传递后置处理器

D:\rabbitMqLearning\timeout\src\test\java\org\example\RabbitMQTest.java:

package org.example;



import org.junit.jupiter.api.Test;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.amqp.core.MessagePostProcessor;



@SpringBootTest

public class RabbitMQTest {

    public static final String EXCHANGE_DIRECT = "exchange.direct.order";

    public static final String ROUTING_KEY = "order";



    public static final String EXCHANGE_DIRECT_TIMEOUT = "exchange.test.timeout";

    public static final String ROUTING_KEY_TIMEOUT = "routing.key.test.timeout";



    @Autowired

    private RabbitTemplate rabbitTemplate;



    @Test

    public void test03SendMessage() {

        // 创建消息后置处理器对象

        MessagePostProcessor postProcessor = message -> {

            // 设置消息的过期时间，单位是毫秒

            message.getMessageProperties().setExpiration("7000");

            return message;

        };

        for (int i = 0; i < 100; i++) {

            rabbitTemplate.convertAndSend(EXCHANGE_DIRECT_TIMEOUT, ROUTING_KEY_TIMEOUT, "Test Timeout " + i,postProcessor);

        }

    }

}

然后再次发送，然后观察queue.test.timeout队列中的Ready和Total

大概7秒时间消息又没有了，这次是因为我们的后置处理器中对所生产的消息设定了超过7秒若不处理就丢弃该消息。

关键字词：RabbitMQ,视频,进阶篇,消息,超时

上一篇：尚硅谷RabbitMQ视频-028-进阶篇-消费端限流

下一篇：返回列表

您当前的位置： 首页 > 慢生活 > 程序人生 网站首页程序人生

尚硅谷RabbitMQ视频-029-进阶篇-消息超时

相关文章

您当前的位置：首页 > 慢生活 > 程序人生网站首页程序人生