本文基于spring-boot-starter-amqp依赖实现
RabbitMQ实现定时消息有两种方式
- 基于消息TTL与死信交换
- 基于插件 RabbitMQ Delayed Message Plugin
基于消息TTL与死信交换
死信交换: 可以为队列设置一个死信exchange和routingKey,当队列上产生死信时,死信会被投递到设置好的exchange及对应的routingKey,说白点就是可以为每个队列设置一个"垃圾桶",产生死信就扔到垃圾桶里,只不过垃圾桶也是一个正常的队列.
如何会产生死信(怎样会被扔到垃圾桶)
- 当消息被拒绝,并且拒绝后没有重新进入队列
- 由于消息TTL(剩余生存时间),过期的消息
- 超过队列长度而被删除的消息
思路:
为消息设置过期时间
把消息投递到一个没有消费者的队列,ttl-queue 使消息成为死信
为ttl-queue设置死信交换,dead-letter-queue
为dead-letter-queue添加消费者,监听消息
先声明那个没有消费者的队列,ttl-queue,并为它指定"垃圾桶"(死信交换)
@Configuration
public class TtlConfig {
@Bean("myTtlExchange")
public Exchange ttlExchange() {
return new DirectExchange("mine-ttl-exchange");
}
@Bean("myTtlQueue")
public Queue ttlQueue() {
//设置在这个队列上的私信交换(垃圾桶)->
// mine-dead-letter-exchange交换机,
// 并且routingKey为mine.dead.letter
Map<String, Object> args = new HashMap<>(4);
args.put("x-dead-letter-exchange", "mine-dead-letter-exchange");
args.put("x-dead-letter-routing-key", "mine.dead.letter");
return new Queue("mine-ttl-queue", true, false, false, args);
}
@Bean
@DependsOn({"myTtlExchange", "myTtlQueue"})
public Binding ttlBinding(Queue myTtlQueue, Exchange myTtlExchange) {
return BindingBuilder.bind(myTtlQueue).to(myTtlExchange).with("mine.ttl").noargs();
}
}
监听上面设置好的"垃圾桶"
@Component
@RabbitListener(bindings = @QueueBinding(
value = @Queue("mine-dead-letter-queue"),
exchange = @Exchange("mine-dead-letter-exchange"),
key = "mine.dead.letter" ))
public class DeadLetterListener {
@RabbitHandler
public void onMessage(String msg) {
System.out.println(msg);
System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:ss:mm")));
}
}
为消息设置过期时间
/**
* 每条消息单独设置过期时间
*/
public void send() {
System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:ss:mm")));
rabbitTemplate.convertAndSend("mine-ttl-exchange", "mine.ttl", "this is a message ",message -> {
//设置10秒过期
message.getMessageProperties().setExpiration("10000");
return message;
});
}
每条消息都设置过期时间,很灵活,若是固定的过期时间,可以在队列上设置, 为队列设置 x-message-ttl 那在这个队列上的每条消息的过期时间都为 x-message-ttl 的值,只需要稍微修改一下ttl-queue
......
@Bean("myTtlQueue")
public Queue ttlQueue() {
//设置在这个队列上的私信的去处->
// mine-dead-letter-exchange交换机,
// 并且routingKey为mine.dead.letter
Map<String, Object> args = new HashMap<>(4);
args.put("x-dead-letter-exchange", "mine-dead-letter-exchange");
args.put("x-dead-letter-routing-key", "mine.dead.letter");
//过期时间由队列统一设置
//注意不是 x-expires,x-expires为队列存活时间,
// x-message-ttl为队列内的消息存活时间
//注意更改队列/交换机设置需要删除原有的
args.put("x-message-ttl", 20000);
return new Queue("mine-ttl-queue", true, false, false, args);
}
.....
/**这样的话也不用为每条消息设置过期时间了
* 消息过期时间由队列统一决定
*/
public void send1() {
System.out.println(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:ss:mm")));
rabbitTemplate.convertAndSend("mine-ttl-exchange", "mine.ttl", "this is a message ");
}
基于插件
插件名称 rabbitmq_delayed_message_exchange
插件下载 (注意版本哦)
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange
解压后把插件放在RabbitMQ的plugin目录下,默认路径
lunix : /usr/lib/rabbitmq/lib/rabbitmq_server-version/plugins/
windows : C:\Program Files\RabbitMQ\rabbitmq_server-version\plugins\
启用插件
# 查看插件是否在列表里
rabbitmq-plugins list
# 启用插件
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
# 启用后再执行
rabbitmq-plugins list
如下,插件前有 [E*] 表示启用了
......
[E*] rabbitmq_delayed_message_exchange 20171201-3.7.x
[ ] rabbitmq_event_exchange 3.7.15
[ ] rabbitmq_federation 3.7.15
[ ] rabbitmq_federation_management 3.7.15
[ ] rabbitmq_jms_topic_exchange 3.7.15
......
使用
-
申明一个延时消息的exchange 即:
- 交换机类型为 x-delayed-message
- 为交换机设置参数 "x-delayed-type" ,值为四个交换机类型(direct ,fanout ,topic ,headers )
- 其余使用与普通交换机无差别
@Component
@RabbitListener(bindings = @QueueBinding(
key = "mine.delay",
value = @Queue("mine-delay-queue"),
exchange = @Exchange( value = "mine-delay-exchange",
type = "x-delayed-message",
arguments = @Argument(name = "x-delayed-type",
value = ExchangeTypes.DIRECT))))
public class DelayListener {
@RabbitHandler
public void onMessage(String msg) {
System.out.println(msg);
System.out.println(LocalDateTime.now()
.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
}
}
- 生产者发送消息时,设置延时的时间 x-delay
@Autowired
private RabbitTemplate rabbitTemplate;
public void send() {
System.out.println(LocalDateTime.now()
.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
rabbitTemplate.convertAndSend("mine-delay-exchange", "mine.delay", "this is a message ",
message -> {
MessageProperties messageProperties = message.getMessageProperties();
//单位,毫秒
messageProperties.setDelay(10000);
return message;
});
}
转载请注明出处
系列文章
