RabbitMQ对延迟任务的两种实现方式及利弊

标签：PRP springframework 利弊 RabbitMQ org import ASSET public 延迟

1.开门见山两种方式分别为原生的死信队列与延迟插件安装而来的延迟交换机功能

2.死信队列原理不再概述，主要实现方式就是先扔入普通消息队列且设置消息的过期时间，一旦消息过期即进入死信队列。此时监听死信队列的消息即为消息的延迟消费；

　　队列-交换机配置类

package com.fawkes.cybereng.asset.mq.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.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description 计划修-任务最大提前时间及最大延期时间的死信队列配置
 * @Author 薛铁琪
 * @CreateTime 2022/8/22 10:27
 * @Version 1.0
 */
@Configuration
public class RabbitMQConfiguration {
    //队列名称
    public final static String CYBERENG_ASSET_PRP_TASK_QUEUE = "cybereng-asset-prp-task-queue";

    //交换机名称
    public final static String CYBERENG_ASSET_PRP_TASK_EXCHANGE = "cybereng-asset-prp-task-exchange";

    // routingKey
    public final static String CYBERENG_ASSET_PRP_TASK_ROUTINGKEY = "cybereng-asset-prp-task-routingkey";

    //死信消息队列名称
    public final static String CYBERENG_ASSET_PRP_TASK_QUEUE_DEAD = "cybereng-asset-prp-task-queue-dead";

    //死信交换机名称
    public final static String CYBERENG_ASSET_PRP_TASK_EXCHANGE_DEAD = "cybereng-asset-prp-task-exchange-dead";

    //死信 routingKey
    public final static String CYBERENG_ASSET_PRP_TASK_ROUTINGKEY_DEAD = "cybereng-asset-prp-task-routingkey-dead";

    //死信队列 交换机标识符
    public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";

    //死信队列交换机绑定键标识符
    public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Bean
    public Queue prpTaskQueue() {
        // 将普通队列绑定到死信队列交换机上
        Map<String, Object> args = new HashMap<>(2);
        args.put(DEAD_LETTER_QUEUE_KEY, CYBERENG_ASSET_PRP_TASK_EXCHANGE_DEAD);
        args.put(DEAD_LETTER_ROUTING_KEY, CYBERENG_ASSET_PRP_TASK_ROUTINGKEY_DEAD);
        return new Queue(RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_QUEUE, true, false, false, args);
    }

    //声明一个direct类型的交换机
    @Bean
    DirectExchange prpTaskExchange() {
        return new DirectExchange(RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_EXCHANGE);
    }

    //绑定Queue队列到交换机,并且指定routingKey
    @Bean
    Binding bindingDirectExchange() {
        return BindingBuilder.bind(prpTaskQueue()).to(prpTaskExchange()).with(CYBERENG_ASSET_PRP_TASK_ROUTINGKEY);
    }

    //创建配置死信队列
    @Bean
    public Queue prpTaskQueueDead() {
        Queue queue = new Queue(CYBERENG_ASSET_PRP_TASK_QUEUE_DEAD, true, false, false);
        return queue;
    }

    //创建死信交换机
    @Bean
    public DirectExchange prpTaskExchangeDead() {
        return new DirectExchange(CYBERENG_ASSET_PRP_TASK_EXCHANGE_DEAD);
    }

    //死信队列与死信交换机绑定
    @Bean
    public Binding bindingDeadExchange() {
        return BindingBuilder.bind(prpTaskQueueDead()).to(prpTaskExchangeDead()).with(CYBERENG_ASSET_PRP_TASK_ROUTINGKEY_DEAD);
    }


}

　　生产者

/**
 * COPYRIGHT HangZhou 99Cloud Technology Company Limited
 * All right reserved.
 */
package com.fawkes.cybereng.asset.mq.provider;

import com.fawkes.cybereng.asset.mq.config.RabbitMQConfiguration;
import com.fawkes.cybereng.asset.mq.message.PrpTaskTimeMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @Description
 * @Author 薛铁琪
 * @CreateTime 2022/8/22 10:27
 * @Version 1.0
 */
@Component
@Slf4j
public class PrpTaskTimeProvider {
    @Autowired
    RabbitTemplate rabbitTemplate;

    public void submit(PrpTaskTimeMessage prpTaskTimeMessage) {
        log.info("扔入队列的任务信息{}", prpTaskTimeMessage);
        this.rabbitTemplate.convertAndSend(
                //发送至订单交换机
                RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_EXCHANGE,
                //routingKey
                RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_ROUTINGKEY,
                prpTaskTimeMessage
                , message -> {
                    // 如果配置了 params.put("x-message-ttl", 5 * 1000);
                    // 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
                    // 自己计算下超时时间（秒放入）
                    Date current = new Date();
                    long ex = prpTaskTimeMessage.getPlanTimeAdvanceMax().getTime() - current.getTime();
                    log.info("延迟秒数{}", ex / 1000);
                    message.getMessageProperties().setExpiration(ex + "");
                    return message;
                });
    }
}

　　消费者

package com.fawkes.cybereng.asset.mq.consumer;

import com.fawkes.cybereng.asset.mq.config.RabbitMQConfiguration;
import com.fawkes.cybereng.asset.mq.message.PrpTaskTimeMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * @Description
 * @Author 薛铁琪
 * @CreateTime 2022/8/22 10:27
 * @Version 1.0
 */
@Component
@Slf4j
public class PrpTaskTimeConsumer {

    @RabbitListener(
            queues = RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_QUEUE_DEAD
            , ackMode = "MANUAL"
    )
    public void process(PrpTaskTimeMessage prpTaskTimeMessage, Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
        log.info("消费死信队列的任务信息{}", prpTaskTimeMessage);
        // TODO 判断消息类型 最大提前  最晚到期
        // TODO 任务类型是否为自动
        // 手动ack
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        // 手动签收
        channel.basicAck(deliveryTag, false);
    }
}

　　优：利用rabbitmq本身机制，不需要额外安装插件，性能好

　　缺：一个致命的问题就是消息顺序，不会按照延迟时间的先后顺序输出，而是按照queue本身先进先出的规则。即10秒延迟的消息如果是在20秒延迟消息后扔入的，那么也要等20秒延迟的消息输出后才能输出。除非消息的延迟时间是一致的否则无法满足业务要求。

3.延迟插件顾名思义，直接实现了延迟扔入队列的功能；

　　配置类

package com.fawkes.cybereng.asset.mq.config;

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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * @Description 计划修-任务最大提前时间及最大延期时间的死信队列配置
 * @Author 薛铁琪
 * @CreateTime 2022/8/22 10:27
 * @Version 1.0
 */
@Configuration
public class RabbitMQConfiguration {
    //队列名称
    public final static String CYBERENG_ASSET_PRP_TASK_QUEUE = "cybereng-asset-prp-task-queue";

    //交换机名称
    public final static String CYBERENG_ASSET_PRP_TASK_EXCHANGE = "cybereng-asset-prp-task-exchange";

    // routingKey
    public final static String CYBERENG_ASSET_PRP_TASK_ROUTINGKEY = "cybereng-asset-prp-task-routingkey";

    /**
     * 初始化延迟交换机
     *
     * @return
     */
    @Bean
    public CustomExchange delayedExchangeInit() {
        Map<String, Object> args = new HashMap<>();
        // 设置类型，可以为fanout、direct、topic
        args.put("x-delayed-type", "direct");
        // 第一个参数是延迟交换机名字，第二个是交换机类型，第三个设置持久化，第四个设置自动删除，第五个放参数
        return new CustomExchange(CYBERENG_ASSET_PRP_TASK_EXCHANGE, "x-delayed-message", true, false, args);
    }

    /**
     * 初始化队列
     *
     * @return
     */
    @Bean
    public Queue delayedQueueInit() {
        return new Queue(CYBERENG_ASSET_PRP_TASK_QUEUE, true, false, false);
    }


    /**
     * 队列绑定到交换机
     *
     * @param delayedSmsQueueInit
     * @param customExchange
     * @return
     */
    @Bean
    public Binding delayedBindingQueue(Queue delayedSmsQueueInit, CustomExchange customExchange) {
        return BindingBuilder.bind(delayedSmsQueueInit).to(customExchange).with(CYBERENG_ASSET_PRP_TASK_ROUTINGKEY).noargs();
    }

}

　　生产者

/**
 * COPYRIGHT HangZhou 99Cloud Technology Company Limited
 * All right reserved.
 */
package com.fawkes.cybereng.asset.mq.provider;

import com.fawkes.cybereng.asset.mq.config.RabbitMQConfiguration;
import com.fawkes.cybereng.asset.mq.message.PrpTaskTimeMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @Description
 * @Author 薛铁琪
 * @CreateTime 2022/8/22 10:27
 * @Version 1.0
 */
@Component
@Slf4j
public class PrpTaskTimeProvider {
    @Autowired
    RabbitTemplate rabbitTemplate;

    public void submit(PrpTaskTimeMessage prpTaskTimeMessage) {
        log.info("扔入队列的任务信息{}", prpTaskTimeMessage);
        this.rabbitTemplate.convertAndSend(
                //发送至订单交换机
                RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_EXCHANGE,
                //routingKey
                RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_ROUTINGKEY,
                prpTaskTimeMessage
                , message -> {
                    // 如果配置了 params.put("x-message-ttl", 5 * 1000);
                    // 那么这一句也可以省略,具体根据业务需要是声明 Queue 的时候就指定好延迟时间还是在发送自己控制时间
                    // 自己计算下超时时间（秒放入）
                    Date current = new Date();
                    Long ex = prpTaskTimeMessage.getPlanTimeAdvanceMax().getTime() - current.getTime();
                    log.info("延迟秒数{}", ex / 1000);
                    message.getMessageProperties().setDelay(ex.intValue());
                    return message;
                });
    }
}

　　消费者

package com.fawkes.cybereng.asset.mq.consumer;

import com.fawkes.cybereng.asset.mq.config.RabbitMQConfiguration;
import com.fawkes.cybereng.asset.mq.message.PrpTaskTimeMessage;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * @Description
 * @Author 薛铁琪
 * @CreateTime 2022/8/22 10:27
 * @Version 1.0
 */
@Component
@Slf4j
public class PrpTaskTimeConsumer {

    @RabbitListener(
            queues = RabbitMQConfiguration.CYBERENG_ASSET_PRP_TASK_QUEUE
            , ackMode = "MANUAL"
    )
    public void process(PrpTaskTimeMessage prpTaskTimeMessage, Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
        log.info("消费死信队列的任务信息{}", prpTaskTimeMessage);
        // TODO 判断消息类型 最大提前  最晚到期
        // TODO 任务类型是否为自动
        // 手动ack
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        // 手动签收
        channel.basicAck(deliveryTag, false);
    }
}

　　效果

　　利：解决了不通消息不通延迟时间的问题

　　弊：需要安装延迟插件，对rabbitmq重启有一定的风险。性能待观察！

标签：PRP,springframework,利弊,RabbitMQ,org,import,ASSET,public,延迟
From： https://www.cnblogs.com/xuetieqi/p/16613525.html

RabbitMQ对延迟任务的两种实现方式及利弊

相关文章

赞助商

阅读排行