rabbitmq死信队列详解（亲手实践）_php rabbitmq 死信队列

原文链接：rabbitmq死信队列详解（亲手实践） – 编程屋

目录

1 概念

2 成为死信队列的条件

2.1 队列指定长度 

2.2  消息ttl时间

2.3 消费者拒收消息

---

1 概念

死信队列：死信队列其实和普通的队列一样，只不过里面存放的消息都是普通队列过期没有消费的。所以，接收没有及时被消费消息的队列为死信队列。

2 成为死信队列的条件

以下条件只要满足一条，即可以成为死信队列。

1. 队列长度满了：排在前面的消息会被拒收或者进入死信交换机
2. 消息的ttl时间到了：消息超时未被消费
3. 消息被拒收了：手动拒绝消息

一个队列设置了队列长度或者过期时间或被拒收，并且设置了死信队列的交换机和死信的路由key。那么消息满足条件就会进入死信队列。

例如：

@Bean("queueB")
public Queue queueB(){
    Map<String, Object> arguments  = new HashMap<>();
    //设置死信交换机
    arguments.put("x-dead-letter-exchange",Y_DEAD_LETTER_XCHANGE);
    //设置死信RoutingKey
    arguments.put("x-dead-letter-routing-key","YD");
    //设置ttl
    arguments.put("x-message-ttl",40000);

    return QueueBuilder.durable(QUEUE_B).withArguments(arguments).build();
}

以上只是声明了一个普通队列queueB，然后在该队列设置了过期时间40s，和死信交换机和死信路由key。

注意：死信队列就是一个普通的队列，只不过声明普通队列的时候指定了死信交换机，二者才产生了联系

2.1 队列指定长度 

配置相关队列和交换机

注意：声明一个队列分为3步（声明交换机、声明队列、将队列和交换机路由绑定）

package com.liubujun.rabbitmqspringbootdemo.config;
 
import com.rabbitmq.client.AMQP;
import com.sun.javafx.collections.MappingChange;
import jdk.nashorn.internal.objects.NativeUint8Array;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
 
import java.util.HashMap;
import java.util.Map;
 
/**
 * @Author: liubujun
 * @Date: 2023/6/3 16:15
 */
 
@Component
public class DeadQueueConfig {
 
    //队列
    public static final String FORMAL_QUEUE = "formal_queue";
    public static final String DEAD_QUEUE = "dead_queue";
 
    //交换机
    public static final String FORMAL_EXCHANGE = "formal_exchange";
    public static final String DEAD_EXCHANGE = "dead_exchaneg";
 
    //路由key
    public static final String FORMAL_ROUNTE_KEY = "formal_rounte_key";
    public static final String DEAD_ROUNTE_KEY = "dead_route_key";
 
    /**
     * 普通队列交换机声明
     * 交换机类型：topic：处理路由键，按模式匹配，向符合规则的队列投递消息
     * name 交换机名称
     * durable 是否持久化
     * autoDelete 是否删除
     * arguments 用于设置其他参数
     * @return
     */
    @Bean
    public Exchange getFormalExchange(){
        return new TopicExchange(FORMAL_EXCHANGE,true,false,null);
    }
 
    /**
     * 声明普通队列，并设置与死信队列联系
     * name 队列名称
     * durable 是否持久化
     * exclusive 是否排外 如果是排外的，该队列 仅对首次声明它的连接（Connection）可见，是该Connection私有的，
     * 类似于加锁，并在连接断开connection.close()时自动删除
     * autoDelete 是否删除
     * arguments 用于设置其他参数
     * @return
     */
    @Bean
    public Queue getFormalQueue(){
        Map<String, Object> map = new HashMap<>();
        //设置队列最大长度
        map.put("x-max-length",5);
        //设置死信队列交换机
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信队列路由key
        map.put("x-dead-letter-routing-key",DEAD_ROUNTE_KEY);
        return new Queue(FORMAL_QUEUE,true,false,false,map);
    }
 
    /**
     * 将普通队列和交换机绑定
     * destination：目标队列或交换器
     * destinationType：DesdinationType指出目标是交换器还是对列
     * exchange：交换机
     * routingKey：路由key
     * arguments：参数设置
     * @return
     */
    @Bean
    public Binding bingFormalQueue(){
        return new Binding(FORMAL_QUEUE, Binding.DestinationType.QUEUE,FORMAL_EXCHANGE,FORMAL_ROUNTE_KEY,null);
    }
 
 
    /**
     * 声明死信队列交换机
     * @return
     */
    @Bean
    public Exchange getDeadExchange(){
        return new TopicExchange(DEAD_EXCHANGE,true,false,null);
    }
 
 
    /**
     * 声明死信队列
     * @return
     */
    @Bean
    public Queue getDeadQueue(){
        return new Queue(DEAD_QUEUE,true,false,false, null);
    }
 
    /**
     * 将死信队列和交换机绑定
     * @return
     */
    @Bean
    public Binding bingDeadQueue(){
        return new Binding(DEAD_QUEUE, Binding.DestinationType.QUEUE,DEAD_EXCHANGE,DEAD_ROUNTE_KEY,null);
    }
 
 
 
 
}

生产者：发送6条消息，看rabbitmq中队列变化

    @GetMapping("/sendMessageTtl/{message}")
    public void sendMessageTtl(@PathVariable String message){
        log.info("当前时间发送:{},发送5条消息给两个TTL队列:{}",new Date().toString(),message);
        for (int i = 0; i < 6; i++) {
            rabbitTemplate.convertAndSend(DeadQueueConfig.FORMAL_EXCHANGE,DeadQueueConfig.FORMAL_ROUNTE_KEY,message);
        }
    }

rabbitmq控制台：

普通队列有5条消息，而死信队列有1条消息。

因为在声明普通队列的时候，已经说明了队列最大长度为5，那么多余的消息就会根据配置的参数找到对应的交换机进而找到对应的路由，然后路由到对应的队列（死信队列） 。

2.2  消息ttl时间

继续沿用上面的配置，只不过修改下普通队列的参数。

    @Bean
    public Queue getFormalQueue(){
        Map<String, Object> map = new HashMap<>();
        //设置队列超时时间
        map.put("x-message-ttl",5000);
        //设置死信队列交换机
        map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信队列路由key
        map.put("x-dead-letter-routing-key",DEAD_ROUNTE_KEY);
        return new Queue(FORMAL_QUEUE,true,false,false,map);
    }

发送消息：

可以发现，发送给普通队列的消息，超时没有被消费，都进入到了死信队列中。 

2.3 消费者拒收消息

沿用上面的配置，并在声明普通队列的时候去掉消息的过期时间。

注意：需要在rabbitmq控制台删除队列，不然项目启动会报错。

添加消费者：

@Slf4j
@Component
public class DeadQueueConsumer {
 
    /**
     * 监听死信队列
     */
//    @RabbitListener(queues = DeadQueueConfig.DEAD_QUEUE)
//    public void listenDeadQueue(Message message, Channel channel){
//    log.info("接收到死信队列消息:{}",message.getBody());
//    }
 
    /**
     * 监听普通队列
     */
    @RabbitListener(queues = "formal_queue")
    public void listenFormalQueue(Message message, Channel channel) throws IOException {
        log.info("接收到普通队列消息:{}",message.getBody());
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //拒绝消息
        channel.basicReject(deliveryTag,false);
    }
}

rabbitmq控制台结果： 

消息在消费者端被拒收后，直接被放进了死信队列。

以上只是部分内容，为了维护方便，本文已迁移到新地址：rabbitmq死信队列详解（亲手实践） – 编程屋