手把手带你掌握SpringBoot RabbitMQ延迟队列

$ docker cp /Users/ludangxin/Downloads/rabbitmq_delayed_message_exchange- 3.8 . 9 -0199d11c.ez rabbitmq:/opt/rabbitmq/plugins/

$ docker exec -it rabbitmq /bin/bash

$ rabbitmq-plugins list

# 输出部分内容如下 [E*] = 明确启用; e = 隐式启用

[  ] rabbitmq_amqp1_0                  3.8 . 3

[  ] rabbitmq_auth_backend_cache       3.8 . 3

[  ] rabbitmq_auth_backend_http        3.8 . 3

[  ] rabbitmq_auth_backend_ldap        3.8 . 3

[  ] rabbitmq_auth_backend_oauth2      3.8 . 3

[  ] rabbitmq_auth_mechanism_ssl       3.8 . 3

[  ] rabbitmq_consistent_hash_exchange 3.8 . 3

[  ] rabbitmq_event_exchange           3.8 . 3

[  ] rabbitmq_federation               3.8 . 3

[  ] rabbitmq_federation_management    3.8 . 3

[  ] rabbitmq_jms_topic_exchange       3.8 . 3

[E*] rabbitmq_management               3.8 . 3

[e*] rabbitmq_management_agent         3.8 . 3

[  ] rabbitmq_mqtt                     3.8 . 3

$ rabbitmq-plugins enable rabbitmq_delayed_message_exchange

< dependency >

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

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

</ dependency >

< dependency >

     < groupId >org.projectlombok</ groupId >

     < artifactId >lombok</ artifactId >

     < optional >true</ optional >

</ dependency >

< dependency >

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

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

     < scope >test</ scope >

</ dependency >

< dependency >

     < groupId >org.springframework.amqp</ groupId >

     < artifactId >spring-rabbit-test</ artifactId >

     < scope >test</ scope >

</ dependency >

spring:

   rabbitmq:

     host: localhost

     port: 5672

     # rabbit 默认的虚拟主机

     virtual-host: /

     # rabbit 用户名密码

     username: admin

     password: admin123

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.amqp.core.QueueBuilder;

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;

/**

  * 延迟队列配置

  *

  * @author ludangxin

  * @date 2021/9/16

  */

@Configuration

public class RabbitDelayedConfig {

     public static final String QUEUE_NAME_DELAYED = "DELAY.QUEUE" ;

     public static final String EXCHANGE_NAME_DELAYED = "DELAY.EXCHANGE" ;

     public static final String ROUTING_KEY_DELAYED = "DELAY.#" ;

     @Bean (QUEUE_NAME_DELAYED)

     public Queue queue() {

        return QueueBuilder.durable(QUEUE_NAME_DELAYED).build();

     }

     @Bean (EXCHANGE_NAME_DELAYED)

     public CustomExchange exchange() {

        Map<String, Object> arguments = new HashMap<>( 1 );

        // 在这里声明一个主题类型的延迟队列，当然其他类型的也可以。

        arguments.put( "x-delayed-type" , "topic" );

        return new CustomExchange(EXCHANGE_NAME_DELAYED, "x-delayed-message" , true , false , arguments);

     }

     @Bean

     public Binding bindingNotify( @Qualifier (QUEUE_NAME_DELAYED) Queue queue, @Qualifier (EXCHANGE_NAME_DELAYED) CustomExchange customExchange) {

        return BindingBuilder.bind(queue).to(customExchange).with(ROUTING_KEY_DELAYED).noargs();

     }

}

import com.ldx.rabbitmq.config.RabbitDelayedConfig;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.core.MessageProperties;

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

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

import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

/**

  * 延迟消息生产者

  *

  * @author ludangxin

  * @date 2021/9/9

  */

@Component

public class DelayProducer {

    @Autowired

    private RabbitTemplate rabbitTemplate;

    public void sendDelayedMsg(String msg, Integer delay) {

       MessageProperties mp = new MessageProperties();

       // 设置过期时间

       mp.setDelay(delay);

       Message message = new Message(msg.getBytes(), mp);

       rabbitTemplate.convertAndSend(RabbitDelayedConfig.EXCHANGE_NAME_DELAYED, "DELAY.MSG" , message);

    }

}

import com.ldx.rabbitmq.config.RabbitDelayedConfig;

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.time.LocalDateTime;

import java.time.format.DateTimeFormatter;

/**

  * 延迟消息消费者

  *

  * @author ludangxin

  * @date 2021/9/9

  */

@Slf4j

@Component

public class DelayConsumer {

     @RabbitListener (queues = {RabbitDelayedConfig.QUEUE_NAME_DELAYED})

     public void delayQueue(Message message){

         log.info( new String(message.getBody()) + "，结束时间为：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern( "yyyy-MM-dd HH:mm:ss" )));

     }

}

@Autowired

private DelayProducer delayProducer;

@Test

@SneakyThrows

public void sendDelayedMsg() {

    for ( int i = 16 ; i >= 10 ; i --) {

       String msg = "我将在" + i + "s后过期，开始时间为：" + LocalDateTime.now().format(DateTimeFormatter.ofPattern( "yyyy-MM-dd HH:mm:ss" ));

       delayProducer.sendDelayedMsg(msg,i * 1000 );

    }

    // 使进程阻塞，方便Consumer监听输出Message

    System.in.read();

}