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
相关文章
|
2月前
|
消息中间件 存储 缓存
RabbitMQ中的交换机
RabbitMQ中的交换机
34 2
|
2月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
SpringBoot实现RabbitMQ的广播交换机(SpringAMQP 实现Fanout广播交换机)
31 2
|
12天前
|
消息中间件 RocketMQ
消息队列 MQ产品使用合集之在开源延时消息插件方案中和原生延时消息方案中,同时设置参数是否会出现错乱
消息队列(MQ)是一种用于异步通信和解耦的应用程序间消息传递的服务,广泛应用于分布式系统中。针对不同的MQ产品,如阿里云的RocketMQ、RabbitMQ等,它们在实现上述场景时可能会有不同的特性和优势,比如RocketMQ强调高吞吐量、低延迟和高可用性,适合大规模分布式系统;而RabbitMQ则以其灵活的路由规则和丰富的协议支持受到青睐。下面是一些常见的消息队列MQ产品的使用场景合集,这些场景涵盖了多种行业和业务需求。
|
17天前
|
消息中间件
02.交换机RabbitMQ交换机
02.交换机RabbitMQ交换机
12 0
|
2月前
|
消息中间件 Java
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
SpringBoot实现RabbitMQ的通配符交换机(SpringAMQP 实现Topic交换机)
22 1
|
2月前
|
消息中间件 Java Spring
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
SpringBoot实现RabbitMQ的定向交换机(SpringAMQP 实现Direct定向交换机)
30 1
|
2月前
|
消息中间件 微服务
RabbitMQ入门指南(十):延迟消息-死信交换机
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了死信交换机、死信交换机实现延迟消息等内容。
93 0
|
2月前
|
消息中间件 Java API
RabbitMQ入门指南(五):Java声明队列、交换机以及绑定
RabbitMQ是一个高效、可靠的开源消息队列系统,广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了Java声明队列、交换机以及绑定队列和交换机等内容。
112 0
|
2月前
|
网络性能优化 网络虚拟化 网络架构
配置接口限速示例(盒式交换机)
接口限速简介 接口限速对通过整个端口的全部报文流量速率进行限制,不对具体流量进行区分,可以实现给某个接口分配固定的带宽,控制方式单一,配置简单。 入方向与出方向的接口限速属于并列关系,用户可以根据需要同时配置,也可以单独配置。
|
2月前
|
网络协议 网络虚拟化 网络架构
交换机入门快速配置
交换机入门快速配置

热门文章

最新文章