基于消息队列（RabbitMQ）实现延迟任务

Posted Lydia Bess

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了基于消息队列（RabbitMQ）实现延迟任务相关的知识，希望对你有一定的参考价值。

一、序言

延迟任务应用广泛，延迟任务典型应用场景有订单超时自动取消；支付回调重试。其中订单超时取消具有幂等性属性，无需考虑重复消费问题；支付回调重试需要考虑重复消费问题。

延迟任务具有如下特点：在未来的某个时间点执行；一般仅执行一次。

1、实现原理

生产者将带有延迟信息的消息发送到RabbitMQ交换机中，等待延迟时间结束方将消息转发到绑定的队列中，消费者通过监听队列消费消息。延迟任务的关键在消息在交换机中停留。

显而易见，基于RabbitMQ实现延迟任务对服务器的可靠性要求极高，交换机内部消息无持久化机制，比如单机模式服务重启，未开始的延迟任务均丢失。

2、组件选型

二、方案设计

（一）服务器

RabbitMQ服务需要安装x-delayed-message插件以处理延迟消息。

（二）生产者

延迟任务的实现对生产者的要求是将消息可靠的投递到交换机，因此使用confirm确认机制即可。

订单生成之后，先入库，然后以订单ID为key将订单详情存入Redis中（持久化），向RabbitMQ发送异步confirm确定请求。如果收到正常投递返回，则删除Redis中订单ID为key的数据，回收内存，否则以订单ID为key，从Redis中查询出订单数据，重新发送。

（三）消费者

延迟任务的实现对消费者的要求是以信息不丢失的方式消费消息，具体表现在：手动确认消息的消费，防止消息丢失；消费端持续稳定，防止消息堆积；消息消费失败有重试机制。

考虑到订单延迟取消属于幂等性操作，因此无需考虑消息的重复消费问题。

三、SpringBoot实现

实现部分仅贴一部分核心源码，完整项目请访问GitHub。

（一）生产者

考虑到下单是极为重要的操作，因此首先将订单落库、存盘，然后进行后续操作。

for (long i = 1; i <= 10; i++) 
    /* 1.模拟生成订单 */
    BuOrder order = createOrder(i);
    /* 2.订单入库 */
    orderService.removeById(order);
    orderService.saveOrUpdate(order);
    /* 3.将订单存入信息Redis */
    RedisUtils.setObject(RabbitTemplateConfig.ORDER_PREFIX + i, order);
    /* 4.向RabbitMQ异步投递消息 */
    rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));

生产者可靠投递消息

public void confirm(CorrelationData correlationData, boolean ack, String cause) 
    if (correlationData == null) 
        return;
    
    String key = ORDER_PREFIX + correlationData.getId();
    if (ack) 
        /* 如果消息投递成功，则删除Redis中订单数据，回收内存 */
        RedisUtils.deleteObject(key);
     else 
        /* 从Redis中读取订单数据，重新投递 */
        BuOrder order = RedisUtils.getObject(key, BuOrder.class);
        /* 重新投递消息 */
        rabbitTemplate.convertAndSend(RabbitmqConfig.DELAY_EXCHANGE_NAME, RabbitmqConfig.DELAY_KEY, order, RabbitUtils.setDelay(30000), RabbitUtils.correlationData(order.getOrderId()));

（二）消费者

消费者端手动确认，避免消息丢失；失败自动重试。

@RabbitListener(queues = RabbitmqConfig.DELAY_QUEUE_NAME)
public void consumeNode01(Channel channel, Message message, BuOrder order) throws IOException 
    if (Objects.equals(0, order.getOrderStatus())) 
        /* 修改订单状态，设置为关闭状态 */
        orderService.updateById(new BuOrder(order.getOrderId(), -1));
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        log.info(String.format("消费者节点01消费编号为【%s】的消息", order.getOrderId()));

消费者可靠消费应至少开启两个及以上应用，确保消息队列中不积压消息。

（三）通用工具包

上述代码涉及一个工具类RabbitUtils，存在于如下依赖中，主要封装RabbitMQ极常用的工具方法。

<dependency>
  <groupId>xin.altitude.cms</groupId>
  <artifactId>ucode-cms-common</artifactId>
  <version>1.4.3.1</version>
</dependency>

愿与诸君共进步，大量的面试题及答案还有资深架构师录制的视频录像：有Spring，MyBatis，Netty源码分析，高并发、高性能、分布式、微服务架构的原理，JVM性能优化、分布式架构等这些成为架构师必备的知识体系
可以关注小编+转发文章+私信【架构资料】获取，最后祝大家都能拿到自己心仪的offer

RabbitMQ 延迟任务（限时订单） 原理 以及代码 实战

  • 一、场景

　　　　我们经常会碰见，一个需求就是，发送一条指令（消息），延迟一段时间执行，比如说常见的淘宝当下了一个订单后，订单支付时间为半个小时，如果半个小时没有支付，则关闭该订单。当然实现的方式有几种，今天来看看rabbitMQ实现的方式。

  • 二、思路：rabbitMQ 如何实现 

  1. rabbitMQ为每个队列设置消息的超时时间。只要给队列设置x-message-ttl 参数，就设定了该队列所有消息的存活时间，时间单位是毫秒。如果声明队列时指定了死信交换器，则过期消息会成为死信消息

  2. 需要设置的参数为：

    技术分享图片技术分享图片

  • 三、原理：上图

    技术分享图片

  1. 将延迟队列（queue）在声明的时候设置参数 “ x-dead-letter-exchange ”，“ x-message-ttl “ 分别对应 死信路由器（dlx_exchange） 和 消息过期时间(比如说30分钟)。

  2.  一个消息从生产者发送到延迟队列 ，在延迟队列里等待，等待30分钟后，会去绑定的死信路由（dlx_exchange）。通过死信路由的规则，走到死信队列。

  3. 这时候监听死信队列的消费者就可以接收到消息，消费消息。比如说查看该订单是否支付，如果没有支付，则关闭该订单。

  • 四、代码实战

  1. 生产者代码：
    import com.dbg.example.connectionFactory.RQconnFactory;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;


public class DelayProducer {

        public static Logger logger = LoggerFactory.getLogger(DelayProducer.class);

        // 交换机
        public static final String exchangeName = "delay_Task";
        // 声明路由键
        public static final String routekey = "delay_order";

        public static void  pend (List<String> orderNos) throws IOException, TimeoutException {

                // 通过单例得到工厂
                ConnectionFactory connectionFactory = RQconnFactory.getrQconnFactory();
                Connection connection = connectionFactory.newConnection();
                final Channel channel = connection.createChannel();

                // 声明一个交换机
                channel.exchangeDeclare(exchangeName , BuiltinExchangeType.DIRECT);

                // 消息绑定
                for (String orderNo :orderNos ) {
                    channel.basicPublish(exchangeName , routekey , null ,orderNo.getBytes());
                    logger.info("发送订单 : "+ orderNo);
                }
                // 关闭频道和连接
                channel.close();
                connection.close();
        }

        public static void main(String[] args) {
                List orders = new ArrayList();
                orders.add("order 0001");
                orders.add("order 0002");
                orders.add("order 0003");
                try {
                        pend(orders);
                } catch (IOException e) {
                        e.printStackTrace();
                } catch (TimeoutException e) {
                        e.printStackTrace();
                }
        }


}

     

  2. 消费者代码：
    import com.dbg.example.connectionFactory.RQconnFactory;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;


public class DelayConsumer {

    public static void receive() throws IOException, TimeoutException {

        // 通过单例得到工厂
        ConnectionFactory connectionFactory = RQconnFactory.getrQconnFactory();
        Connection connection = connectionFactory.newConnection();
        final Channel channel = connection.createChannel();

        // 声明一个死信路由
        String dlxExchangeName = "dlx_delay";
        channel.exchangeDeclare(dlxExchangeName , BuiltinExchangeType.TOPIC);
        // 声明一个死信订单队列
        String dlxQueueName = "dlx_delay_order";
        channel.queueDeclare( dlxQueueName ,true , true ,false ,null);
        // 绑定
        channel.queueBind(dlxQueueName , dlxExchangeName , "#");

        // 声明一个交换机
        channel.exchangeDeclare(DelayProducer.exchangeName , BuiltinExchangeType.DIRECT);
        // 声明一个延迟订单队列 ,并绑定死信路由器
        String queueName = "delay_order";
        Map<String ,Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange",dlxExchangeName);
        arguments.put("x-message-ttl", 5 * 1000);  // 5秒过期时间
        channel.queueDeclare(queueName,true , false ,false , arguments );
        // 绑定
        channel.queueBind(queueName , DelayProducer.exchangeName ,DelayProducer.routekey);

        // 声明一个消费者（关闭为支付订单的服务）
        final Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received and close [" +envelope.getRoutingKey() +"]"+message);
            }
        };

        // 消费者消费--队列()
        channel.basicConsume(dlxQueueName,true , consumer);

    }


    public static void main(String[] args) {
        try {
            receive();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }

}

    3.得到单例连接工厂

    public class RQconnFactory {

    private volatile static ConnectionFactory rQconnFactory;

    private RQconnFactory(){
        
    }

    public static ConnectionFactory getrQconnFactory() {
        if (rQconnFactory == null) {
            synchronized (RQconnFactory.class){
                if (rQconnFactory  == null ){
                    rQconnFactory = new ConnectionFactory();
                    rQconnFactory.setHost("192.168.31.220");
                    return rQconnFactory;
                }
            }
        }
        return rQconnFactory;
    }
}

     





以上是关于基于消息队列（RabbitMQ）实现延迟任务的主要内容，如果未能解决你的问题，请参考以下文章

rabbitmq的延迟消息队列实现

RabbitMQ面试题：啥是延迟队列？ --- 2022-04-03

RabbitMQ 延迟任务（限时订单） 原理 以及代码 实战

RabbitMQ 实现延迟队列的两种方式

rabbitmq死信队列及延迟队列

基于rabbitMQ 消息延时队列方案 模拟电商超时未支付订单处理场景