4-4 插件实现消息延迟投递

安装rabbitmq_delayed_message_exchange

查看mq的版本

[docker@quanwugou-dev-001 ~]$ docker image inspect rabbitmq:latest | grep -i version
                "OPENSSL_VERSION=1.1.1e",
                "OTP_VERSION=22.3",
                "RABBITMQ_VERSION=3.8.3",
        "DockerVersion": "18.09.7",
                "OPENSSL_VERSION=1.1.1e",
                "OTP_VERSION=22.3",
                "RABBITMQ_VERSION=3.8.3",

下载插件

wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/v3.8.0/rabbitmq_delayed_message_exchange-3.8.0.ez

拷贝至docker容器内

docker cp rabbitmq_delayed_message_exchange-3.8.0.ez 5192576ec93d:/plugins

再次进入docker容器内

docker exec -it 5192576ec93d /bin/bash

启动延时插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

查看已安装的插件

rabbitmq-plugins list

继续完成在集群每台机器中安装

延时机制

安装插件后会生成新的Exchange类型x-delayed-message

image.png

该类型消息支持延迟投递机制：
接收到消息后并未立即将消息投递至目标队列中，而是存储在mnesia(一个分布式数据系统)表中，检测消息延迟时间，如达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递至目标队列。

java客户端实现

package com.finlay.scaffold.delayed;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

/**
 * @author: Finlay
 * @description:
 * @date: 2020-07-17 4:59 下午
 */
@Configuration
public class DelayedConfig {

    private static final String EXCHANGE_NAME = "delayed.exchange";
    private static final String QUEUE_NAME = "delayed.queue";
    private static final String ROUTING_KEY = "delayed";

    @Bean
    public Queue delayedQueue() {
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    @Bean
    public Exchange delayedExchange() {
        //声明交换机类型
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        //声明延时类型
        return new CustomExchange(EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    @Bean
    public Binding delayedBinding() {
        return new Binding(QUEUE_NAME, Binding.DestinationType.QUEUE, EXCHANGE_NAME, ROUTING_KEY, null);
    }
}

package com.finlay.scaffold.delayed;

import lombok.extern.slf4j.Slf4j;
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.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author: Finlay
 * @description:
 * @date: 2020-07-17 5:28 下午
 */
@Slf4j
@Component
public class DelayedSender {

    private static final String EXCHANGE_NAME = "delayed.exchange";
    /*发送key*/
    private static final String SEND_ROUTING_KEY = "delayed";

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send() {
        MessageProperties properties = new MessageProperties();
        properties.setDelay(2); // 持久化消息
        properties.setHeader("x-delay", 30000);//30秒后投递
        String msg = "Test-Exchange--------------";

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

        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("发送时间：" + sf.format(new Date()));

        rabbitTemplate.convertAndSend(EXCHANGE_NAME, SEND_ROUTING_KEY, message);
    }
}

package com.finlay.scaffold.delayed;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author: Finlay
 * @description:
 * @date: 2020-07-17 5:46 下午
 */
@Slf4j
@Component
public class DelayedReceiver {
    private static final String QUEUE_NAME = "delayed.queue";

    @RabbitListener(queues = QUEUE_NAME)
    @RabbitHandler
    public void rec(Message msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        log.info("结束时间：" + sf.format(new Date()));
        //body
        System.out.println("receiver: " + msg.getBody());
        channel.basicAck(tag, false);
    }
}