• 7. RabbitMQ之延时队列

    文章目录


    延时队列的特性就是体现在对队列中的消息进行延时处理上，延时队列中的元素会在指定时间到达后被消费处理。
    延时队列使用场景有很多，比如客户下的订单在10分钟内未支付就自动取消。如果使用定时任务的方式处理超时订单，假设短时间内生成了大批量的订单，每条订单都要插入数据库，使用定时任务定时10分钟开始处理消息时，需要大量的轮询数据库会对数据库造成压力，另外在10分钟时定时任务开始处理的话，1s时间内也不一定能把所有订单轮询一遍，导致未支付的订单未能在10分钟内取消，时间上延迟不一定准确。而RabbitMQ的延迟队列却能对大量需要延迟的消息进行精准处理。

    1. 延迟队列的实现方式

    延迟队列是通过TTL实现的，TTL指Time To Live，消息存活时间，TTL是RabbitMQ中消息或者队列的属性，表明可以设置一条消息或者队列中所有消息的存活时间。可以对一条消息设置TTL，表示这条消息的存活时间；也可以对队列设置TTL，表示队列中所有消息的存活时间都是一致的。

    如果某条消息被设置了TTL，或者队列被设置了TTL属性，那么在TTL时间内这些消息没有被消费的话就会变成死信消息，被投递到死信队列中。

    在上一篇文章中已经介绍了2种设置TTL的方式：一种就是为队列设置x-message-ttl参数，使队列中所有消息有相同的TTL；另一种就是在生产端设置消息的TTLAMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build()。

    两种设置TTL的方式还是有一定区别的：第一种对队列设置 TTL 属性，那么一旦消息过期，就会被队列丢弃(如果配置了死信队列被丢到死信队列中)；而第二种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间；另外，还需要注意的一点是，如果不设置 TTL，表示消息永远不会过期，如果将 TTL 设置为 0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

    如果上述两种方式都设置了TTL，那么设置TTL时间较小的一种方式生效。

    2. 延迟队列案例——为队列设置TTL

    如下图所示，生产者生产的消息经过normal_exchange交换机分别路由到queue_10和queue_30队列中，queue_10队列会对消息进行延迟10s，queue_30队列会对消息延迟30s，当消息TTL到期后如还未被消费就会经过dead_exchange交换机被投递到死信队列queue_dead中。
    在这里插入图片描述
    下面通过代码实现上述延迟队列的案例。
    本案例采用SpringBoot搭建工程
    1. 首先创建一个SpringBoot工程或者maven工程，pom中依赖如下所示

    
<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.0modelVersion>

    <parent>
        <groupId>org.springframework.bootgroupId>
        <artifactId>spring-boot-starter-parentartifactId>
        <version>2.3.11.RELEASEversion>
    parent>

    <groupId>com.lzjgroupId>
    <artifactId>spring-rabbitmqartifactId>
    <version>1.0-SNAPSHOTversion>

    <properties>
        <maven.compiler.source>8maven.compiler.source>
        <maven.compiler.target>8maven.compiler.target>
    properties>

    <dependencies>
        
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-webartifactId>
        dependency>
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-testartifactId>
        dependency>
        
        <dependency>
            <groupId>org.springframework.bootgroupId>
            <artifactId>spring-boot-starter-amqpartifactId>
        dependency>
        
        <dependency>
            <groupId>org.springframework.amqpgroupId>
            <artifactId>spring-rabbit-testartifactId>
        dependency>

        <dependency>
            <groupId>org.projectlombokgroupId>
            <artifactId>lombokartifactId>
        dependency>
    dependencies>



project>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58

    然后application.properties配置文件中添加RabbitMQ的服务器配置

    spring.rabbitmq.host=192.168.85.100
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123

    • 1
    • 2
    • 3
    • 4

    2. 声明交换机和队列并进行绑定
    下面分别声明normal_exchange和dead_exchange交换机，声明queue_10、queue_30和queue_dead队列，并把normal_exchange绑定到queue_10和queue_30队列上，把dead_exchange交换机绑定到queue_dead死信队列上。其中queue_10设置的TTL为10s，queue_30设置的TTL为30s。

    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;

@Configuration
public class TtlConfig {
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchange";

    public static final String QUEUE_10 = "queue_10";
    public static final String QUEUE_30 = "queue_30";
    public static final String QUEUE_DEAD = "queue_dead";

    /*声明正常交换机*/
    @Bean("normalExchange")
    public DirectExchange normalExchange(){
        return new DirectExchange(NORMAL_EXCHANGE);
    }

    /*声明死信交换机*/
    @Bean("deadExchange")
    public DirectExchange deadExchange(){
        return new DirectExchange(DEAD_EXCHANGE);
    }

    /*声明队列queue_10，设置TTL消息过期时间为10s并绑定死信队列dead_exchange*/
    @Bean("queue10")
    public Queue queue10(){
        Map<String, Object> args = new HashMap<>(3);
        /*绑定死信交换机*/
        args.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        /*设置路由到死信队列中的routingKey*/
        args.put("x-dead-letter-routing-key", "dead_signals");
        /*声明队列的TTL为10s, 消息在队列中存活时间超过10s的都会发送到死信队列中*/
        args.put("x-message-ttl", 10000);
        return QueueBuilder.nonDurable(QUEUE_10).withArguments(args).build();
    }

    /*声明队列queue_30，设置TTL消息过期时间为30s并绑定死信队列dead_exchange*/
    @Bean("queue30")
    public Queue queue30(){
        Map<String, Object> args = new HashMap<>(3);
        /*绑定死信交换机*/
        args.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        /*设置路由到死信队列中的routingKey*/
        args.put("x-dead-letter-routing-key", "dead_signals");
        /*声明队列的TTL为10s, 消息在队列中存活时间超过10s的都会发送到死信队列中*/
        args.put("x-message-ttl", 30000);
        return QueueBuilder.nonDurable(QUEUE_30).withArguments(args).build();
    }

    /*创建死信队列*/
    @Bean("queueDead")
    public Queue queueDead(){
        return new Queue(QUEUE_DEAD);
    }

    /*声明队列queue_10通过routingKey=normal_10绑定到normal_exchange中*/
    @Bean
    public Binding queue10BindingNormalExchange(@Qualifier("queue10")Queue queue10,
                                                @Qualifier("normalExchange")DirectExchange normalExchange){
        return BindingBuilder.bind(queue10).to(normalExchange).with("normal_10");
    }

    /*声明队列queue_30通过routingKey=normal_30绑定到normal_exchange中*/
    @Bean
    public Binding queue30BindingNormalExchange(@Qualifier("queue30")Queue queue30,
                                                @Qualifier("normalExchange")DirectExchange normalExchange){
        return BindingBuilder.bind(queue30).to(normalExchange).with("normal_30");
    }

    /*声明queue_dead队列通过routingKey=dead_sginals绑定到dead_exchange*/
    @Bean
    public Binding queueDeadBindingDeadExchange(@Qualifier("queueDead")Queue queueDead,
                                                @Qualifier("deadExchange")DirectExchange deadExchange){
        return BindingBuilder.bind(queueDead).to(deadExchange).with("dead_signals");
    }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82

    3. 创建Producer生产者用于生产消息
    例如把"hello rabbitmq"消息分别发到通过normal_exchange交换机发到queue_10和queue_30队列中。

    @Slf4j
@Component
public class Producer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void produceMessage(){
        String message = "hello rabbitmq";
        log.info("当前时间为：{}, 生产者生产消息：{}", new Date().toString(), message);
        rabbitTemplate.convertAndSend(TtlConfig.NORMAL_EXCHANGE, "normal_10", message);
        rabbitTemplate.convertAndSend(TtlConfig.NORMAL_EXCHANGE, "normal_30", message);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    4. 创建Consumer消费者用于消费死信队列中的消息
    queue_10和queue_30队列中消息延迟时间到达后就会把消息发到死信队列中，然后才消费死信队列中消息

    import com.lzj.config.TtlConfig;
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;

@Slf4j
@Component
public class Consumer {

    @RabbitListener(queues = TtlConfig.QUEUE_DEAD)
    public void consuemrMessage(Message message, Channel channel){
        String msg = new String(message.getBody());
        log.info("当前时间为：{}, 收到死信队列的消息为：{}", new Date().toString(), msg);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    5. 测试
    下面执行生产者发布消息观察结果

    import com.lzj.producer.Producer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
@Slf4j
public class SpringbootDemo {

    public static void main(String[] args) {
        ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
        Producer producer = app.getBean(Producer.class);
        producer.produceMessage();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    执行该测试代码，分别经过10s和30s后输出结果如下所示，生产者在01:57:43的时候生产了一个消息，延迟10s后，也即在01:57:53时消费者消费了一个消息，说明该消息在queue_10延迟队列中等待10s后被发向了死信队列，然后被死信消费者消费了。同理在01:58:13处消费者消费了消息，说明消息在queue_30队列中延时了30s然后发向了死信队列，经死信队列的消费者消费掉。

    ……
2022-07-11 01:57:43.675  INFO 12804 --- [           main] com.lzj.producer.Producer                : 当前时间为：Mon Jul 11 01:57:43 CST 2022, 生产者生产消息：hello rabbitmq
2022-07-11 01:57:53.712  INFO 12804 --- [ntContainer#0-1] com.lzj.consumer.Consumer                : 当前时间为：Mon Jul 11 01:57:53 CST 2022, 收到死信队列的消息为：hello rabbitmq
2022-07-11 01:58:13.693  INFO 12804 --- [ntContainer#0-1] com.lzj.consumer.Consumer                : 当前时间为：Mon Jul 11 01:58:13 CST 2022, 收到死信队列的消息为：hello rabbitmq
……

    • 1
    • 2
    • 3
    • 4
    • 5

    3. 延迟队列案例——为消息设置TTL

    上述案例中，对消息有2种延迟策略，因此创建了2个延迟队列，queue_10和queue_30分别对消息延迟10s和30s，如果对消息有多重延迟策略的话，那么就要建多个延迟队列，不方便管理。本案例在上述案例基础上在生产者为每条消息设置TTL，那么就只需要一个延迟队列管理消息即可。
    如图所示，queue_delay就是新增加的一个延迟队列，用于存放生产端设置TTL的消息。
    在这里插入图片描述
    1. 声明queue_delay队列，并绑定normal交换机和死信交换机

    在上述代码基础上，对TtlConfig类添加如下配置

    public static final String QUEUE_DELAY = "queue_delay";

/*声明队列queue_delay，消息的延迟时间由生产者设置*/
@Bean("queueDelay")
public Queue queueDelay(){
    Map<String, Object> args = new HashMap<>(3);
    /*绑定死信交换机*/
    args.put("x-dead-letter-exchange", DEAD_EXCHANGE);
    /*设置路由到死信队列中的routingKey*/
    args.put("x-dead-letter-routing-key", "dead_signals");
    return QueueBuilder.nonDurable(QUEUE_DELAY).withArguments(args).build();
}

/*声明queue_delay队列通过routingKey=normal绑定到normal_exchange交换机*/
@Bean
public Binding queueDelayBindingNormalExchange(@Qualifier("queueDelay")Queue queueDelay,
                                             @Qualifier("normalExchange")DirectExchange normalExchange){
    return BindingBuilder.bind(queueDelay).to(normalExchange).with("normal");
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    2. 创建生产者Producer2，发消息时设置消息的TTL

    import com.lzj.config.TtlConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;

@Slf4j
@Component
public class Producer2 {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void produceMessage(){
        String message = "hello rabbitmq 2";
        log.info("当前时间为：{}, 生产者生产消息：{}", new Date().toString(), message);
        /*hello rabbitmq 2 消息延迟50s后由queue_delay队列发向了queue_dead队列*/
        rabbitTemplate.convertAndSend(TtlConfig.NORMAL_EXCHANGE, "normal", message, msg -> {
            msg.getMessageProperties().setExpiration(String.valueOf(50000));
            return msg;
        });
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    3. 修改启动类，测试Producer2发消息

    @SpringBootApplication
@Slf4j
public class SpringbootDemo {

    public static void main(String[] args) {
        ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
//        Producer producer = app.getBean(Producer.class);
//        producer.produceMessage();
        Producer2 producer2 = app.getBean(Producer2.class);
        producer2.produceMessage();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    启动上面启动类，得到测试结果如下，在23:10:10发布的hello rabbitmq 2消息，在23:11:00时hello rabbitmq 2在死信队列中被死信消费者消费掉，期间正好延迟了50s。说明在生产端为消息设置TTL达到了预期延迟的期望。

    ……
2022-07-15 23:10:10.315  INFO 9316 --- [           main] com.lzj.producer.Producer2               : 当前时间为：Fri Jul 15 23:10:10 CST 2022, 生产者生产消息：hello rabbitmq 2
2022-07-15 23:11:00.362  INFO 9316 --- [ntContainer#0-1] com.lzj.consumer.Consumer                : 当前时间为：Fri Jul 15 23:11:00 CST 2022, 收到死信队列的消息为：hello rabbitmq 2

    • 1
    • 2
    • 3

    但是，但是，但是
    通过上述生产端设置消息的TTL是有一定隐患的，尤其是当生产端要发送的消息设置的TTL不同时，问题非常严重。

    下面设想生产端发送2条消息，第一条消息设置TTL为50s，第二条消息TTL设置20s，重新测试该案例看一下结果如何。

    首先把Producer2改成下面形式，连续发2条消息分别为hello rabbitmq 2 和 hello rabbitmq 3

    @Slf4j
@Component
public class Producer2 {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void produceMessage(){
        String message = "hello rabbitmq 2";
        log.info("当前时间为：{}, 生产者生产消息：{}", new Date().toString(), message);
        /*hello rabbitmq 2 消息延迟50s后由queue_delay队列发向了queue_dead队列*/
        rabbitTemplate.convertAndSend(TtlConfig.NORMAL_EXCHANGE, "normal", message, msg -> {
            msg.getMessageProperties().setExpiration(String.valueOf(50000));
            return msg;
        });
        /*hello rabbitmq 3 消息延迟20s后由queue_delay队列发向了queue_dead队列*/
        message = "hello rabbitmq 3";
        log.info("当前时间为：{}, 生产者生产消息：{}", new Date().toString(), message);
        rabbitTemplate.convertAndSend(TtlConfig.NORMAL_EXCHANGE, "normal", message, msg -> {
            msg.getMessageProperties().setExpiration(String.valueOf(20000));
            return msg;
        });
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23

    然后我们重新启动启动类查看测试结果

    @SpringBootApplication
@Slf4j
public class SpringbootDemo {

    public static void main(String[] args) {
        ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
//        Producer producer = app.getBean(Producer.class);
//        producer.produceMessage();
        Producer2 producer2 = app.getBean(Producer2.class);
        producer2.produceMessage();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    经过测试输出如下所示，发现在00:00:13时间发布了消息hello rabbitmq 2和hello rabbitmq 3，在00:01:03时间消费了死信队列中消息hello rabbitmq 2，hello rabbitmq 2延迟了50s，正好在发布hello rabbitmq 2消息时也是设置的TTL为50s，完全吻合消息的延迟特性；但是也是在00:01:03时间消费了死信队列中的hello rabbitmq 3消息，期间也是延迟了50s，但是hello rabbitmq 3在发布时设置 TTL为20s，没有符合消息设置的延迟时间，为什么会出现hello rabbitmq 3消息没有如期死亡这种问题呢？
    是因为对于生产端设置消息TTL延迟时间的消息，RabbitMQ只会检查队列中第一个消息是否过期，如果过期就丢到死信队列中，如果未过期就一直待在队列中。那么就会出现这种问题，当第一个消息的TTL延迟时间比较长，而第二个消息延迟时间比较短就会导致第二个消息一直得不到执行。比如本案例中的hello rabbitmq 2设置的TTL为50s，hello rabbitmq 3设置的TTL为20s，导致hello rabbitmq 2在50s内一直未过期，则hello rabbitmq 3就无法被检查，即使20s时间到了也不会被丢到死信队列中。

    2022-07-16 00:00:13.797  INFO 8952 --- [           main] com.lzj.producer.Producer2               : 当前时间为：Sat Jul 16 00:00:13 CST 2022, 生产者生产消息：hello rabbitmq 2
2022-07-16 00:00:13.814  INFO 8952 --- [           main] com.lzj.producer.Producer2               : 当前时间为：Sat Jul 16 00:00:13 CST 2022, 生产者生产消息：hello rabbitmq 3
2022-07-16 00:01:03.834  INFO 8952 --- [ntContainer#0-1] com.lzj.consumer.Consumer                : 当前时间为：Sat Jul 16 00:01:03 CST 2022, 收到死信队列的消息为：hello rabbitmq 2
2022-07-16 00:01:03.835  INFO 8952 --- [ntContainer#0-1] com.lzj.consumer.Consumer                : 当前时间为：Sat Jul 16 00:01:03 CST 2022, 收到死信队列的消息为：hello rabbitmq 3

    • 1
    • 2
    • 3
    • 4

    4. 延迟队列案例——通过交换机插件延迟消息

    为了解决上面问题，可以通过官方提供的插件形式设置交换机具有延迟特性，以达到延迟消息的目的。生产端为不同消息设置不同延迟时间TTL的话，延迟交换机会进行判断，如果达到TTL，消息才会被送到队列中。

    1. 安装插件
    要想实现插件延迟消息，首先要从官网下载插件rabbitmq_delayed_message_exchange-3.8.0.ez，安装插件的方式非常简单，只需要把插件放到RabbitMQ安装目录下的plugins下面即可/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.8/plugins，最后执行下面命令使插件生效

    rabbitmq-plugins enable rabbitmq_delayed_message_exchange

    • 1

    插件生效后，可以观察到不一样的地方，在RabbitMQ的浏览器插件管理端可以看到交换机多了一种类型， x-delayed-message类型的交换机，表示通过插件的形式延迟消息的交换机。
    在这里插入图片描述
    2. 通过代码实现案例
    生产者生产的不同TTL的消息经过延迟交换机，消息达到TTL的就会被路由到队列中，被消费者进行消费。流程图如下所示
    在这里插入图片描述
    2.1 首先配置延迟交换机和延迟队列，并通过routingKey进行绑定。

    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 DelayedExchangeConfig {
    public static final String DELAYED_EXCHANGE = "delayed_exchange";
    public static final String DELAYED_QUEUE = "delayed_queue";
    public static final String DELAYED_ROUTING_KEY = "delay";

    /*自定义一个延迟交换机*/
    @Bean
    public CustomExchange delayedExchange(){
        Map<String, Object> args = new HashMap<>();
        /*消息的延迟类型*/
        args.put("x-delayed-type", "direct");
        /*
        * 1. 第一个参数表示延迟交换机名字
        * 2. 第二个参数表示延迟交换机类型
        * 3. 第三个参数表示交换机持久化
        * 4. 第4个参数表示交换机中消息不自动删除
        * 5. 其他参数
        * */
        return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", true, false, args);
    }

    /*创建延迟队列*/
    @Bean
    public Queue delayedQueue(){
        return new Queue(DELAYED_QUEUE);
    }

    /*延迟交换机绑定延迟队列*/
    @Bean
    public Binding delayedExchangeBindingQueue(@Qualifier("delayedExchange") CustomExchange delayedExchange,
                                               @Qualifier("delayedQueue") Queue delayedQueue){
        return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45

    2.2 创建生产者生产消息

    import com.lzj.config.DelayedExchangeConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.Date;

@Slf4j
@Component
public class Producer2 {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void produceMessage(){
        String message = "hello rabbitmq 4";
        log.info("当前时间为：{}, 生产者生产消息：{}", new Date().toString(), message);
        /*hello rabbitmq 2 消息延迟50s后由queue_delay队列发向了queue_dead队列*/
        rabbitTemplate.convertAndSend(DelayedExchangeConfig.DELAYED_EXCHANGE, DelayedExchangeConfig.DELAYED_ROUTING_KEY, message, msg -> {
            msg.getMessageProperties().setDelay(50000);
            return msg;
        });
        /*hello rabbitmq 3 消息延迟20s后由queue_delay队列发向了queue_dead队列*/
        message = "hello rabbitmq 5";
        log.info("当前时间为：{}, 生产者生产消息：{}", new Date().toString(), message);
        rabbitTemplate.convertAndSend(DelayedExchangeConfig.DELAYED_EXCHANGE, DelayedExchangeConfig.DELAYED_ROUTING_KEY, message, msg -> {
            msg.getMessageProperties().setDelay(20000);
            return msg;
        });
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30

    2.3 创建消费者消费消息

    @Slf4j
@Component
public class Consumer2 {

    @RabbitListener(queues = DelayedExchangeConfig.DELAYED_QUEUE)
    public void consuemrMessage(Message message, Channel channel){
        String msg = new String(message.getBody());
        log.info("当前时间为：{}, 收到死信队列的消息为：{}", new Date().toString(), msg);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    2.4 测试
    下面进行测试插件延迟交换机延迟消息的案例

    @SpringBootApplication
@Slf4j
public class SpringbootDemo {

    public static void main(String[] args) {
        ConfigurableApplicationContext app = SpringApplication.run(SpringbootDemo.class, args);
        Producer2 producer2 = app.getBean(Producer2.class);
        producer2.produceMessage();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    输出结果如下所示，发现消息hello rabbitmq 4和hello rabbitmq 5同在20 00:46:54时刻进行生产，消息hello rabbitmq 5延迟了20s，因此消息hello rabbitmq 5在20 00:47:14时刻被消费；而hello rabbitmq 4延迟了50s，消息hello rabbitmq 4在20 00:47:44时刻被消费。从而证明每个消息都被精准的进行了延迟。

    ……
2022-07-20 00:46:54.784  INFO 8252 --- [           main] com.lzj.producer.Producer2               : 当前时间为：Wed Jul 20 00:46:54 CST 2022, 生产者生产消息：hello rabbitmq 4
2022-07-20 00:46:54.801  INFO 8252 --- [           main] com.lzj.producer.Producer2               : 当前时间为：Wed Jul 20 00:46:54 CST 2022, 生产者生产消息：hello rabbitmq 5
2022-07-20 00:47:14.630  INFO 8252 --- [ntContainer#1-1] com.lzj.consumer.Consumer2               : 当前时间为：Wed Jul 20 00:47:14 CST 2022, 收到死信队列的消息为：hello rabbitmq 5
2022-07-20 00:47:44.326  INFO 8252 --- [ntContainer#1-1] com.lzj.consumer.Consumer2               : 当前时间为：Wed Jul 20 00:47:44 CST 2022, 收到死信队列的消息为：hello rabbitmq 4
……

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6

    5. 总结

    1. 为队列设置TTL，优点：可以对消息进行精准延迟，缺点：如果不同消息需要延迟不同的时间，就需要创建管理很多个延迟队列，成本 高；
    2. 为消息设置TTL，只需一个延迟队列即可，但不能对消息进行精准的延迟，有的消息已到达TTL还未被及时消费；
    3. 通过交换机插件延迟消息，仅需一个队列即可，且可对消息进行精准延迟。
  • 相关阅读:
    js的闭包例题
    RabbitMQ系列【17】RabbitOperations接口详解
    学习STM32第十六天
    专栏 | 解析“全闪对象存储”（二）
    unsafe value used in a resource URL context，angular框架下嵌入iframe报错问题解决
    Github进行fork后如何与原仓库同步
    神经网络数据融合特征提取,神经网络 信息融合
    14. 机器学习 - KNN & 贝叶斯
    vue3项目实战中的接口调用方法（二）fetch用法 （前后端交互） get请求/post请求/put请求/delete请求
    机器学习 sklearn数据集
  • 原文地址：https://blog.csdn.net/u010502101/article/details/125734862