1, Delay queue and dead letter queue
Dead letter queue: elements are not consumed in time after they are generated and are always stored in the queue.
Delay queue: the delay queue has a sequence within the queue. Its most important feature is the delay time. It wants to be taken out and processed after or before the specified time. It is similar to a scheduled task, but it saves resources than a scheduled task. Delay queue is actually a special dead letter queue.
However, the queue is characterized by first in first out. If there are different times corresponding to different elements, the corresponding effect cannot be achieved if there are elements with small delay time after elements with large delay time.
How to achieve the effect of similar scheduled tasks?
After the message is sent, consumers do not want to get the message immediately, but wait for a specified time before they get the message for consumption.
There are generally two methods
1. Delay queue (TTL) + dead letter queue (DLX)
2. Dead letter queue (DLX) + corresponding plug-in
2, Implementation method
1.TTL+DLX
The purpose of this method is to first generate two queues, one as the dead letter queue and the other as the delay queue for binding dead letters. By setting TTL (delay time: it can be set here or set for messages in the producer, but this method requires that a dead letter queue stores messages with the same delay time).
/**
* Build TTL dead letter delay exchange and name it
*/
@Bean
public DirectExchange delayDirectExchange() {
return ExchangeBuilder
.directExchange(QueueEnum.DELAY_QUEUE.getExchangeName())
.durable(true)
.build();
}
/**
* Exchange used to forward dead letter (that is, an ordinary exchange)
*/
@Bean
public DirectExchange delayBindExchange() {
return ExchangeBuilder
.directExchange(QueueEnum.DELAY_BIND_QUEUE.getExchangeName())
.durable(true)
.build();
}
/**
* TTL Dead letter delay queue
*/
@Bean
public Queue delayQueue() {
return QueueBuilder
.durable(QueueEnum.DELAY_QUEUE.getQueueName())
.withArgument("x-message-ttl", CommonConstant.OUT_ORDER_TIME)
.withArgument("x-dead-letter-exchange", QueueEnum.DELAY_BIND_QUEUE.getExchangeName())
.withArgument("x-dead-letter-routing-key", QueueEnum.DELAY_BIND_QUEUE.getPath())
.build();
}
/**
* The queue responsible for forwarding TTL (also an ordinary queue)
*/
@Bean
public Queue delayBindQueue() {
return QueueBuilder
.durable(QueueEnum.DELAY_BIND_QUEUE.getQueueName())
.build();
}
/**
* TTL Dead letter forwarding binding
*/
@Bean
Binding delayBindBinding (){
return BindingBuilder
.bind(delayBindQueue())
.to(delayBindExchange())
.with(QueueEnum.DELAY_BIND_QUEUE.getPath());
}
}Delay is an important attribute of TTL queue
x-message-ttl: delay time
x-dead-letter-exchange: bound dead letter exchange
x-dead-letter-routing-key: bound route, generally corresponding to the name of dead letter queue.
The producer can add the delay time of the message by calling the method, and mq takes the shortest delay time as the standard.
/**
* Send delay message
* @param msg Message content
* @param delayTime Delay time in milliseconds
*/
private static AmqpTemplate amqpTemplate;
public static void sendDelayMsg(String msg, Long delayTime,QueueEnum queueEnum){
amqpTemplate.convertAndSend(queueEnum.getExchangeName(),queueEnum.getPath(),msg,message -> {
message.getMessageProperties().setExpiration(delayTime.toString());
return message;
});
}This is a self written method, in which the most important method for producers to produce messages is
convertAndSend()
2. Dead letter queue (DLX) + corresponding plug-in
The above demonstrates the delay queue implemented through TTL+DLX. In this way, the more time periods, the more switches and queues are required. The implementation of the previous method is a little complicated. In fact, there is a simpler implementation. Rabbitmq version 3.5.7 and above provides a plug-in (rabbitmq delayed message exchange) to implement the delay queue function. Meanwhile, the plug-in depends on Erlang/OPT 18.0 and above.
I didn't try this method
https://blog.csdn.net/u012988901/article/details/89296451