RabbitMQ没有延时队列？学会这一招玩转延时队列-低调大师

RabbitMQ没有延时队列？学会这一招玩转延时队列

2021-03-03 465

什么是延时队列

延时队列：顾名思义，是一个用于做消息延时消费的队列。但是它也是一个普通队列，所以它具备普通队列的特性，相比之下，延时的特性就是它最大的特点。所谓的延时就是将我们需要的消息，延迟多久之后被消费。普通队列是即时消费的，延时队列是根据延时时间，多久之后才能消费的。

延时队列使用场景

订单在十分钟之内未支付则自动取消。
会员续费的定时推送
用户注册成功后，如果三天内没有登陆则进行短信提醒。
预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。
优惠券过期提醒

核心的应用内容基本都是基于需要设定过期时间的

RabbitMQ如何实现延时队列

方式1、通过RabbitMQ的高级特性TTL和配合死信队列
方式2、安装rabbitmq_delayed_message_exchange插件

RabbitMQ中的高级特性TTL

TTL是什么呢？TTL是RabbitMQ中一个消息或者队列的属性，表明一条消息或者该队列中的所有消息的最大存活时间，单位是毫秒，为什么延时队列要介绍它？TTL就是一种消息过期策略。给我们的消息做过期处理，当消息在队列中存活了指定时候之后，改队列就会将这个消息直接丢弃。在RabbitMQ中并没有直接实现好的延时队列，我们可以使用TTL这种高级特性，然后配合死信队列，即可实现延时队列的功能。

那么，如何设置这个TTL值呢？有两种方式，第一种是在创建队列的时候设置队列的“x-message-ttl”属性，如下：方式一：

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 6000);
channel.queueDeclare(queueName, durable, exclusive, autoDelete, args);

使用这种方式，消息被设定TTL，一旦消息过期，就会被队列丢弃

方式二：

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
builder.expiration("6000");
AMQP.BasicProperties properties = builder.build();
channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());

使用这种方式，消息即使过期，也不一定会被马上丢弃，因为消息是否过期是在即将投递到消费者之前判定的，如果当前队列有严重的消息积压情况，则已过期的消息也许还能存活较长时间。

另外，还需要注意的一点是，如果不设置TTL，表示消息永远不会过期，如果将TTL设置为0，则表示除非此时可以直接投递该消息到消费者，否则该消息将会被丢弃。

RabbitMQ到底怎么实现延时队列

步骤一：创建一个正常的队列，指定消息过期时间，并且指定消息过期后需要投递的死信交换器和死信交换队列。
步骤二：创建死信队列和死信交换器

RabbitMQ实现延时队列实例

package com.example.demo;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

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

/**
 * @author echo
 * @date 2021-01-14 14:35
 */
public class TopicDealProductTest {

    /**
     * 延时队列交换机
     */
    private static final String DIRECT_EXCHANGE_DELAY = "dir_exchange_delay";
    /**
     * 死信队列交换机
     */
    private static final String DIRECT_EXCHANGE_DEAD = "dir_exchange_dead";
    /**
     * 延时队列
     */
    private static final String DIRECT_QUEUE_DELAY = "dir.queue.delay";
    /**
     * 死信队列
     */
    private static final String DIRECT_QUEUE_DEAD = "dir.queue.dead";
    /**
     * 延时队列ROUTING_KEY
     */
    private static final String DIRECT_DELAY_ROUTING_KEY = "delay.queue.routingKey";
    /**
     * 延时队列ROUTING_KEY
     */
    private static final String DIRECT_DEAD_ROUTING_KEY = "dead.queue.routingKey";
    private static final String IP_ADDRESS = "192.168.230.131";
    private static final int PORT = 5672;


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = createConnection();
        // 创建一个频道
        Channel channel = connection.createChannel();
        sendMsg(channel);
        Thread.sleep(10000);
        closeConnection(connection, channel);
    }

    private static void sendMsg(Channel channel) throws IOException {

        // 创建延时队列和延时交换器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DELAY, BuiltinExchangeType.DIRECT);
        Map<String, Object> map = new HashMap<>(16);
        // 在延时交换器上指定死信交换器
        map.put("x-dead-letter-exchange", DIRECT_EXCHANGE_DEAD);
        // 在延时交换器上指定死信队列的routing-key
        map.put("x-dead-letter-routing-key", DIRECT_DEAD_ROUTING_KEY);
        // 设定延时队列的延长时长 10s
        map.put("x-message-ttl", 10000);
        // 创建延时队列
        channel.queueDeclare(DIRECT_QUEUE_DELAY, true, false, false, map);
        // 在延时交换器上绑定延时队列
        channel.queueBind(DIRECT_QUEUE_DELAY, DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY);

        // 创建死信队列和死信交换器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DEAD, BuiltinExchangeType.TOPIC, true, false, null);
        // 创建死信队列
        channel.queueDeclare(DIRECT_QUEUE_DEAD, true, false, false, null);
        // 在死信交换器上绑定死信队列
        channel.queueBind(DIRECT_QUEUE_DEAD, DIRECT_EXCHANGE_DEAD, DIRECT_DEAD_ROUTING_KEY);

        channel.basicPublish(DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY, null, "hello world".getBytes());

    }

    private static void closeConnection(Connection connection, Channel channel) throws IOException, TimeoutException {
        // 关闭资源
        channel.close();
        connection.close();
    }

    private static Connection createConnection() throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置RabbitMQ的链接参数
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername("echo");
        factory.setPassword("123456");
        // 和RabbitMQ建立一个链接
        return factory.newConnection();
    }

}

到这里，其实我们不难发现，我们无非是利用了TTL这个特性，让消息在过期的时候丢弃到指定队列，死信队列其实也是一个普通队列。

执行之后，我们来看看结果，在Exchange里面，我们创建了两个交换器和两个队列，但是两个队列和交换器还是有区别的，我们来看图片

我们可以看到两个队列的Features标志是不一样的

TTL: 消息在队列中的过期时间
DLX: 该队列绑定了死信交换器
DLK: 该队列绑定的死信队列的ROUTING_KEY

在我们执行完成只有，我们可以看到，消息先被投递到了delay，该队列里面的消息，到达过期时间之后就被投递到了dead队列中去了。

那么我们上面介绍了TTL和设置AMQP.BasicProperties，这两种有一定的区别，前一个是设置队列消息过期时间，后一个是设定每条消息的过期时间。那他们的区别在哪里？

设置每条消息和设置TTL的区别

其实这两种方式的区别就在于怎么判断该消息是否要被丢弃。TTL设定的队列，只要消息到达过期时间，立马就会将消息丢弃。如果是后者，可能我们队列里面有很多的消息，然后每条消息的过期时间又不一致，这个时候，如果队列出口处堵了很多没有设定过期时间的消息又不被消费的时候，队列后面的消息及时设定了过期时间也不会被丢弃，只有在设定了过期时间的消息到了队列该消费的位置，才会判定

怎么使用AMQP.BasicProperties？

package com.example.demo;

import com.rabbitmq.client.*;

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

/**
 * @author echo
 * @date 2021-01-14 14:35
 */
public class TopicDealProductTest {

    /**
     * 延时队列交换机
     */
    private static final String DIRECT_EXCHANGE_DELAY = "dir_exchange_delay";
    /**
     * 死信队列交换机
     */
    private static final String DIRECT_EXCHANGE_DEAD = "dir_exchange_dead";
    /**
     * 延时队列
     */
    private static final String DIRECT_QUEUE_DELAY = "dir.queue.delay";
    /**
     * 死信队列
     */
    private static final String DIRECT_QUEUE_DEAD = "dir.queue.dead";
    /**
     * 延时队列ROUTING_KEY
     */
    private static final String DIRECT_DELAY_ROUTING_KEY = "delay.queue.routingKey";
    /**
     * 延时队列ROUTING_KEY
     */
    private static final String DIRECT_DEAD_ROUTING_KEY = "dead.queue.routingKey";
    private static final String IP_ADDRESS = "192.168.230.131";
    private static final int PORT = 5672;


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = createConnection();
        // 创建一个频道
        Channel channel = connection.createChannel();
        sendMsg(channel);
        Thread.sleep(10000);
        closeConnection(connection, channel);
    }

    private static void sendMsg(Channel channel) throws IOException {

        // 创建延时队列和延时交换器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DELAY, BuiltinExchangeType.DIRECT);
        Map<String, Object> map = new HashMap<>(16);
        // 在延时交换器上指定死信交换器
        map.put("x-dead-letter-exchange", DIRECT_EXCHANGE_DEAD);
        map.put("x-dead-letter-routing-key", DIRECT_DEAD_ROUTING_KEY);
        // 设定延时队列的延长时长 10s
//        map.put("x-message-ttl", 10000);
        // 创建延时队列
        channel.queueDeclare(DIRECT_QUEUE_DELAY, true, false, false, map);
        // 在延时交换器上绑定延时队列
        channel.queueBind(DIRECT_QUEUE_DELAY, DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY);

        // 创建死信队列和死信交换器
        channel.exchangeDeclare(DIRECT_EXCHANGE_DEAD, BuiltinExchangeType.TOPIC, true, false, null);
        // 创建死信队列
        channel.queueDeclare(DIRECT_QUEUE_DEAD, true, false, false, null);
        // 在死信交换器上绑定死信队列
        channel.queueBind(DIRECT_QUEUE_DEAD, DIRECT_EXCHANGE_DEAD, DIRECT_DEAD_ROUTING_KEY);

        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        builder.expiration("10000");
        AMQP.BasicProperties properties = builder.build();
        channel.basicPublish(DIRECT_EXCHANGE_DELAY, DIRECT_DELAY_ROUTING_KEY, false, properties,  "hello world".getBytes());

    }

    private static void closeConnection(Connection connection, Channel channel) throws IOException, TimeoutException {
        // 关闭资源
        channel.close();
        connection.close();
    }

    private static Connection createConnection() throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置RabbitMQ的链接参数
        factory.setHost(IP_ADDRESS);
        factory.setPort(PORT);
        factory.setUsername("echo");
        factory.setPassword("123456");
        // 和RabbitMQ建立一个链接
        return factory.newConnection();
    }

}

我们运行完成成之后，可以看到和我们之前那一种方式的效果是一样的

两种设定过期时间的方式其实区别就在于一个统一设定了过期时间，一个指定每条过期时间。但是并不影响我们延时队列的实现，那我们怎么选择呢？

怎么选择TTL设定方式？

根据两种方式的特性来选定使用场景才是最合理的。我们如果用来做延时队列的，想将延时队列的特性应用到实际场景的，并且对时时性要求比较高的，建议选择第一种方式。

总结

延时队列的实现并不难，关键是我们要知道他的一个原理，了解RabbitMQ他的TTL和死信对了。掌握了它的这些特性之后，我们就可以很好的应用延时队列。延时队列在工作中对我们的帮组也非常大，不过RabbiTMQ没有原生延时队列，我们用这种方式实现了它并不意味着我们一定要选择它。其实还有很多的方式，比如Java中的DelayQueu、kafka的时间轮等。

微信关注我们

原文链接：https://my.oschina.net/u/4154037/blog/4969222

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

CTF 一次PWN解题的小技巧

前几天在国外的某个ctf社区发现了一道好玩的赛题。建议ctfer在阅读这篇文章的时候，首先要掌握以下的一些内容，因为这些东西对于ctf比赛来说，都是很有必要掌握的。基本的Linux知识对于X86有基本的了解了解堆栈工作原理 C语言的基本知识了解缓冲区溢出漏洞的原理基本的python开发能力本文涉及知识点实操练习：PWN综合练习（一）（CTF PWN进阶训练实战，尝试溢出一个URL解码程序。）计算机字节和shell的魅力我第一次接触到网络安全编程时，我就已经发现开发其实是一门艺术，并且研究员需要很高的水平才能胜任这一工作，因为很多时候你需要各种领域的知识进行结合。例如。C语言、汇编、堆栈、利用python进行漏洞开发利用等，当然，这些只是冰山一角。对于一个刚进入漏洞利用开发领域的小白来说，这可能是很难的，因为这是一个从零开始进行积累的过程。作为一名小菜鸡，我决定先挑战这个网站的最简单的pwn赛题。别问为什么，问就是别的试题太难了。进入正题让我们先进入SSH！ ssh [tiny_easy@pwnable.kr](mailto:tiny_easy@pwnable...

2021-03-03

688

PHPMQTT v1.3.1 已经发布，这是一个 MQTT 协议解析 & 协程客户端。此版本更新内容包括：增强增加 isMQTT5 方法 (f27f85c) 增加常量可见性 (#41) 为 getContents 方法添加 getArray 参数来用于客户端回复对端 ACK (#42) 添加 Message 测试 (8f7fe30) 更新 Message 文档 (a41654b) 更新 ProtocolException 并增加测试 (19a6bee) 修复修复遗嘱消息的 topic 不能为空 (b56fda1) 详情查看：https://gitee.com/phpiot/mqtt/releases/v1.3.1

2021-03-03

665

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。