@Configuration
public class TTLRabbitMqConfig {
//1.声明注册fanout模式的交换机
@Bean
public DirectExchange ttlDirectExchange(){
return new DirectExchange("ttl_direct_order_exchange",true,false);
}
@Bean
public Queue ttlSmsQueue(){
//参数 args
Map<String,Object> args = new HashMap<>();
//设置过期时间
args.put("x-message-ttl",5000);
return new Queue("ttl.sms.direct.queue",true,false,false,args);
}
@Bean
public Binding ttlSmsBinding() {
return BindingBuilder.bind(ttlSmsQueue()).to(ttlDirectExchange()).with("ttl");
}
}
//模拟用户下单
public void makeOrder(String userid,String productid,int num){
//1.根据商品id查询库存是否足够
//2.保存订单
String orderId = UUID.randomUUID().toString();
System.out.println("dd订单生产成功:" + orderId);
//3.通过MQ来完成消息的分发
//参数1:交换机 参数2:路由key/queue队列名称 参数3:消息内容
String exchangeName = "ttl_direct_order_exchange";
rabbitTemplate.convertAndSend(exchangeName,"ttl",orderId);
}
@Test
void contextLoadsTtlMessage() {
orderService.makeOrderTtlMessage( "12afdsf342","fsdas234sfddf",2);
}
/**
* 一个普通队列, 在消息中设置过期时间
*/
@Bean
public Queue ttlMessageSmsQueue(){
return new Queue("ttl.message.sms.direct.queue",true);
}
@Bean
public Binding ttlMessageSmsBinding() {
return BindingBuilder.bind(ttlMessageSmsQueue()).to(ttlDirectExchange()).with("ttlmessage");
}
public void makeOrderTtlMessage(String userid,String productid,int num){
//1.根据商品id查询库存是否足够
//2.保存订单
String orderId = UUID.randomUUID().toString();
System.out.println("dd订单生产成功:" + orderId);
//3.通过MQ来完成消息的分发
//参数1:交换机 参数2:路由key/queue队列名称 参数3:消息内容
String exchangeName = "ttl_direct_order_exchange";
//给消息设置过期时间
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setExpiration("5000");
message.getMessageProperties().setContentEncoding("UTF-8");
return message;
}
};
rabbitTemplate.convertAndSend(exchangeName,"ttlmessage",orderId,messagePostProcessor);
}
@Test
void contextLoadsTtlMessage() {
orderService.makeOrderTtlMessage( "12afdsf342","fsdas234sfddf",2);
}
DLX,全称 Dead-Letter-Exchange,可以称之为死信交换机,也有人称之为死信邮箱。当消息在一个队列中变成死信之后,它能被重新发送到另一个交换机中,这个交换机就是 DLX,绑定 DLX的队列就称之为死信队列。消息变成死信,可能是由于以下原因:
DLX也是一个普通的交换机,和一般的交换机没有区别,它能在任何的队列上被指定,实际上就是设置某一个队列的属性,当这个队列中存在死信时,Rabbitmq就会自动地将这个消息重新发布到设置的 DLX上去,进而被路由到另一个队列,即死信队列。要想使用死信队列,只需要在定义队列的时候设置队列参数x-dead-letter-exchange指定DLX(死信交换机)即可;
@Configuration
public class DeadRabbitMqConfiguration{
//1.声明注册direct模式的交换机
@Bean
public DirectExchange deadDirectExchange(){
return new DirectExchange("dead_direct_exchange",true,false);
}
//2.队列的过期时间
@Bean
public Queue deadQueue() {
return new Queue("dead.direct.queue",true);
}
@Bean
public Binding deadbinds(){
return BindingBuilder.bind(deadQueue()).to(deadDirectExchange()).with("dead");
}
}
/**
* 一个有过期时间的队列
*/
@Bean
public Queue ttlSmsQueue(){
//参数 args
Map<String,Object> args = new HashMap<>();
//设置过期时间
args.put("x-message-ttl",5000);
//指定死信队列
args.put("x-dead-letter-exchange","dead_direct_exchange");
args.put("x-dead-letter-routing-key","dead");
return new Queue("ttl.sms.direct.queue",true,false,false,args);
}
@Test
void contextLoadsTTL() {
orderService.makeOrderTTL( "12afdsf342","fsdas234sfddf",2);
}
延迟队列实现的思路也很简单,就是上篇文章我们所说的 DLX(死信交换机)+TTL(消息超时时间)。我们可以把死信队列就当成延迟队列。
具体来说是这样: 假如一条消息需要延迟 30 分钟执行,我们就设置这条消息的有效期为 30 分钟,同时为这条消息配置死信交换机和死信 routing_key,并且不为这个消息队列设置消费者,那么 30 分钟后,这条消息由于没有被消费者消费而进入死信队列,此时我们有一个消费者就在“蹲点”这个死信队列,消息一进入死信队列,就立马被消费了。
下载地址:http://www.rabbitmq.com/community-plugins.html (选择要安装的插件,跳转到github下载,注意选择合适的版本;)
安装插件的教程地址https://www.rabbitmq.com/installing-plugins.html(我这里选择将下载的插件放在/usr/lib/rabbitmq/lib/rabbitmq_server-{version}/plugins中)

执行rabbitmq-plugins enable rabbitmq_delayed_message_exchange,如下;
[root@VM-4-17-centos plugins]# rabbitmq-plugins enable rabbitmq_delayed_message_exchange
Enabling plugins on node rabbit@VM-4-17-centos:
rabbitmq_delayed_message_exchange
The following plugins have been configured:
rabbitmq_delayed_message_exchange
rabbitmq_management
rabbitmq_management_agent
rabbitmq_web_dispatch
Applying plugin configuration to rabbit@VM-4-17-centos...
The following plugins have been enabled:
rabbitmq_delayed_message_exchange
started 1 plugins.

当内存使用超过配置的阈值或者磁盘空间剩余空间小于配置的阈值时,RabbitMQ会暂时阻塞客户端的连接,并且停止接收从客户端发来的消息,以此避免服务器的崩溃,客户端与服务端的心态检测机制也会失效。

官方文档:http://www.rabbbitmq.com/configure.html
有两种方式:
rabbitmqctl set_vm_memory_high_watermark <fraction>
rabbitmqctl set_vm_memory_high_watermark absolute 50MB
fraction/value 为内存阈值。默认情况是:0.4/2GB,代表的含义是:当 RabbitMQ的内存超过40%时,就会产生警告并且会阻塞所有生产者的连接。通过此命令修改阈值在 Broker重启以后将会失效,通过修改配置文件设置的阈值则不会随着重启而消失,但修改了配置文件一样要重启 Broker才会生效
vm_memory_high_watermark.relative = 0.6
# KB,MB,GB
vm_memory_high_watermark.absolute = 2GB
配置文件的位置:
默认是没有配置文件的,需要自己创建
官方文档:https://www.rabbitmq.com/configure.html#config-location