这篇文章主要讲解了“RabbitMQ消息有效期与死信的处理过程是什么”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“RabbitMQ消息有效期与死信的处理过程是什么”吧!
RabbitMQ的TTL全称为Time-To-Live,表示的是消息的有效期。消息如果在队列中一直没有被消费并且存在时间超过了TTL,消息就会变成了"死信" (Dead Message),后续无法再被消费了。如果不设置TTL,则表示此消息永久有效(默认消息是不会失效的)。如果将TTL设为0,则表示如果消息不能被立马消费则会被立即丢掉,这个特性可以部分替代RabbitMQ3.0以前支持的immediate参数,之所以所部分代替,是应为immediate参数在投递失败会有basic.return方法将消息体返回(这个功能可以利用死信队列来实现)。
设置TTL有两种方式:
队列有效期:是声明队列的时候,在队列的属性中设置,这样该队列中的消息都会有相同的有效期
消息有效期:发送消息时给消息设置属性,可以为每条消息都设置不同的TTL
如果两种方式都设置了,则以设置的较小的为准。
区别:如果声明队列时设置了有效期,则消息过期了就会被删掉;如果是发消息时设置的有效期,消息过期了也不会被立马删掉,因为这时消息是否过期是在要投递给消费者时判断的。
定义队列的方法如下:
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException;
该方法的arguments参数可以设置队列的属性,属性名为x-message-ttl,单位为毫秒。后台添加的话如下:
代码中设置如下:
Map<String, Object> arguments= new HashMap<String , Object>(); arguments.put("x-message-ttl " , 10000);//10秒钟 单位为毫秒 channel.queueDeclare(queueName , durable , exclusive , autoDelete , arguments) ;
命令行模式来设置:
rabbitmqctl set_policy TTL ".*" '{"message-ttl":100000}' --apply-to queues
通过HTTP接口调用:
$ curl -i -u guest:guest -H "content-type:application/json" -XPUT -d'{"auto_delete":false,"durable":true,"arguments":{"x-message-ttl": 100000}}' http://ip:15672/api/queues/{vhost}/{queuename}
有效期Expire可以让队列在指定时间内 “未被使用” 的话会自动过期删除,未使用的意思是 queue 上没有任何 consumer,queue 没有被重新声明,并且在过期时间段内未调用过 basic.get 命令。该方式可用于,例如,RPC-style 的回复 queue, 其中许多queue 会被创建出来,但是却从未被使用。
服务器会确保在过期时间到达后 queue 被删除,但是不保证删除的动作有多么的及时。在服务器重启后,持久化的queue 的超时时间将重新计算。 x-expires 参数值以毫秒为单位,并且服从和 x-message-ttl 一样的约束条件,且不能设置为 0 。所以,如果该参数设置为 10000 ,则表示该 queue 如果在 10s之内未被使用则会被删除。
代码如下:
Map<String, Object> args = new HashMap<String, Object>(); args.put("x-expires", 10000); channel.queueDeclare("queue", false, false, false, args);
发送消息的方法如下:
void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
在该方法的props参数可以设置其有效期:
Map<String, Object> headers = new HashMap<String, Object>(); AMQP.BasicProperties properties = new AMQP.BasicProperties().builder() .deliveryMode(2) // 消息持久 .contentEncoding("UTF-8") // 编码方式 .contentType("text/plain") .expiration("100000") .headers(headers) .build(); channel.basicPublish("", queueName, properties, message.getBytes());
通过HTTPAPI 接口设置:
$ curl -i -u guest:guest -H "content-type:application/json" -XPOST -d '{"properties":{"expiration":"100000"},"routing_key":"routingkey","payload":"bodys","payload_encoding":"string"}' http://localhost:15672/api/exchanges/{vhost}/{exchangename}/publish
介绍
死信队列:DLX,dead-letter-exchange
利用DLX,当消息在一个队列中变成死信 (dead message) 之后,它能被重新publish到另一个Exchange,这个Exchange就是DLX
消息变成死信几种情况
消息被拒绝(Basic.Reject/Basic.Nack) ,井且设置requeue 参数为false
消息过期
队列达到最大长度
死信处理过程
DLX也是一个正常的Exchange,和一般的Exchange没有区别,它能在任何的队列上被指定,实际上就是设置某个队列的属性。
当这个队列中有死信时,RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去,进而被路由到另一个队列。
可以监听这个队列中的消息做相应的处理。
用途
通过监控消费死信队列中消息,来观察和分析数据。
结合TTL实现延迟队列(比如下单超过多长时间自动关闭)
使用
代码如下:
channel.exchangeDeclare("dlx_exchange" , "direct"); //创建DLX: dlx_exchange Map<String, Object> args = new HashMap<String, Object>(); args.put("x-dead-letter-exchange" , "dlx_exchange ");//设置死信交换机 args.put("x-dead-letter-routing-key" , "dlx-routing-key");//设置DLX的路由键(可以不设置) channel.queueDeclare("myqueue" , false , false , false , args);
实例
public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); //声明一个交换机,做死信交换机用 channel.exchangeDeclare("dlx_exchange", "topic", true, false, null); //声明一个队列,做死信队列用 channel.queueDeclare("dlx_queue", true, false, false, null); //队列绑定到交换机上 channel.queueBind("dlx_queue", "dlx_exchange", "dlx.*"); channel.exchangeDeclare("normal_exchange", "fanout", true, false, null); Map<String, Object> arguments=new HashMap<String, Object>(); arguments.put("x-message-ttl" , 1000);//设置消息有效期1秒,过期后变成私信消息,然后进入DLX arguments.put("x-dead-letter-exchange" , "dlx_exchange");//设置DLX arguments.put("x-dead-letter-routing-key" , "dlx.test");//设置DLX的路由键 //为队列normal_queue 添加DLX channel.queueDeclare("normal_queue", true, false, false, arguments); channel.queueBind("normal_queue", "normal_exchange", ""); channel.basicPublish("normal_exchange", "", MessageProperties.PERSISTENT_TEXT_PLAIN, ("测试死信消息").getBytes()); System.out.println("发送消息时间:"+ConnectionUtil.formatDate(new Date())); channel.close(); connection.close(); }
说明:
申明死信队列dlx_queue的绑定如下,与死信交换机dlx_exchange(topic类型)进行绑定,routing key为"dlx.*"
申明队列normal_queue,与交换机normal_exchange(fanout类型)进行绑定
执行流程:
消息发送到交换机normal_exchange,然后路由到队列normal_queue上
因为队列normal_queue没有消费者,消息过期后成为死信消息
死信消息携带设置的x-dead-letter-routing-key=dlx.test进入到死信交换机dlx_exechage
dlx_exechage与dlx_queue绑定的routing key为"dlx.*",死信消息的路由键dlx.test符合该规则被路由到dlx.queue上面。
感谢各位的阅读,以上就是“RabbitMQ消息有效期与死信的处理过程是什么”的内容了,经过本文的学习后,相信大家对RabbitMQ消息有效期与死信的处理过程是什么这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。