Message Queue-RabbitMQ: Dead Letter Queue
Message Queue-RabbitMQ: Dead Letter Queue
15. Dead letter queue
1. The concept of dead letter
Let’s first clarify this definition from a conceptual explanation. Dead letters, as the name suggests, are messages that cannot be consumed. The literal meaning can be understood like this. Generally speaking, the producer delivers the message to the broker or directly to the queue, and the consumer takes out the message from the queue. Consumption, but sometimes due to specific reasons**, some messages in the queue cannot be consumed** . If such messages are not processed later, they become dead letters. If there are dead letters, there will naturally be a dead letter queue .
Application scenario: In order to ensure that the message data of the order business is not lost , the dead letter queue mechanism of RabbitMQ needs to be used When an exception occurs in message consumption, put the message into the dead message queue. For example: Users who have successfully placed an order in the mall and clicked Go to Payment will be automatically invalidated if the payment is not made within a specified period of time..
2. Source of dead letters
-
Message TTL expiration: TTL is the abbreviation of Time To Live, which is the time to live.
-
The queue reaches the maximum length: the queue is full and no more data can be added to mq.
-
The message was rejected: (basic.reject or basic.nack) and requeue=false.
3. Dead letter practice
Dead letter TTL
(1) Consumer 1
① Define ordinary switches, dead letter switches, normal queues, and dead letter queues
②Declare dead letter switches, ordinary switches and ordinary queues
③Set the connection between the normal queue and the dead letter switch, using the argument parameter
④Binding: ordinary switch and normal queue, dead letter switch and dead letter queue
⑤Receive messages normally
public class Consumer01 {
//normal_exchange_name
private static final String NORMAL_EXCHANGE = "normal_exchange";
//Name of the dead letter switch
private static final String DEAD_EXCHANGE = "dead_exchange"; // Dead letter switch name
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
// Declare dead and normal exchanges of type direct
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
// Declare deadQueue
String deadQueue = "dead-queue";
channel.queueDeclare(deadQueue, false, false, false, null);
// dead letter queue binding: queue, switch, routingKey
channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
// Normal queue bind deadQueue information
Map<String, Object> params = new HashMap<>(); //Normal queue binds deadQueue information.
//normal queue set dead message switch parameter key is fixed value
params.put("x-dead-letter-exchange", DEAD_EXCHANGE);; //Normal queue set dead letter exchange parameter key is fixed value.
//normal queue set dead-letter routing-key parameter key is fixed value
params.put("x-dead-letter-routing-key", "lisi");; //normal queue set dead-letter routing-key parameter key is fixed value
//normal queue
String normalQueue = "normal-queue";
channel.queueDeclare(normalQueue, false, false, false, params);
channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan"); channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
System.out.println("Waiting to receive message ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Consumer01 received message" + message);
};
channel.basicConsume(normalQueue, true, deliverCallback, consumerTag -> {
});
}
}
(2) Producer:
①Define the name of a common switch
②Set the TTL of the message and the settings of the dead letter message
③Send messages normally
public class Producer {
private static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] argv) throws Exception {
Channel channel = RabbitMqUtils.getChannel(); channel.exchangeDeclare(channel); channel.exchangeDeclare(channel)
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
//Set the TTL time of the message to 10s.
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
//This information is used to demonstrate the limit of the number of queues.
for (int i = 1; i < 11; i++) {
String message = "info" + i;
channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
System.out.println("Producer sends message:" + message);
}
}
}
(3) Test effect 1:
Start consumer 1 first, then start the producer
Note: There are more DLX and DLK in the Features of a normal switch.
(4) Comparison of changes in the number of messages in normal queues and dead letter queues
When the TTL of the message in the normal queue arrives and has not been accepted, the message in the normal queue will be forwarded to the dead letter switch and then to the dead letter queue. This process is very fast.
(5) Consumer 2
Just receive messages from the dead letter queue
public class Consumer02 {
// Dead letter exchange name
private static final String DEAD_EXCHANGE = "dead_exchange";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
// Declare the exchange
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
// Declare the queue
String deadQueue = "dead-queue";
channel.queueDeclare(deadQueue, false, false, false, null);
channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi"); channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
System.out.println("Waiting to receive deadQueue message ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Consumer02 received message" + message);
};
channel.basicConsume(deadQueue, true, deliverCallback, consumerTag -> {
});
}
}
(6) Test effect 2:
Start consumer 2
Maximum length of dead letter
①The producer and consumer delete the TTL code
②Set the length limit of the normal queue in consumer 1
//Set the length limit of the normal queue, e.g. send 10, 4 will be a dead letter
params.put("x-max-length",6);
Note: Delete the originally generated queue first, because the condition parameters have changed and are no longer the same, otherwise an error will be reported.
Test results:
Dead letter message rejected
First comment out the code with the maximum length, and then receive all messages before continuing to send messages. Also delete the queue with the longest TTL and maximum length created previously to avoid affecting the test.
public class Consumer01 {
//normal_exchange_name
private static final String NORMAL_EXCHANGE = "normal_exchange";
//Name of the dead letter switch
private static final String DEAD_EXCHANGE = "dead_exchange"; // Dead letter switch name
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
// Declare dead and normal exchanges of type direct
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);
// Declare deadQueue
String deadQueue = "dead-queue";
channel.queueDeclare(deadQueue, false, false, false, null);
// dead letter queue binding: queue, switch, routingKey
channel.queueBind(deadQueue, DEAD_EXCHANGE, "lisi");
// Normal queue bind deadQueue information
Map<String, Object> params = new HashMap<>(); //Normal queue binds deadQueue information.
//normal queue set dead message switch parameter key is fixed value
params.put("x-dead-letter-exchange", DEAD_EXCHANGE);; //Normal queue set dead-letter-exchange parameter key is fixed value
//normal queue set dead-letter routing-key parameter key is fixed value
params.put("x-dead-letter-routing-key", "lisi");
// // Set the length limit of the normal queue, e.g. send 10, 4 is a dead letter
// params.put("x-max-length",6); // // normal queue length limit.
// Normal queue
String normalQueue = "normal-queue";
channel.queueDeclare(normalQueue, false, false, false, params);
channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan"); channel.queueBind(normalQueue, NORMAL_EXCHANGE, "zhangsan");
System.out.println("Waiting to receive message ........... ");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
if (message.equals("info5")) {
System.out.println("Consumer01 received the message " + message + "and refused to sign for the message");
//requeue is set to false to refuse to requeue, and the queue will be sent to the dead message queue if the switch is configured for dead messages.
channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
} else {
System.out.println("Consumer01 received message " + message); channel.basicAck(delivery.getEnvelope(.getDeliveryTag(), false); } else {
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
// Enable manual response
channel.basicConsume(normalQueue, false, deliverCallback, consumerTag -> {
});
}
}
Test results:
Message Queue - RabbitMQ: This is the end of the dead letter queue. It is not easy for the author to summarize and create. Let the big guys give me a 3rd party before taking off.