在实际的业务中我们会遇见生产者产生的消息&＃xff0c;不立即消费&＃xff0c;而是延时一段时间在消费。RabbitMQ本身没有直接支持延迟队列功能&＃xff0c;但是我们可以根据其特性Per-Queue Message TTL和 Dead Letter Exchanges实现延时队列。也可以通过改特性设置消息的优先级。
1.Per-Queue Message TTL
RabbitMQ可以针对消息和队列设置TTL(过期时间)。队列中的消息过期时间(Time To Live, TTL)有两种方法可以设置。第一种方法是通过队列属性设置&＃xff0c;队列中所有消息都有相同的过期时间。第二种方法是对消息进行单独设置&＃xff0c;每条消息TTL可以不同。如果上述两种方法同时使用&＃xff0c;则消息的过期时间以两者之间TTL较小的那个数值为准。消息在队列的生存时间一旦超过设置的TTL值&＃xff0c;就成为dead message,消费者将无法再收到该消息。
2.Dead Letter Exchanges
当消息在一个队列中变成死信后&＃xff0c;它能被重新publish到另一个Exchange。消息变成Dead Letter一向有以下几种情况&＃xff1a;
消息被拒绝(basic.reject or basic.nack)并且requeue&＃61;false
消息TTL过期
队列达到最大长度
实际上就是设置某个队列的属性&＃xff0c;当这个队列中有Dead Letter时&＃xff0c;RabbitMQ就会自动的将这个消息重新发布到设置的Exchange中去&＃xff0c;进而被路由到另一个队列&＃xff0c;publish可以监听这个队列中消息做相应的处理&＃xff0c;这个特性可以弥补RabbitMQ 3.0.0以前支持的immediate参数中的向publish确认的功能。
虽然 consumer 从来看不到过期的 message &＃xff0c;但是在过期 message 到达 queue 的头部时确实会被真正的丢弃(或者 dead-lettered )。当对每一个 queue 设置了 TTL 值时不会产生任何问题&＃xff0c;因为过期的 message 总是会出现在 queue 的头部。当对每一条 message 设置了 TTL 时&＃xff0c;过期的 message 可能会排队于未过期 message 的后面&＃xff0c;直到这些消息被 consume 到或者过期了。在这种情况下&＃xff0c;这些过期的 message 使用的资源将不会被释放&＃xff0c;且会在 queue 统计信息中被计算进去(例如&＃xff0c;queue 中存在的 message 的数量)。对于第一种设置队列TTL属性的方法&＃xff0c;一旦消息过期&＃xff0c;就会从队列中抹去&＃xff0c;而第二种方法里&＃xff0c;即使消息过期&＃xff0c;也不会马上从队列中抹去&＃xff0c;因为每条消息是否过期时在即将投递到消费者之前判定的&＃xff0c;为什么两者得处理方法不一致&＃xff1f;因为第一种方法里&＃xff0c;队列中已过期的消息肯定在队列头部&＃xff0c;RabbitMQ只要定期从队头开始扫描是否有过期消息即可&＃xff0c;而第二种方法里&＃xff0c;每条消息的过期时间不同&＃xff0c;如果要删除所有过期消息&＃xff0c;势必要扫描整个队列&＃xff0c;所以不如等到此消息即将被消费时再判定是否过期&＃xff0c;如果过期&＃xff0c;再进行删除。
一、在队列上设置TTL
1.建立delay.exchange
这里Internal设置为NO&＃xff0c;否则将无法接受dead letter&＃xff0c;YES表示这个exchange不可以被client用来推送消息&＃xff0c;仅用来进行exchange和exchange之间的绑定。
2.建立延时队列(delay queue)
如上配置延时5min队列(x-message-ttl&＃61;300000)
x-max-length:最大积压的消息个数&＃xff0c;可以根据自己的实际情况设置&＃xff0c;超过限制消息不会丢失&＃xff0c;会立即转向delay.exchange进行投递
x-dead-letter-exchange:设置为刚刚配置好的delay.exchange,消息过期后会通过delay.exchange进行投递
这里不需要配置"dead letter routing key"否则会覆盖掉消息发送时携带的routingkey&＃xff0c;导致后面无法路由为刚才配置的delay.exchange
3.配置延时路由规则
需要延时的消息到exchange后先路由到指定的延时队列
1)创建delaysync.exchange通过Routing key将消息路由到延时队列
2.配置delay.exchange 将消息投递到正常的消费队列
配置完成。
下面使用代码测试一下&＃xff1a;
生产者&＃xff1a;
package cn.slimsmart.study.rabbitmq.delayqueue.queue;
import java.io.IOException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
private static String queue_name &＃61; "test.queue";
public static void main(String[] args) throws IOException {
ConnectionFactory factory &＃61; new ConnectionFactory();
factory.setHost("10.1.199.169");
factory.setUsername("admin");
factory.setPassword("123456");
Connection connection &＃61; factory.newConnection();
Channel channel &＃61; connection.createChannel();
// 声明队列
channel.queueDeclare(queue_name, true, false, false, null);
String message &＃61; "hello world!" &＃43; System.currentTimeMillis();
channel.basicPublish("delaysync.exchange", "deal.message", null, message.getBytes());
System.out.println("sent message: " &＃43; message &＃43; ",date:" &＃43; System.currentTimeMillis());
// 关闭频道和连接
channel.close();
connection.close();
}
}
消费者&＃xff1a;
package cn.slimsmart.study.rabbitmq.delayqueue.queue;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
private static String queue_name &＃61; "test.queue";
public static void main(String[] args) throws Exception {
ConnectionFactory factory &＃61; new ConnectionFactory();
factory.setHost("10.1.199.169");
factory.setUsername("admin");
factory.setPassword("123456");
Connection connection &＃61; factory.newConnection();
Channel channel &＃61; connection.createChannel();
// 声明队列
channel.queueDeclare(queue_name, true, false, false, null);
QueueingConsumer consumer &＃61; new QueueingConsumer(channel);
// 指定消费队列
channel.basicConsume(queue_name, true, consumer);
while (true) {
// nextDelivery是一个阻塞方法(内部实现其实是阻塞队列的take方法)
QueueingConsumer.Delivery delivery &＃61; consumer.nextDelivery();
String message &＃61; new String(delivery.getBody());
System.out.println("received message:" &＃43; message &＃43; ",date:" &＃43; System.currentTimeMillis());
}
}
}
二、在消息上设置TTL
实现代码&＃xff1a;
生产者&＃xff1a;
package cn.slimsmart.study.rabbitmq.delayqueue.message;
import java.io.IOException;
import java.util.HashMap;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Producer {
private static String queue_name &＃61; "message_ttl_queue";
public static void main(String[] args) throws IOException {
ConnectionFactory factory &＃61; new ConnectionFactory();
factory.setHost("10.1.199.169");
factory.setUsername("admin");
factory.setPassword("123456");
Connection connection &＃61; factory.newConnection();
Channel channel &＃61; connection.createChannel();
HashMap arguments &＃61; new HashMap();
arguments.put("x-dead-letter-exchange", "amq.direct");
arguments.put("x-dead-letter-routing-key", "message_ttl_routingKey");
channel.queueDeclare("delay_queue", true, false, false, arguments);
// 声明队列
channel.queueDeclare(queue_name, true, false, false, null);
// 绑定路由
channel.queueBind(queue_name, "amq.direct", "message_ttl_routingKey");
String message &＃61; "hello world!" &＃43; System.currentTimeMillis();
// 设置延时属性
AMQP.BasicProperties.Builder builder &＃61; new AMQP.BasicProperties.Builder();
// 持久性 non-persistent (1) or persistent (2)
AMQP.BasicProperties properties &＃61; builder.expiration("300000").deliveryMode(2).build();
// routingKey &＃61;delay_queue 进行转发
channel.basicPublish("", "delay_queue", properties, message.getBytes());
System.out.println("sent message: " &＃43; message &＃43; ",date:" &＃43; System.currentTimeMillis());
// 关闭频道和连接
channel.close();
connection.close();
}
}
消费者&＃xff1a;
package cn.slimsmart.study.rabbitmq.delayqueue.message;
import java.util.HashMap;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
public class Consumer {
private static String queue_name &＃61; "message_ttl_queue";
public static void main(String[] args) throws Exception {
ConnectionFactory factory &＃61; new ConnectionFactory();
factory.setHost("10.1.199.169");
factory.setUsername("admin");
factory.setPassword("123456");
Connection connection &＃61; factory.newConnection();
Channel channel &＃61; connection.createChannel();
HashMap arguments &＃61; new HashMap();
arguments.put("x-dead-letter-exchange", "amq.direct");
arguments.put("x-dead-letter-routing-key", "message_ttl_routingKey");
channel.queueDeclare("delay_queue", true, false, false, arguments);
// 声明队列
channel.queueDeclare(queue_name, true, false, false, null);
// 绑定路由
channel.queueBind(queue_name, "amq.direct", "message_ttl_routingKey");
QueueingConsumer consumer &＃61; new QueueingConsumer(channel);
// 指定消费队列
channel.basicConsume(queue_name, true, consumer);
while (true) {
// nextDelivery是一个阻塞方法(内部实现其实是阻塞队列的take方法)
QueueingConsumer.Delivery delivery &＃61; consumer.nextDelivery();
String message &＃61; new String(delivery.getBody());
System.out.println("received message:" &＃43; message &＃43; ",date:" &＃43; System.currentTimeMillis());
}
}
}
spring-rabbit整合教程
maven依赖&＃xff1a;
org.springframework.amqp
spring-rabbit
1.4.6.RELEASE
spring配置文件(在文件头部引入rabbit的命名空间和约束文件)&＃xff1a;
xmlns:aop&＃61;"http://www.springframework.org/schema/aop"
xmlns:context&＃61;"http://www.springframework.org/schema/context"
xmlns:mvc&＃61;"http://www.springframework.org/schema/mvc"
xmlns:task&＃61;"http://www.springframework.org/schema/task"
xmlns:tx&＃61;"http://www.springframework.org/schema/tx"
xmlns:xsi&＃61;"http://www.w3.org/2001/XMLSchema-instance"
xmlns:rabbit&＃61;"http://www.springframework.org/schema/rabbit"
xsi:schemaLocation&＃61;"http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd">
那么在项目中装配amqpTemplate中就可以发送消息了
京公网安备 11010802041100号