Introduction to dead letter queue (two consumers, one producer)

Sort out the dead letter queue .

Messages that cannot be consumed

Dead letter source

news TTL Be overdue

The queue has reached its maximum length

The queue is full , No more messages can be added to mq

The news was rejected

50 Dead letter code architecture diagram

Direct switch zhangsan By C1 consumption

Dead letter switch lisi Add to dead-queue By C2 consumption

51 c1 consumer

public class Consumer01{
    
    // Common switch name 
    public static final String NORMAL_EXCHANGE="normal_exchange";
    // Dead letter switch 
    public static final String DEAD_EXCHANGE="dead_exchange";
    // Common queue name 
    public static final String NORMAL_QUEUE="normal_queue";
    // Dead letter queue name 
    public static final String DEAD_QUEUE="dead_queue";
    public static void main() throws Exception{
    
        
        Channel channel=RabbitMqUtil.getChannel();

        // Declare a common switch  / Dead letter switch 
        channel.exchangeDeclare(NORMAL_EXCHANGE,BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
        // Declare a normal queue 
        Map<String ,Object> arguments=new HashMap<>();
        // Expiration time 
        //arguments.put("x-message-ttl",1000000);
        // The dead letter switch after the normal queue setting expires 
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        // Set dead letter RoutingKey
        arguments.put("x-dead-letter-routing-key","lisi");
    	channel.queueDeclare(NORMAL_QUEUE,false,false,false,null);
        
        // Dead letter queue 
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
        // Bind common switches and queues 
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        
        // Bind dead letter switch and dead letter queue 
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
        sout(" Waiting to receive message ");
        // Callback function 
        DeliverCallback deliverCallback=(consumerTag,message)->{
    
            sout("Consumer01 When receiving messages "+new String(message.getBody(),"UTF-8"));
        };
        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumeTag->{
    });    
    }
}

52 producer

public class Producer{
    
    // Common switch name 
    public static final String NORMAL_EXCHANGE="normal_exchange";
    public static void main() throws Exception{
    
        Channel channel=RabbitMaUtils.getChannel();
        // Dead letter news 
AMQP.BasicProperties=new AMQP.BasicProperties().builder().expration("10000").build();
        
        // Dead letter news   Set up ttl Time 
        for(int i=1;i<11;i++){
    
            String message="info"+i;
    channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes());
		}
    }
}

Stop the consumer , The message is forwarded to the dead letter queue

53 consumer 2

This simple , It is simply to receive messages forwarded by ordinary queues for consumption .

public class Consumer02{
    
   
    // Dead letter queue name 
    public static final String DEAD_QUEUE="dead_queue";
    public static void main() throws Exception{
    
        Channel channel=RabbitMqUtil.getChannel();
        sout(" Waiting to receive message ");
        
        DeliverCallback deliverCallback=(consumerTag,message)->{
    
            sout("Consumer02 When receiving messages "+new String(message.getBody(),"UTF-8"));
        };
        channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,consumeTag->{
    });    
    }
}