Rabbitmq延迟队列实现定时任务的方法

场景

开发中经常需要用到定时任务，对于商城来说，定时任务尤其多，比如优惠券定时过期、订单定时关闭、微信支付2小时未支付关闭订单等等，都需要用到定时任务，但是定时任务本身有一个问题，一般来说我们都是通过定时轮询查询数据库来判断是否有任务需要执行，也就是说不管怎么样，我们需要先查询数据库，而且有些任务对时间准确要求比较高的，需要每秒查询一次，对于系统小倒是无所谓，如果系统本身就大而且数据也多的情况下，这就不大现实了，所以需要其他方式的，当然实现的方式有多种多样的，比如redis实现定时队列、基于优先级队列的jdk延迟队列、时间轮等。因为我们项目中本身就使用到了rabbitmq，所以基于方便开发和维护的原则，我们使用了rabbitmq延迟队列来实现定时任务,不知道rabbitmq是什么的和不知道springboot怎么集成rabbitmq的可以查看我之前的文章 spring boot集成rabbitmq

rabbitmq延迟队列

rabbitmq本身是没有延迟队列的，只能通过rabbitmq本身队列的特性来实现，想要rabbitmq实现延迟队列，需要使用rabbitmq的死信交换机（exchange）和消息的存活时间ttl（time to live）

死信交换机

一个消息在满足如下条件下，会进死信交换机，记住这里是交换机而不是队列，一个交换机可以对应很多队列。

1. 一个消息被consumer拒收了，并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里，被其他消费者使用。
2. 上面的消息的ttl到了，消息过期了。
3. 队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。

死信交换机就是普通的交换机，只是因为我们把过期的消息扔进去，所以叫死信交换机，并不是说死信交换机是某种特定的交换机

消息ttl（消息存活时间）

消息的ttl就是消息的存活时间。rabbitmq可以对队列和消息分别设置ttl。对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的ttl，因为它才是实现延迟任务的关键。

byte[] messagebodybytes = "hello, world!".getbytes(); 
amqp.basicproperties properties = new amqp.basicproperties(); 
properties.setexpiration("60000"); 
channel.basicpublish("my-exchange", "queue-key", properties, messagebodybytes);

可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间，两者是一样的效果。只是expiration字段是字符串参数，所以要写个int类型的字符串： 当上面的消息扔到队列中后，过了60秒，如果没有被消费，它就死了。不会被消费者消费到。这个消息后面的，没有“死掉”的消息对顶上来，被消费者消费。死信在队列中并不会被删除和释放，它会被统计到队列的消息数中去

处理流程图

创建交换机（exchanges）和队列（queues）

创建死信交换机

 

如图所示，就是创建一个普通的交换机，这里为了方便区分，把交换机的名字取为：delay

创建自动过期消息队列

这个队列的主要作用是让消息定时过期的，比如我们需要2小时候关闭订单，我们就需要把消息放进这个队列里面，把消息过期时间设置为2小时

 

创建一个一个名为delay_queue1的自动过期的队列，当然图片上面的参数并不会让消息自动过期，因为我们并没有设置x-message-ttl参数，如果整个队列的消息有消息都是相同的，可以设置，这里为了灵活，所以并没有设置，另外两个参数x-dead-letter-exchange代表消息过期后，消息要进入的交换机，这里配置的是delay，也就是死信交换机，x-dead-letter-routing-key是配置消息过期后，进入死信交换机的routing-key,跟发送消息的routing-key一个道理，根据这个key将消息放入不同的队列

创建消息处理队列

这个队列才是真正处理消息的队列，所有进入这个队列的消息都会被处理

 

消息队列的名字为delay_queue2

消息队列绑定到交换机

进入交换机详情页面，将创建的2个队列（delay queue1和delay queue2）绑定到交换机上面

 

自动过期消息队列的routing key 设置为delay

绑定delay queue2

delay queue2 的key要设置为创建自动过期的队列的x-dead-letter-routing-key参数，这样当消息过期的时候就可以自动把消息放入delay_queue2这个队列中了

绑定后的管理页面如下图：

当然这个绑定也可以使用代码来实现，只是为了直观表现，所以本文使用的管理平台来操作

发送消息

string msg = "hello word"; 
messageproperties messageproperties = new messageproperties(); 
  messageproperties.setexpiration("6000");
  messageproperties.setcorrelationid(uuid.randomuuid().tostring().getbytes());
  message message = new message(msg.getbytes(), messageproperties);
  rabbittemplate.convertandsend("delay", "delay",message);

主要的代码就是

messageproperties.setexpiration("6000"); 

设置了让消息6秒后过期

注意：因为要让消息自动过期，所以一定不能设置delay_queue1的监听，不能让这个队列里面的消息被接受到，否则消息一旦被消费，就不存在过期了

接收消息

接收消息配置好delay_queue2的监听就好了

package wang.raye.rabbitmq.demo1;
import org.springframework.amqp.core.acknowledgemode; 
import org.springframework.amqp.core.binding; 
import org.springframework.amqp.core.bindingbuilder; 
import org.springframework.amqp.core.directexchange; 
import org.springframework.amqp.core.message; 
import org.springframework.amqp.core.queue; 
import org.springframework.amqp.rabbit.connection.cachingconnectionfactory; 
import org.springframework.amqp.rabbit.connection.connectionfactory; 
import org.springframework.amqp.rabbit.core.channelawaremessagelistener; 
import org.springframework.amqp.rabbit.listener.simplemessagelistenercontainer; 
import org.springframework.beans.factory.annotation.autowired; 
import org.springframework.context.annotation.bean; 
import org.springframework.context.annotation.configuration;
@configuration
public class delayqueue { 
 /** 消息交换机的名字*/
 public static final string exchange = "delay";
 /** 队列key1*/
 public static final string routingkey1 = "delay";
 /** 队列key2*/
 public static final string routingkey2 = "delay_key";

 /**
  * 配置链接信息
  * @return
  */
 @bean
 public connectionfactory connectionfactory() {
  cachingconnectionfactory connectionfactory = new cachingconnectionfactory("120.76.237.8",5672);

  connectionfactory.setusername("kberp");
  connectionfactory.setpassword("kberp");
  connectionfactory.setvirtualhost("/");
  connectionfactory.setpublisherconfirms(true); // 必须要设置
  return connectionfactory;
 }

 /** 
  * 配置消息交换机
  * 针对消费者配置 
  fanoutexchange: 将消息分发到所有的绑定队列，无routingkey的概念 
  headersexchange ：通过添加属性key-value匹配 
  directexchange:按照routingkey分发到指定队列 
  topicexchange:多关键字匹配 
  */ 
 @bean 
 public directexchange defaultexchange() { 
  return new directexchange(exchange, true, false);
 } 

 /**
  * 配置消息队列2
  * 针对消费者配置 
  * @return
  */
 @bean
 public queue queue() { 
  return new queue("delay_queue2", true); //队列持久 

 }
 /**
  * 将消息队列2与交换机绑定
  * 针对消费者配置 
  * @return
  */
 @bean 
 @autowired
 public binding binding() { 
  return bindingbuilder.bind(queue()).to(defaultexchange()).with(delayqueue.routingkey2); 
 } 

 /**
  * 接受消息的监听，这个监听会接受消息队列1的消息
  * 针对消费者配置 
  * @return
  */
 @bean 
 @autowired
 public simplemessagelistenercontainer messagecontainer2(connectionfactory connectionfactory) { 
  simplemessagelistenercontainer container = new simplemessagelistenercontainer(connectionfactory()); 
  container.setqueues(queue()); 
  container.setexposelistenerchannel(true); 
  container.setmaxconcurrentconsumers(1); 
  container.setconcurrentconsumers(1); 
  container.setacknowledgemode(acknowledgemode.manual); //设置确认模式手工确认 
  container.setmessagelistener(new channelawaremessagelistener() {

   public void onmessage(message message, com.rabbitmq.client.channel channel) throws exception {
    byte[] body = message.getbody(); 
    system.out.println("delay_queue2 收到消息 : " + new string(body)); 
    channel.basicack(message.getmessageproperties().getdeliverytag(), false); //确认消息成功消费 
   } 
  }); 
  return container; 
 } 
}

在消息监听中处理需要定时处理的任务就好了，因为rabbitmq能发送消息，所以可以把任务特征码发过来，比如关闭订单就把订单id发过来，这样就避免了需要查询一下那些订单需要关闭而加重mysql负担了，毕竟一旦订单量大的话，查询本身也是一件很费io的事情

总结

基于rabbitmq实现定时任务，就是将消息设置一个过期时间，放入一个没有读取的队列中，让消息过期后自动转入另外一个队列中，监控这个队列消息的监听处来处理定时任务具体的操作

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。