RabbitMQ 延时交换机

简介: RabbitMQ 延时交换机

rabbitmq_delayed_message_exchange


640.png

机制

安装插件后会生成新的Exchange类型x-delayed-message,该类型消息支持延迟投递机制,接收到消息后并未立即将消息投递至目标队列中,而是存储在mnesia(一个分布式数据系统)表中,并且当前节点是磁盘节点,那么节点重启后,消息还能保留。检测消息延迟时间,如达到可投递时间时并将其通过x-delayed-type类型标记的交换机类型投递至目标队列。但是要注意的是,如果集群中只有一个磁盘节点,如果说磁盘节点丢失,或者节点上的插件失效。意味着消息将会丢失。


特性

  • 可通过 x-delayed-type 指定类型 为 direct fanout topic 等
  • 检测消息延迟时间,如果达到投递时间,通过x-delayed-type 标记的交换机类型进行投递。
  • 消息存储在声明交换机的那个节点上,消息发送到延时交换机上,消息还未到发送时间,此时停机该节点,消费者不能消费此延时消息,后启动该节点,消息会被重新投递,消费者能够消费此延时消息。
  • 磁盘节点,消息持久化,交换机持久化,消息不会丢失。


延时队列申明

Map<String,Object> map =new HashMap<>();
map.put("x-delayed-type", "direct");
channel.exchangeDeclare(exchangeName, "x-delayed-message", true, false, map);


插件安装


下载


https://www.rabbitmq.com/community-plugins.html


wget https://dl.bintray.com/rabbitmq/community-plugins/3.6.x/rabbitmq_delayed_message_exchange/rabbitmq_delayed_message_exchange-20171215-3.6.x.zip


解压

下载后解压,并将其拷贝至(使用Linux Debian/RPM部署)rabbitmq服务器目录:/usr/local/rabbitmq/plugins中(windows安装目录\rabbitmq_server-version\plugins )

/usr/lib/rabbitmq/lib/rabbitmq_server-3.6.12/plugins


unzip rabbitmq_delayed_message_exchange-20171215-3.6.x.zip


启用插件

使用命令rabbitmq-plugins enable rabbitmq_delayed_message_exchang启用插件

rabbitmq-plugins enable rabbitmq_delayed_message_exchang


输出结果

The following plugins have been enabled:
rabbitmq_delayed_message_exchange


代码示例


生产者

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setVirtualHost("/");
        connectionFactory.setHost("172.31.1.135");
        connectionFactory.setUsername("xx");
        connectionFactory.setPassword("xx");
        connectionFactory.setPort(5672);
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String exchangeName = "delay-exchange";
        String routingkey = "delay.delay";
        String queueName = "delay_queueName";
        //x-delayed-message 声明
        Map<String,Object> map =new HashMap<>();
        map.put("x-delayed-type", "direct");
        channel.exchangeDeclare(exchangeName, "x-delayed-message", true, false, map);
        //注意:arguments需要声明在队列上,声明在交换机上是不会起作用的。
        channel.queueDeclare(queueName, true, false, false, map);
        channel.queueBind(queueName,exchangeName,routingkey);
        for (int i = 0; i < 3; i++) {
             // deliveryMode=2 持久化,expiration 消息有效时间
            String msg = "delayed payload".getBytes("UTF-8") +" "+new Date().getTime();
            byte[] messageBodyBytes = msg.getBytes();
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("x-delay", 50000);
            AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers);
            channel.basicPublish(exchangeName, routingkey, props.build(), messageBodyBytes);
        }
    }
}

消费者


import java.io.IOException;
import java.util.Date;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
public class Consumer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setVirtualHost("/");
        connectionFactory.setHost("172.31.1.135");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("xxx");
        connectionFactory.setPassword("xxx");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        String queueName = "delay_queueName";
        channel.queueDeclare(queueName,true,false,false,null);
        channel.basicConsume(queueName, false, "myConsumer Tag", new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
                  String routingKey = envelope.getRoutingKey();
                  String convernType = properties.getContentType();
                  long deliveryTag = envelope.getDeliveryTag();
                  System.out.println("routingKey:"+routingKey+",convernType:"+convernType+",deliveryTag:"+deliveryTag+",Msg body:"+new String(body)+ " "+new Date().getTime());
                  channel.basicAck(deliveryTag, false);
            }
        });
    }
}


相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
7月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
87 2
|
3月前
|
消息中间件 存储 缓存
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
292 7
RabbitMQ:交换机详解(Fanout交换机、Direct交换机、Topic交换机)
|
3月前
|
消息中间件 JSON Java
玩转RabbitMQ声明队列交换机、消息转换器
玩转RabbitMQ声明队列交换机、消息转换器
101 0
|
3月前
|
消息中间件 存储
RabbitMQ-死信交换机和死信队列
死信队列和死信交换机是RabbitMQ提供的一个非常实用的功能,通过合理使用这一机制,可以大大增强系统的健壮性和可靠性。它们不仅能有效解决消息处理失败的情况,还能为系统的错误追踪、消息延迟处理等提供支持。在设计系统的消息体系时,合理规划和使用死信队列和死信交换机,将会为系统的稳定运行提供一个有力的
73 0
|
6月前
|
消息中间件 RocketMQ
消息队列 MQ产品使用合集之在开源延时消息插件方案中和原生延时消息方案中,同时设置参数是否会出现错乱
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
6月前
|
消息中间件
02.交换机RabbitMQ交换机
02.交换机RabbitMQ交换机
87 0
|
7月前
|
消息中间件 Java
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
67 1
|
7月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
81 1
|
1月前
|
网络协议 安全 网络虚拟化
思科交换机配置命令归纳
【11月更文挑战第8天】本文总结了思科交换机的常见配置命令,包括模式转换、基本配置、查看命令、VLAN 配置、Trunk 配置、以太网通道配置、VTP 配置、三层交换机配置、生成树配置以及其他常用命令,适用于网络管理和维护。
129 2
|
2月前
|
安全 网络安全 数据安全/隐私保护
Cisco-交换机配置聚合端口
Cisco-交换机配置聚合端口