RabbitMQ 安装延迟队列插件实现延时消息

RabbitMQ 延迟队列的实现，是通过死信队列与转发队列配合实现延时消息的功能
本篇将利用RabbitMQ的延时队列插件，实现延时功能，下面分两部分介绍怎么实现延时功能

1.用Docker安装 RabbitMQ和延时插件
插件下载地址：https://www.rabbitmq.com/community-plugins.html

安装方式有两种：
  1.使用Dockerfile 做成镜像，运行镜像即可，简单方便
  2.启动rabbitmq容器，拿到容器Id，然后把插件复制进去，再启用插件

第一种方式（先下载好插件，放入Dockerfile文件所在目录）：
 1.Dockerfile内容
   FROM rabbitmq:3.8.2-management
   COPY ["rabbitmq_delayed_message_exchange-3.8.0.ez" , "/plugins/"]
   RUN rabbitmq-plugins enable rabbitmq_delayed_message_exchange
 2.构建：docker build -t rabbitmq:3.8.2-management .
 3.运行：docker run -it -d --hostname my-rabbit --name rabbitmq -p 15672:15672 -p 5672:5672 -v /data/rabbitmq/data:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin123456 rabbitmq:3.8.2-management

第二种方式：
  1.先运行rabbitmq： docker run -it -d --hostname rabbit-host --name rabbitmq -v /data/rabbitmq/data:/var/lib/rabbitmq -e RABBITMQ_DEFAULT_USER=写你想要的名字-e RABBITMQ_DEFAULT_PASS=写你想要的密码 -p 15672:15672 -p 5672:5672 rabbitmq:3.8.2-management
  2.登录RabbitMq客户端  http://自己的地址:15672/
  3.把插件上传到服务器目录
  4.查看容器的ID：docker ps
  5.把插件复制到容器：
    格式：docker cp 本地文件路径 counterID全称:容器路径
    docker cp rabbitmq_delayed_message_exchange-3.8.0.ez 982b190bf852:/plugins

  6.进入容器：docker exec -it rabbitmq /bin/bash ，查看插件是否复制进去了
  7.激活插件：cd /plugins ，rabbitmq-plugins enable rabbitmq_delayed_message_exchange
  8.重启容器：docker restart rabbitmq
  9.进入15672后台，可以看到rabbitMq客户端就会出现消息延迟队列的类型：delayedExchange 队列

我使用第一种方式，在本地打包了一个镜像，运行该镜像后访问 http://localhost:15672，在Exchanges–下面的Add a new exchanges — type下拉框中看到 x-delayed-message 就说明插件安装、启动成功了

2.使用SpringBoot项目测试延时队列和普通队列
1.引入依赖

<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>

2.添加rabbitmq 地址和帐号

spring:
  rabbitmq:
    host: 192.168.68.134
    port: 5672
    username: admin
    password: admin123456
    listener:
      direct:
        acknowledge-mode: MANUAL
      simple:
        acknowledge-mode: MANUAL

3.配置队列信息

@Configuration
public class RabbitConfig {

    // 支付超时延时交换机
    public static final String Delay_Exchange_Name = "delay.exchange";

    //exchange name
    public static final String Default_Exchange_Name = "test.exchange";

    //微信支付后，发送支付队列
    public static final String Order_Pay_Queue_Name = "order_pay";

    //超时订单关闭队列
    public static final String Timeout_Trade_Queue_Name = "close_trade";

    @Bean
    public Queue orderPayQueue(){
        return new Queue(RabbitConfig.Order_Pay_Queue_Name , true);
    }

    @Bean
    public Queue delayPayQueue() {
        return new Queue(RabbitConfig.Timeout_Trade_Queue_Name, true);
    }

    // 定义广播模式的延时交换机 无需绑定路由
    @Bean
    FanoutExchange delayExchange(){
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-delayed-type", "direct");
        FanoutExchange topicExchange = new FanoutExchange(RabbitConfig.Delay_Exchange_Name, true, false, args);
        topicExchange.setDelayed(true);
        return topicExchange;
    }
    @Bean
    public DirectExchange defaultExchange() {
        return new DirectExchange(RabbitConfig.Default_Exchange_Name, true, false);
    }

    // 绑定延时队列与交换机
    @Bean
    public Binding delayPayBind() {
        return BindingBuilder.bind(delayPayQueue()).to(delayExchange());
    }

    // 绑定普通消息队列
    @Bean
    public Binding orderPayBind(){
        return BindingBuilder.bind(orderPayQueue()).to(defaultExchange()).with(RabbitConfig.Order_Pay_Queue_Name);
    }

    // 定义消息转换器
    @Bean
    Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // 定义消息模板用于发布消息，并且设置其消息转换器
    @Bean
    RabbitTemplate rabbitTemplate(final ConnectionFactory connectionFactory) {
        final RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        return rabbitTemplate;
    }
    @Bean
    RabbitAdmin rabbitAdmin(final ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }
}

4.添加生产者

@Component
public class RabbitmqPublish {

    @Autowired
    RabbitTemplate rabbitTemplate;

    /**
     * 发送普通消息
     * @param routingKey
     * @param content
     */
    public void sendMsg(String routingKey, String content){
        rabbitTemplate.convertAndSend(RabbitConfig.Default_Exchange_Name , routingKey , content);
    }

    /**
     * 发送延时信息
     * @param content 内容
     * @param routingKey   routingKey
     * @param delay   延时时间，秒
     */
    public void sendTimeoutMsg(String content , String routingKey, int delay){
        // 通过广播模式发布延时消息 延时30分钟 持久化消息 消费后销毁 这里无需指定路由，会广播至每个绑定此交换机的队列
        rabbitTemplate.convertAndSend(RabbitConfig.Delay_Exchange_Name, routingKey, content, message ->{
            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
            message.getMessageProperties().setDelay(delay * 1000);   // 毫秒为单位，指定此消息的延时时长
            return message;
        });
    }
}

5.添加消费者

@Component
public class RabbitmqConsumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    //消费延时消息
    @RabbitListener(queues = RabbitConfig.Timeout_Trade_Queue_Name)
    public void process(String content, Message message, Channel channel) throws IOException {
        try {
            logger.info("延迟队列的内容[{}]", content);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            logger.info("超时信息处理完毕");
        } catch (Exception e) {
            logger.error("处理失败:{}", e.getMessage());
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

    //消费普通消息
    @RabbitListener(queues = RabbitConfig.Order_Pay_Queue_Name)
    public void process1(String content, Message message, Channel channel) throws IOException {
        try {
            logger.info("普通队列的内容[{}]", content);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            logger.error("处理失败:{}", e.getMessage());
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}

6.添加一个控制器，发送队列消息，看控制台结果，发了3个延时消息，分别延时 40秒、20秒、60秒，看那个先被消费，再发送了两个普通消息，测试结果请看下面的截图

/**
 * @Author admin
 * @Date
 */
@RestController
public class IndexController {

    @Autowired
    RabbitmqPublish rabbitmqPublish;

    @RequestMapping("/send")
    public String send() {
        //发送多个延时消息
        rabbitmqPublish.sendTimeoutMsg("hello1" , "routingKey1" ,40);
        rabbitmqPublish.sendTimeoutMsg("hello2" , "routingKey2" ,20);
        rabbitmqPublish.sendTimeoutMsg("hello3" , "routingKey1" ,60);

        //发送普通消息
        rabbitmqPublish.sendMsg(RabbitConfig.Order_Pay_Queue_Name , "weixin");
        rabbitmqPublish.sendMsg(RabbitConfig.Order_Pay_Queue_Name , "alipay");

        return "success";
    }
}

可以看到普通消息先被消费，后面延时消息第一个被消费的是 延时20秒的 hello2，测试成功