安裝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容器內(nèi)

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

再次進(jìn)入docker容器內(nèi)

docker exec -it 5192576ec93d /bin/bash

啟動(dòng)延時(shí)插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

查看已安裝的插件

rabbitmq-plugins list

繼續(xù)完成在集群每臺(tái)機(jī)器中安裝

延時(shí)機(jī)制

安裝插件后會(huì)生成新的Exchange類型x-delayed-message

image.png

該類型消息支持延遲投遞機(jī)制：
接收到消息后并未立即將消息投遞至目標(biāo)隊(duì)列中藏鹊，而是存儲(chǔ)在mnesia(一個(gè)分布式數(shù)據(jù)系統(tǒng))表中，檢測(cè)消息延遲時(shí)間呻纹，如達(dá)到可投遞時(shí)間時(shí)并將其通過x-delayed-type類型標(biāo)記的交換機(jī)類型投遞至目標(biāo)隊(duì)列忿晕。

java客戶端實(shí)現(xiàn)

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() {
        //聲明交換機(jī)類型
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        //聲明延時(shí)類型
        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";
    /*發(fā)送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("發(fā)送時(shí)間：" + 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("結(jié)束時(shí)間：" + sf.format(new Date()));
        //body
        System.out.println("receiver: " + msg.getBody());
        channel.basicAck(tag, false);
    }
}