运维做开发

这个屌丝很懒，什么也没留下！

热门标签

rabbitMQ引入死信队列_php rabbitmq 死信队列

作者：运维做开发 | 2024-08-05 00:33:15

踩

php rabbitmq 死信队列

一、基本概念

1、死信定义

指的是，从队列当中取出来的消息，到达消费方后，因为某些原因导致消息并没有被正常消费掉，这些没有被后续处理的消息就是“死信”，而保存死信的队列，就是死信队列。

2、死信出现的场景举例

为了保证订单业务的消息数据不丢失，需要使用死信队列机制，在消息消费发生异常的时候，将消息给投入到死信队列当中；

用户在商城下单成功并点进去准备支付，超过指定时间未支付时，消息自动失效成为死信（消息超时情况）；

3、死信的来源

消息TTL过期；

队列已经到达最大长度，数据无法再添加到MQ；

消息被拒绝了；

4、死信架构图

分析：

对于消息生产者而言，只需要关注将消息发送给交换机即可；

而对于普通消费者C1而言，需要关注普通交换机、普通队列、死信交换机的相关信息，要做两次绑定操作（普通交换机和普通队列绑定，普通队列和死信队列绑定），难点在于---普通队列怎么与死信交换机进行绑定？

而消费者C2也是一个普通消费者，专注于死信队列当中消息的处理，需要关注死信交换机、死信队列的信息，在死信交换机和死信队列绑定之后，从队列当中拿到死信进行处理；

二、代码部分

(0)提前准备工具类

封装获取MQ的connection方法，以及释放资源的方法


public class AMQPUtils {
    //用于获取客户端和MQ绑定的连接对象
    public static Connection getConnection() throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("/test");//用于隔离资源的虚拟主机
        factory.setHost("MQServer的ip地址");
        factory.setPort(5672);
        factory.setUsername("zhangsan");
        factory.setPassword("1234");
        Connection connection = factory.newConnection();
        return connection;
    }
 
    //释放资源
    public static void close(Channel channel,Connection connection) throws Exception{
        channel.close();
        connection.close();
    }
}

(一)模拟消息超时情况

1、消息发布者


public class Provider {
    public static void main(String[] args) throws Exception {
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明一个普通交换机
        String normalExchange = "normal_Exchange";
        channel.exchangeDeclare(normalExchange,"direct");
        String key = "zhangsan";
        //设置消息的ttl时间为5s
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("5000").build();
        //发布消息-发布多条消息验证
        for (int i = 1; i <= 10 ; i++) {
            channel.basicPublish(normalExchange,key,properties,("message---->"+i).getBytes());
        }
        //释放资源
        AMQPUtils.close(channel,connection);
    }
}

2、消费者C1

代码：


public class Consumer1 {
    public static void main(String[] args) throws Exception{
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //提前准备一些名字
        String normalExchange = "normal_exchange";
        String deadExchange = "dead_exchange";
        String normalQueue = "normal_queue";
        String deadQueue = "dead_queue";
        String normal_key = "zhangsan";
        String dead_key = "lisi";
        //声明普通交换机
        channel.exchangeDeclare(normalExchange,"direct");
        //声明死信交换机
        channel.exchangeDeclare(deadExchange,"direct");
        //设置普通队列当中需要携带的其他信息（死信交换机、死信队列、路由key）
        Map<String, Object> params = new HashMap<>();
        params.put("x-dead-letter-exchange",deadExchange);
        params.put("x-dead-letter-routing-key",dead_key);
        //声明普通队列
        channel.queueDeclare(normalQueue,false,false,false,params);
 
        //声明死信队列
        channel.queueDeclare(deadQueue,false,false,false,null);
 
        //binding
        //将普通交换机和普通队列绑定
        channel.queueBind(normalQueue,normalExchange,normal_key);
        //将将死信交换机和死信队列绑定
        channel.queueBind(deadQueue,deadExchange,dead_key);
 
        //消费消息
        channel.basicConsume(normalQueue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("C1消费的消息是---->"+new String(body));
            }
        });
    }
}

注意：开启C1接受普通交换机的消息之后，关闭C1，让普通队列当中的消息超过超时时间，成为死信，后被死信交换机路由进入dead_queue当中，下图所示的就是消息超时之后进入死信队列：

点击dead_queue可以查看具体的死信来源、交换机、路由key等信息；

此时：注意此时死信消息是保存在MQ当中的；

3、消费者c2

消费者C2要去消费死信队列当中的消息：


public class Consumer2 {
    public static void main(String[] args)throws Exception {
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //准备一些名字
        String deadExchange = "dead_exchange";
        String deadQueue = "dead_queue";
        String key = "lisi";
        //声明死信交换机
        channel.exchangeDeclare(deadExchange,"direct");
        //声明死信队列
        channel.queueDeclare(deadQueue,false,false,false,null);
        //交换机和队列绑定
        channel.queueBind(deadQueue,deadExchange,key);
        //消费死信消息
        channel.basicConsume(deadQueue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("C2消费了死信----->"+new String(body));
            }
        });
    }
}

控制台结果：

此时存储在死信队列当中的消息已经被C2消费了！

（二）模拟队列达到最大长度

请提前在MQ的控制台上，将情况1当中设置的队列给删除；

1、消息发布者


public class Provider {
    public static void main(String[] args) throws Exception {
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明一个普通交换机
        String normalExchange = "normal_exchange";
        channel.exchangeDeclare(normalExchange,"direct");
        String key = "zhangsan";
        //发布消息-发布多条消息验证
        for (int i = 1; i <= 10 ; i++) {
            channel.basicPublish(normalExchange,key,null,("message---->"+i).getBytes());
        }
        //释放资源
        AMQPUtils.close(channel,connection);
    }
}

2、消费者C1


public class Consumer1 {
    public static void main(String[] args) throws Exception{
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //提前准备一些名字
        String normalExchange = "normal_exchange";
        String deadExchange = "dead_exchange";
        String normalQueue = "normal_queue";
        String deadQueue = "dead_queue";
        String normal_key = "zhangsan";
        String dead_key = "lisi";
        //声明普通交换机
        channel.exchangeDeclare(normalExchange,"direct");
        //声明死信交换机
        channel.exchangeDeclare(deadExchange,"direct");
        //设置普通队列当中需要携带的其他信息（死信交换机、死信队列、路由key）
        Map<String, Object> params = new HashMap<>();
        params.put("x-dead-letter-exchange",deadExchange);
        params.put("x-dead-letter-routing-key",dead_key);
        //设置普通队列的最大长度
        params.put("x-max-length",6);
 
        //声明普通队列
        channel.queueDeclare(normalQueue,false,false,false,params);
 
        //声明死信队列
        channel.queueDeclare(deadQueue,false,false,false,null);
 
        //binding
        //将普通交换机和普通队列绑定
        channel.queueBind(normalQueue,normalExchange,normal_key);
        //将将死信交换机和死信队列绑定
        channel.queueBind(deadQueue,deadExchange,dead_key);
 
        //消费消息
        channel.basicConsume(normalQueue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("C1消费的消息是---->"+new String(body));
            }
        });
       
    }
}

开启C1，然后关闭，再启动消息生产者，结果：

3、消费者C2


public class Consumer2 {
    public static void main(String[] args)throws Exception {
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //准备一些名字
        String deadExchange = "dead_exchange";
        String deadQueue = "dead_queue";
        String key = "lisi";
        //声明死信交换机
        channel.exchangeDeclare(deadExchange,"direct");
        //声明死信队列
        channel.queueDeclare(deadQueue,false,false,false,null);
        //交换机和队列绑定
        channel.queueBind(deadQueue,deadExchange,key);
        //消费死信消息
        channel.basicConsume(deadQueue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("C2消费了死信----->"+new String(body));
            }
        });
    }
}

结果：

死信队列当中的消息已经被消费了；

（三）模拟消息被拒绝

1、消息发布者


public class Provider {
    public static void main(String[] args) throws Exception {
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //声明一个普通交换机
        String normalExchange = "normal_exchange";
        channel.exchangeDeclare(normalExchange,"direct");
        String key = "zhangsan";
        //发布消息-发布多条消息验证
        for (int i = 1; i <= 10 ; i++) {
            channel.basicPublish(normalExchange,key,null,("message---->"+i).getBytes());
        }
        //释放资源
        AMQPUtils.close(channel,connection);
    }
}

2、消费者C1


public class Consumer1 {
    public static void main(String[] args) throws Exception{
        Connection connection = AMQPUtils.getConnection();
        final Channel channel = connection.createChannel();
        //提前准备一些名字
        String normalExchange = "normal_exchange";
        String deadExchange = "dead_exchange";
        String normalQueue = "normal_queue";
        String deadQueue = "dead_queue";
        String normal_key = "zhangsan";
        String dead_key = "lisi";
        //声明普通交换机
        channel.exchangeDeclare(normalExchange,"direct");
        //声明死信交换机
        channel.exchangeDeclare(deadExchange,"direct");
        //设置普通队列当中需要携带的其他信息（死信交换机、死信队列、路由key）
        Map<String, Object> params = new HashMap<>();
        params.put("x-dead-letter-exchange",deadExchange);
        params.put("x-dead-letter-routing-key",dead_key);
 
        //声明普通队列
        channel.queueDeclare(normalQueue,false,false,false,params);
 
        //声明死信队列
        channel.queueDeclare(deadQueue,false,false,false,null);
 
        //binding
        //将普通交换机和普通队列绑定
        channel.queueBind(normalQueue,normalExchange,normal_key);
        //将将死信交换机和死信队列绑定
        channel.queueBind(deadQueue,deadExchange,dead_key);
 
        //消费消息--注意关闭自动确认
        channel.basicConsume(normalQueue,false,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                /*System.out.println("C1消费的消息是---->"+new String(body));*/
                //模拟消息被拒绝--把所有消息都拒绝
                channel.basicReject(envelope.getDeliveryTag(),false);
            }
        });
      
    }
}

注意：先启动C1，然后关闭，启动消息发布者，结果如下：

这10条消息目前都保存在MQ当中，然后再启动C1，把消息全部拒绝掉，让它们成为死信：

点击dead_queue，去查看死信队列当中的一些信息：

3、消费者C2


public class Consumer2 {
    public static void main(String[] args)throws Exception {
        Connection connection = AMQPUtils.getConnection();
        Channel channel = connection.createChannel();
        //准备一些名字
        String deadExchange = "dead_exchange";
        String deadQueue = "dead_queue";
        String key = "lisi";
        //声明死信交换机
        channel.exchangeDeclare(deadExchange,"direct");
        //声明死信队列
        channel.queueDeclare(deadQueue,false,false,false,null);
        //交换机和队列绑定
        channel.queueBind(deadQueue,deadExchange,key);
        //消费死信消息
        channel.basicConsume(deadQueue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("C2消费了死信----->"+new String(body));
            }
        });
    }
}

启动消费者C2，消费死信队列当中的消息！

三、小结

死信队列的出现，是为了保存因为特殊原因无法被消费的消息，避免消息直接失效！这些消息通过rabbitMQ的死信队列机制，可以保存在MQ服务的死信队列当中，等待被其他的消费者进行处理！

需要注意的是：

只有针对消息的设置会放在消息发布方进行，队列等操作，因为发布方无法自己决定消息被路由到哪个队列，只能决定把消息交给哪个交换机，以及给定路由规则；

对于消息消费方而言，需要确定交换机、消息队列，已经完成交换机和队列的绑定操作，所以针对于队列的设置都是放在消费方完成的！

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/运维做开发/article/detail/930116