一元云购网站黑客攻击,手机家装绘图软件,石家庄区号,邢台网约车平台有哪些系列文章目录
手把手教你#xff0c;本地RabbitMQ服务搭建#xff08;windows#xff09; 消息队列选型——为什么选择RabbitMQ RabbitMQ灵活运用#xff0c;怎么理解五种消息模型 RabbitMQ 能保证消息可靠性吗 推或拉#xff1f; RabbitMQ 消费模式该如何选择 死信是什么…系列文章目录
手把手教你本地RabbitMQ服务搭建windows 消息队列选型——为什么选择RabbitMQ RabbitMQ灵活运用怎么理解五种消息模型 RabbitMQ 能保证消息可靠性吗 推或拉 RabbitMQ 消费模式该如何选择 死信是什么如何运用RabbitMQ的死信机制 真的好用吗鲜有人提的 RabbitMQ-RPC模式 如何利用RabbitMQ的延迟队列提高消息处理效率 系列文章目录一、什么是延迟队列二、延迟队列的实现1. x-delayed-message插件2. TTL 死信队列 三、手写延时队列1. 时间轮概念2. JAVA演示 四、应用场景与注意事项1. 应用场景2. 注意事项 总结 前面我们讲到了RabbitMQ的死信队列其实除了死信队列RabbitMQ还有一个常用的延迟队列设计。今天我们就来说一下这个延迟队列 作者简介战斧从事金融IT行业有着多年一线开发、架构经验爱好广泛乐于分享致力于创作更多高质量内容 本文收录于 RabbitMQ 有需要者可直接订阅专栏实时获取更新 高质量专栏 云原生、RabbitMQ、Spring全家桶 等仍在更新欢迎指导 Zookeeper Redis kafka docker netty等诸多框架以及架构与分布式专题即将上线敬请期待 提示以下是本篇文章正文内容下面案例可供参考
一、什么是延迟队列
延迟队列指的是当我们将消息发送到RabbitMQ时可以指定消息的有效期或者消息需要在未来某个时间点才能被消费。这种消息被称为“延迟消息”。因此RabbitMQ支持通过延迟队列来实现延迟消息的发送和消费。
二、延迟队列的实现
延迟队列的实现原理其实就是将消息放入到一个普通的队列中只不过这个队列有一个特殊的属性消息的消费被延迟一段时间。这个延迟时间可以是任意的也可以是固定的。当消息进入队列时会有一个定时器在计时当计时器到达设定的时间时消息会被转移至消费队列等待被消费。
在RabbitMQ中延迟队列的实现有两种方式一种是通过x-delayed-message插件实现另一种是通过TTLTime To Live和死信队列实现。
1. x-delayed-message插件
x-delayed-message插件可以让RabbitMQ支持延迟消息功能它是一个非官方插件需要自行下载并安装。其源码地址如下github地址 或 gitee地址如果你是从笔者之前的安装博客 手把手教你本地RabbitMQ服务搭建windows 过来的那么你用的可能是RabbitMQ V3.12可以直接下载我上传的资源 3.12-插件
首先需要在RabbitMQ服务器上安装x-delayed-message插件。把上述的插件复制进我们RabbitMQ的服务插件目录下 然后执行插件的启用 rabbitmq-plugins enable rabbitmq_delayed_message_exchange 即可 然后在Java代码中定义queue、exchange和connectionFactory代码如下
ConnectionFactory connectionFactory new ConnectionFactory();
connectionFactory.setHost(HOST_NAME);
connectionFactory.setUsername(USERNAME);
connectionFactory.setPassword(PASSWORD);
connectionFactory.setPort(PORT);Connection connection connectionFactory.newConnection();
Channel channel connection.createChannel();MapString, Object arguments new HashMapString, Object();
arguments.put(x-delayed-type, direct);
channel.exchangeDeclare(delayed_exchange, x-delayed-message, true, false, arguments);
channel.queueDeclare(delayed_queue, true, false, false, null);
channel.queueBind(delayed_queue, delayed_exchange, delayed_routing_key);
不难发现此时其实是交换机在做延迟
当然除了交换机的设置在发送消息时还需要在消息头部设置x-delay属性代码如下
AMQP.BasicProperties.Builder builder new AMQP.BasicProperties.Builder();
builder.deliveryMode(2);
builder.headers(new HashMapString, Object(){{put(x-delay, 5000);}});
AMQP.BasicProperties properties builder.build();
channel.basicPublish(delayed_exchange, delayed_routing_key, properties, message.getBytes());
2. TTL 死信队列
此种方式的原理其实我们在学习死信队列的时候应该就察觉到了就是利用消息超时TTL后会转入死信交换机的机制其模型如下
首先需要在Java代码中定义queue、exchange和connectionFactory代码如下
ConnectionFactory connectionFactory new ConnectionFactory();
connectionFactory.setHost(HOST_NAME);
connectionFactory.setUsername(USERNAME);
connectionFactory.setPassword(PASSWORD);
connectionFactory.setPort(PORT);Connection connection connectionFactory.newConnection();
Channel channel connection.createChannel();MapString, Object arguments new HashMapString, Object();
arguments.put(x-dead-letter-exchange, dead_letter_exchange);
arguments.put(x-dead-letter-routing-key, dead_letter_routing_key);
arguments.put(x-message-ttl, 5000);channel.exchangeDeclare(normal_exchange, direct, true, false, null);
channel.exchangeDeclare(dead_letter_exchange, direct, true, false, null);
channel.queueDeclare(normal_queue, true, false, false, arguments);
channel.queueDeclare(dead_letter_queue, true, false, false, null);
channel.queueBind(normal_queue, normal_exchange, normal_routing_key);
channel.queueBind(dead_letter_queue, dead_letter_exchange, dead_letter_routing_key);
在发送消息时只需要将消息发送到normal_exchange交换机下代码如下
channel.basicPublish(normal_exchange, normal_routing_key, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());三、手写延时队列
当然除了RabbitMQ实现延时队列的方式还有很多我们甚至可以自己实现本节我们就尝试自己写个延时队列
1. 时间轮概念
在关于计时或定时的设计里时间轮是一种用于处理定时任务的数据结构。它通过将时间划分为一系列的时刻每个时刻对应一个槽将任务存储在相应的槽中 时间轮通常包含多个槽和指针其中指针指向当前时刻对应的槽每过单位时间指针就指向下一个槽这样任务调度时按照指针的移动依次执行槽中的任务
2. JAVA演示
我们先使用JUC相关内容实现一个时间轮
import java.util.*;
import java.util.concurrent.*;class TimeWheel {private int size;private int currentIndex;private ListBlockingQueueTask slots;private Executor executor;public TimeWheel(int size, Executor executor) {this.size size;this.slots new ArrayList(size);for (int i 0; i size; i) {slots.add(new LinkedBlockingQueue());}this.executor executor;}public void addTask(Task task) {int expireIndex (int)(currentIndex task.getDelay() / 1000) % size;slots.get(expireIndex).add(task);}public void start() {new Thread(() - {while (true) {currentIndex (currentIndex 1) % size;BlockingQueueTask currentSlot slots.get(currentIndex);ListTask tasks new ArrayList();currentSlot.drainTo(tasks);for (Task task : tasks) {executor.execute(task);}try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}}).start();}
}class Task implements Runnable {private long delay; // 延迟时间单位毫秒private Runnable task; // 任务public Task(long delay, Runnable task) {this.delay delay;this.task task;}public long getDelay() {return delay;}Overridepublic void run() {task.run();}
}
我们可以使用main方法来尝试验证这个时间轮效果 public static void main(String[] args) {TimeWheel timeWheel new TimeWheel(60 * 60, Executors.newFixedThreadPool(10));// 添加任务延迟5秒执行timeWheel.addTask(new Task(5000, () - System.out.println(Task 1 executed!)));// 添加任务延迟10秒执行timeWheel.addTask(new Task(10000, () - System.out.println(Task 2 executed!)));// 启动时间轮timeWheel.start();}当然以上代码只是一个简化的实现实际情况中需要考虑任务执行时间和时间轮的精度等问题。
四、应用场景与注意事项
1. 应用场景 红包预告 在现在的抢红包的场景下当用户发起红包活动后可能不希望立即开抢而是设定在一段时间后开启。那么我们可以将将红包信息发送到一个延迟队列中一定时间后系统会自动激活红包此时用户才可以真正抢红包 订单系统 在订单系统中有一些订单需要在未来某个时间点才能被处理。例如有些订单需要在一定的时间之后才能发货或者确认收货。这时候我们可以将这些订单放到延迟队列中当时间到达时再进行处理。 优惠券系统 在优惠券系统中有一些优惠券需要在未来某个时间点才能使用。这时候我们可以将这些优惠券放到延迟队列中当时间到达时再进行激活。
2. 注意事项 延迟队列不要使用太多 使用延迟队列可以在一定程度上减少系统的负载但是使用过多的延迟队列会导致系统变得更加复杂维护起来也更加困难。 延迟队列可能会导致消息丢失 在RabbitMQ中当一个带有TTL消息被发送到队列中时如果队列中的消息太多或者队列的消费者速度太慢就会导致消息失效如果没有使用死信机制消息就会被丢失。为了避免这种情况发生我们需要对队列进行监控及时发现问题并进行处理。 设置合适的延迟时间 在使用延迟队列时需要根据实际需求设置合适的延迟时间。如果延迟时间太短可能会导致消息延迟效果不明显如果延迟时间太长可能会导致系统累积大量的消息导致负载过高。
总结
RabbitMQ的延迟队列是一种非常实用的特性可以帮助我们实现定时任务、限流、削峰等功能。但是在使用延迟队列时需要谨慎对待根据实际需求设置合适的延迟时间并及时监控队列中的消息避免出现消息丢失的情况。
