spring boot rabbitMQ延时队列 实现

spring boot 使用rabbitMQ
需要导入相关的相关依赖如下：

 

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

AMQP默认实现为rabbitMQ。

 

为了定制化的配置放弃了boot的自动配置，So，以下为RabbitMQ的配置文件：

 

package com.fengbaogu.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * rabbitmq 的配置类
 * @author LiaoLe
 */
@Configuration
public class RabbitMQConfig {
    /** 消息交换机的名字*/
    public static final String EXCHANGE = "delayed_exchange";
    /** 队列key1*/
    public static final String ROUTING = "notify";
    /** 队列名*/
    public static final String QUEUE = "delayed_queue";
    /** rabbit MQ 账号*/
    public static final String USERNAME = "liaoke";
    /**rabbit MQ 密码*/
    public static final String PASSWORD = "198461lk";
    /**rabbit MQ 地址*/
    public static final String RABBITMQ_ADDRESS="localhost";
    /**rabbit MQ 端口*/
    public static final Integer RABBITMQ_PORT=5672;

    /**
     * 配置链接
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(RABBITMQ_ADDRESS,RABBITMQ_PORT);
        connectionFactory.setUsername(USERNAME);
        connectionFactory.setPassword(PASSWORD);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true); /**需要设置 接受回调信息*/
        return connectionFactory;
    }


    /**
     * 交换机设置
     * 开启了延时消息
     * @return
     */
    @Bean
    public DirectExchange defaultExchange() {
        DirectExchange directExchange = new DirectExchange(EXCHANGE, true, false);
        directExchange.setDelayed(true);/**只需简单一步开启延时消息，md找了好久，原来这么简单*/
        return directExchange;
    }

    /**
     * 队列设置
     * @return
     */
    @Bean
    public Queue notifyQueue() {
        return new Queue(QUEUE,true);/**消息持久化*/
    }

    /**
     * 绑定队列到交换机
     * @return
     */
    @Bean
    public Binding bindingNotify() {
        return BindingBuilder.bind(notifyQueue()).to(defaultExchange()).with(ROUTING);
    }

}

 rabbitMQ默认是不开启delayed功能的哟，所以同学们还需要到官网或者githup上面下载插件：

 

命令行： rabbitmq-plugins enable rabbitmq_delayed_message_exchange
 

#安装插件 https://github.com/rabbitmq/rabbitmq-delayed-message-exchange 

 

#插件是有版本要求的哦

 

下面贴发送的代码，随便用了个http入口：

package com.fengbaogu.web;

import com.fengbaogu.config.RabbitMQConfig;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.UUID;

/**
 * Created by LiaoKe on 2017/5/2.
 */
@RestController
public class TestController implements RabbitTemplate.ConfirmCallback{


    /**
     * 使用RabbitTemplate发送消息
     */
    private RabbitTemplate rabbitTemplate;

    /**
     * 构造方法
     * @param rabbitTemplate
     */
    public TestController(RabbitTemplate rabbitTemplate){
        this.rabbitTemplate = rabbitTemplate;
        //设置消费回调
        this.rabbitTemplate.setConfirmCallback(this);
    }

    private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");


    /**
     * 向延迟队列中发送消息
     * @param msg
     * @return
     */
    @RequestMapping("send")
    public String send3(String msg){
        String uuid = UUID.randomUUID().toString();
        CorrelationData correlationId = new CorrelationData(uuid);
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,RabbitMQConfig.ROUTING , msg, new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        message.getMessageProperties().setDelay(20000);/**延时20秒发送*/
                        message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);/**消息持久化*/
                        System.out.println(sdf.format(new Date()) + " Delay sent.");
                        return message;
                    }
                }, correlationId);
        return null;
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println(" 回调id:" + correlationData);
        if (ack) {
            System.out.println("消息成功消费");
        } else {
            System.out.println("消息消费失败:" + cause+"\n重新发送");
        }
    }
}

 

 

启动后可以在RabbitMQ管理界面上有：

 这样的交换机。

Queue同理也可以在管理界面上看到，此处不贴图了。

 

现在来看 Consumer 端，代码异常简洁：

package com.fengbaogu.listener;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Date;


/**
 * Created by LiaoKe on 2017/5/3.
 */
@Component
@RabbitListener(queues = {"delayed_queue"})
public class Receiver {


    @RabbitHandler
    public void process(String msg)  {
        System.out.println("start one work and now data:"+new Date().toString());
        System.out.println("This msg is:"+msg);
    }



}

开启消费端，访问send入口：

服务端打印：

2017-05-03 20:24:14 Delay sent.

 回调id:CorrelationData [id=277ce017-af15-4571-b556-762fec62720d]

消息成功消费

消费端打印：

start one work and now data:Wed May 03 20:24:34 CST 2017

This msg is:我是消息2

 

So,相差20秒。延时队列Over。

 

 

 

评论

1 楼 liaoke0123 2017-05-04  

https://bintray.com/rabbitmq/community-plugins/rabbitmq_delayed_message_exchange/v3.6.x


附上官方插件下载地址