Dead-letter queue

concept

A dead letter is a message that cannot be consumed. Generally speaking, producer sends the message to the broker or directly to the queue, while a consumer takes the message from the queue and consumes it. However, some messages in the queue cannot be consumed ** due to certain reasons. Such messages become dead letters if there is no subsequent processing, and there is a dead letter queue naturally. Application scenario: To ensure that the message data of the order service is not lost, use the RabbitMQ dead-letter queue mechanism. When the message consumption is abnormal, the RabbitMQ dead-letter queue sends the message. And for example: users in the mall to place a successful order and click to pay after the specified time does not pay automatically invalid

The source of dead letters

Message TTL (live time) expired.
The queue reaches its maximum length (the queue is full and no more data can be added to MQ);
The message is rejected (basic.reject or basic.nack) and requeue=false;

Dead letter of actual combat

Code architecture diagram

Demonstrates a message entering a dead letter queue due to TTL expiration

producers

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/ * * *@author vleus
 * @date2021 07月24 22:23 * * Dead letter queue producer production message */
public class Producer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtils.getChannel();

        // Set TTL (time to Live)
        AMQP.BasicProperties basicProperties =
                new AMQP.BasicProperties()
                        .builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",basicProperties,message.getBytes(StandardCharsets.UTF_8)); }}}Copy the code

Consumer C1 (consumes messages in the normal queue)

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.util.HashMap;
import java.util.Map;

/ * * *@author vleus
 * @date2021 July 24 22:02 * < P > * Dead letter queue actual combat * Normal consumer 1 */
public class Consumer01 {

    // Define a common switch
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    // Dead letter queue switch
    public static final String DEAD_EXCHANGE = "dead_exchange";

    // Common queue name
    public static final String NORMAL_QUEUE = "normal_queue";
    // The name of the dead letter queue
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        // Get the channel
        Channel channel = RabbitMqUtils.getChannel();

        // Declare dead letter and common switch, type direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // Declare a normal queue
        Map<String, Object> argumentMap = new HashMap<>();
        // Set the expiration time for normal messages
        argumentMap.put("x-message-ttl".10000);
        // Who is the dead letter switch?
        argumentMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // Set a dead letter routingKey
        argumentMap.put("x-dead-letter-routing-key"."lisi");
        // Set the queue length limit for normal queues
// argumentMap.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE, false.false.false, argumentMap);

        // Declare a dead letter queue (messages are forwarded from a normal queue to a dead letter queue)
        channel.queueDeclare(DEAD_QUEUE, false.false.false.null);

        // Common switch and common queue binding
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        // Bind a dead letter switch to a dead letter queue
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        System.out.println("Normal consumer C1 waiting to receive message");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer01 receives the message:" + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> { }); }}Copy the code

Consumers C2

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.util.HashMap;
import java.util.Map;

/ * * *@author vleus
 * @date2021 July 24 22:02 * < P > * Dead letter queue actual combat * Normal consumer 1 */
public class Consumer02 {

    // The name of the dead letter queue
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        // Get the channel
        Channel channel = RabbitMqUtils.getChannel();

        System.out.println("Normal consumer C1 waiting to receive message");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer02 receives the message:" + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> { }); }}Copy the code

Demo message exceeds normal queue length and enters dead letter queue

producers

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/ * * *@author vleus
 * @date2021 07月24 22:23 * * Dead letter queue producer production message */
public class Producer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtils.getChannel();

        // Set TTL (time to Live)
// AMQP.BasicProperties basicProperties =
// new AMQP.BasicProperties()
// .builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan".null,message.getBytes(StandardCharsets.UTF_8)); }}}Copy the code

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.util.HashMap;
import java.util.Map;

/ * * *@author vleus
 * @date2021 July 24 22:02 * < P > * Dead letter queue actual combat * Normal consumer 1 */
public class Consumer01 {

    // Define a common switch
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    // Dead letter queue switch
    public static final String DEAD_EXCHANGE = "dead_exchange";

    // Common queue name
    public static final String NORMAL_QUEUE = "normal_queue";
    // The name of the dead letter queue
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        // Get the channel
        Channel channel = RabbitMqUtils.getChannel();

        // Declare dead letter and common switch, type direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // Declare a normal queue
        Map<String, Object> argumentMap = new HashMap<>();
        // Set the expiration time for normal messages
// argumentMap.put("x-message-ttl", 10000);
        // Who is the dead letter switch?
        argumentMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // Set a dead letter routingKey
        argumentMap.put("x-dead-letter-routing-key"."lisi");
        // Set the queue length limit for normal queues
        argumentMap.put("x-max-length".6);
        channel.queueDeclare(NORMAL_QUEUE, false.false.false, argumentMap);

        // Declare a dead letter queue (messages are forwarded from a normal queue to a dead letter queue)
        channel.queueDeclare(DEAD_QUEUE, false.false.false.null);

        // Common switch and common queue binding
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        // Bind a dead letter switch to a dead letter queue
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        System.out.println("Normal consumer C1 waiting to receive message");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer01 receives the message:" + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, consumerTag -> { }); }}Copy the code

Consumers C2

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.util.HashMap;
import java.util.Map;

/ * * *@author vleus
 * @date2021 July 24 22:02 * < P > * Dead letter queue actual combat * Normal consumer 1 */
public class Consumer02 {

    // The name of the dead letter queue
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        // Get the channel
        Channel channel = RabbitMqUtils.getChannel();

        System.out.println("Normal consumer C1 waiting to receive message");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer02 receives the message:" + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> { }); }}Copy the code

The demo message was denied entry to the dead-letter queue

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.impl.AMQBasicProperties;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.nio.charset.StandardCharsets;

/ * * *@author vleus
 * @date2021 07月24 22:23 * * Dead letter queue producer production message */
public class Producer {

    public static final String NORMAL_EXCHANGE = "normal_exchange";

    public static final String NORMAL_QUEUE = "normal_queue";

    public static void main(String[] args) throws Exception {

        Channel channel = RabbitMqUtils.getChannel();

        // Set TTL (time to Live)
// AMQP.BasicProperties basicProperties =
// new AMQP.BasicProperties()
// .builder().expiration("10000").build();

        for (int i = 1; i < 11; i++) {
            String message = "info" + i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan".null,message.getBytes(StandardCharsets.UTF_8)); }}}Copy the code

Consumers C1

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.util.HashMap;
import java.util.Map;

/ * * *@author vleus
 * @date2021 July 24 22:02 * < P > * Dead letter queue actual combat * Normal consumer 1 */
public class Consumer01 {

    // Define a common switch
    public static final String NORMAL_EXCHANGE = "normal_exchange";

    // Dead letter queue switch
    public static final String DEAD_EXCHANGE = "dead_exchange";

    // Common queue name
    public static final String NORMAL_QUEUE = "normal_queue";
    // The name of the dead letter queue
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        // Get the channel
        Channel channel = RabbitMqUtils.getChannel();

        // Declare dead letter and common switch, type direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT);

        // Declare a normal queue
        Map<String, Object> argumentMap = new HashMap<>();
        // Set the expiration time for normal messages
// argumentMap.put("x-message-ttl", 10000);
        // Who is the dead letter switch?
        argumentMap.put("x-dead-letter-exchange", DEAD_EXCHANGE);
        // Set a dead letter routingKey
        argumentMap.put("x-dead-letter-routing-key"."lisi");
        // Set the queue length limit for normal queues
// argumentMap.put("x-max-length", 6);
        channel.queueDeclare(NORMAL_QUEUE, false.false.false, argumentMap);

        // Declare a dead letter queue (messages are forwarded from a normal queue to a dead letter queue)
        channel.queueDeclare(DEAD_QUEUE, false.false.false.null);

        // Common switch and common queue binding
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE, "zhangsan");
        // Bind a dead letter switch to a dead letter queue
        channel.queueBind(DEAD_QUEUE, DEAD_EXCHANGE, "lisi");

        System.out.println("Normal consumer C1 waiting to receive message");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), "UTF-8");
            if (msg.equals("info5")) {
                System.out.println("Consumer01 receives the message:" + msg + ", this message was rejected.");
                // The consumer rejects the message. False indicates that the message is placed in a dead-letter queue
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            }else{
                System.out.println("Consumer01 receives the message:" + msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false); }};// If manual reply is enabled, the message is rejected
        channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> { }); }}Copy the code

Consumers C2

package com.vleus.rabbitmq.dead_exchange;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;

import java.util.HashMap;
import java.util.Map;

/ * * *@author vleus
 * @date2021 July 24 22:02 * < P > * Dead letter queue actual combat * Normal consumer 1 */
public class Consumer02 {

    // The name of the dead letter queue
    public static final String DEAD_QUEUE = "dead_queue";

    public static void main(String[] args) throws Exception {

        // Get the channel
        Channel channel = RabbitMqUtils.getChannel();

        System.out.println("Normal consumer C1 waiting to receive message");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("Consumer02 receives the message:" + new String(message.getBody(), "UTF-8"));
        };

        channel.basicConsume(DEAD_QUEUE, true, deliverCallback, consumerTag -> { }); }}Copy the code

mo4tech.com (Moment For Technology) is a global community with thousands techies from across the global hang out!Passionate technologists, be it gadget freaks, tech enthusiasts, coders, technopreneurs, or CIOs, you would find them all here.

RabbitMQ learning (6) – dead letter queue

Dead-letter queue

concept

The source of dead letters

Dead letter of actual combat

Code architecture diagram

Demonstrates a message entering a dead letter queue due to TTL expiration

Demo message exceeds normal queue length and enters dead letter queue

The demo message was denied entry to the dead-letter queue

RabbitMQ learning (6) – dead letter queue

Dead-letter queue

concept

The source of dead letters

Dead letter of actual combat

Code architecture diagram

Demonstrates a message entering a dead letter queue due to TTL expiration

Demo message exceeds normal queue length and enters dead letter queue

The demo message was denied entry to the dead-letter queue

Related Posts

Linux file systems and inodes

FFmpeg+Nginx converts video to HLS on-demand video stream

The average salary of programmers in July 2021 is 15,302 YUAN