上一篇文章写了docker安装RabbitMQ及延迟插件的安装,这篇的话是基于RabbitMQ延迟插件实现延迟队列的示例

那么废话不多说直接上代码!!

首先创建延迟队列配置类 DelayedQueueConfig

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

    //交换机

    public static final String DELAYED_EXCHANGE_NAME = "delayed.exchange";

    //队列

    public static final String DELAYED_QUEUE_NAME = "delayed.queue";

    //routingkey

    public static final String DELAYED_ROUTING_KEY = "delayed.routingkey";

    //声明交换机

    @Bean

    public CustomExchange delayedExchange() {

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

        //设置延迟交换机为直接类型

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

        /**

         * 参数说明

         * 1.交换机名字

         * 2.交换机类型

         * 3.是否持久化

         * 4.是否自动删除

         * 5.其他参数

         */

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

    }

    //声明队列

    @Bean

    public Queue delayedQueue() {

        return new Queue(DELAYED_QUEUE_NAME);

    }

    //绑定

    @Bean

    public Binding delayedQueueBindingdelayedExchange(

            @Qualifier("delayedQueue") Queue delayedQueue,

            @Qualifier("delayedExchange") CustomExchange delayedExchange) {

        return BindingBuilder.bind(delayedQueue).to(delayedExchange).with(DELAYED_ROUTING_KEY).noargs();

    }

}

接着编写生产者代码

   //基于插件发送延迟消息

    @GetMapping("/senddelayedMsg/{message}/{delayedTime}")

    public void sendMsg(@PathVariable String message, @PathVariable Integer delayedTime) {

        log.info("当前时间:{},发送一条时长{}毫秒给延迟队列delayed.queue:{}", new Date().toString(), delayedTime, message);

        /**

         * convertAndSend 方法参数说明

         * 1.交换机名称

         * 2.routingkey

         * 3.发送的消息

         * 4.发送完消息后的回调

         */

        rabbitTemplate.convertAndSend(DELAYED_EXCHANGE_NAME, DELAYED_ROUTING_KEY, message, msg -> {

            msg.getMessageProperties().setDelay(delayedTime);

            return msg;

        });

    }

最后编写消费者代码

import com.zfd.config.DelayedQueueConfig;

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

    @RabbitListener(queues = DelayedQueueConfig.DELAYED_QUEUE_NAME)

    public void receiveDelayQueue(Message message) {

        String msg = new String(message.getBody());

        log.info("当前时间:{},收到延迟队列的消息:{}", new Date().toString(), msg);

    }

}

接着运行程序发送请求

到这里我们基于插件实现RabbitMQ延迟队列就写完了!!

巴特西

基于RabbltMQ延迟插件实现延迟队列代码示例

上一篇文章写了docker安装RabbitMQ及延迟插件的安装,这篇的话是基于RabbitMQ延迟插件实现延迟队列的示例

最新文章

热门文章