RabbitMQ (4)

RabbitMQ (4)

文章目录

前言

上文我们已经学习完 交换机 ，知道了几个交换机的使用 ，下面我们来学习一下 死信队列

1. 死信的概念

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说，producer 将消息投递到 broker 或者直接到queue 里了，consumer 从 queue 取出消息 进行消费，但某些时候由于特定的原因导致 queue 中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景：为了保证订单业务的消息数据不丢失，需要使用到 RabbitMQ 的死信队列机制，当消息消费发生异常时，将消息投入死信队列中。还有比如说：用户在商城下单成功并点击去支付后在指定时间未支付时自动失效。

2. 死信的来源

1. 消息 TTL 过期 : TTL 是 Time To Live 的缩写， TTL 就是 生存时间
2. 队列达到最长长度 : 队列满了 ， 无法添加数据到 MQ 中
3. 消息被拒绝 （basic.reject 或 basic.nack） 并且 requeue = false

3. 死信代码案例

这里 创建一个 direct 交换机 ，两个消费者 ， 一个生产者 ， 两个 队列 (一个为 消息队列 ， 一个为死信队列)

图:

代码 :

3.1 TTL 过期时间

生产者：

packageorg.example.seven;importcom.rabbitmq.client.AMQP;importcom.rabbitmq.client.Channel;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;// 生产者publicclassProducer{// 普通交换机的名称publicstaticfinalStringNORMAL_EXCHANGE="normal_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 设置消息的过期时间 （TTL） 单位是 ms --> 设置消息的过期时间为 10sAMQP.BasicProperties properties =newAMQP.BasicProperties().builder().expiration("10000").build();for(int i =1; i <11; i++){String message ="info"+ i;
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan", properties, message.getBytes());}}}

消费者 c1 （启动之后关闭该消费者， 模拟其接受不到消息）

packageorg.example.seven;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.HashMap;importjava.util.Map;importjava.util.Objects;importjava.util.concurrent.TimeoutException;publicclassConsumer01{// 普通交换机的名称publicstaticfinalStringNORMAL_EXCHANGE="normal_exchange";// 死信交换机的名称publicstaticfinalStringDEAD_EXCHANGE="dead_change";// 普通队列的名称publicstaticfinalStringNORMAL_QUEUE="normal_queue";// 死刑队列的名称publicstaticfinalStringDEAD_QUEUE="dead_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 声明死信和普通交换机 , 类型为 direct （直接交换机）// 普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE,BuiltinExchangeType.DIRECT);// 死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);// 声明普通队列Map<String,Object> arguments =newHashMap<>();// 过期时间 10s 由生产者指定 更加灵活//        arguments.put("x-message-ttl", 10000);// 正常的队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);// 设置死信 路由键 (routingKey)
        arguments.put("x-dead-letter-routing-key","lisi");// 声明队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false, arguments);// 申明死刑队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);// 绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");// 绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");System.out.println("等待接受消息");DeliverCallback deliverCallback =(tag, message)->{System.out.println("C1 接收到的消息为: "+newString(message.getBody(),"UTF-8"));};
        channel.basicConsume(NORMAL_QUEUE,true, deliverCallback,(tag)->{});}}

先启动消费者 C1，创建出队列，然后停止该 C1 的运行，则 C1 将无法收到队列的消息，无法收到的消息 10 秒后进入死信队列。启动生产者 producer 生产消息

c1 看完，我们在来写 c2 消费者 ，将进入到死信队列的消息 进行消费.

消费者c2

packageorg.example.seven;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassConsumer02{// 死信队列的名称publicstaticfinalStringDEAD_QUEUE="dead_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();System.out.println("等待接受死信消息......");DeliverCallback deliverCallback =(tag, message)->{System.out.println("C2 接收到的消息： "+newString(message.getBody(),"UTF-8"));};

        channel.basicConsume(DEAD_QUEUE,true, deliverCallback,(tag)->{});}}

图:

看完 消息过期后 ，消息转发到 死信队列 被 c2 消费，下面我们来 尝试使用 死信最大长度 (队列满了，将多的消息转发到死信队列中）

3.2 超过队列最大长度

消息生产者代码 去掉 TTL 属性 ， 将

basicPublish

的第三个参数改为

null

生产者:

packageorg.example.seven;importcom.rabbitmq.client.AMQP;importcom.rabbitmq.client.Channel;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;// 生产者publicclassProducer{// 普通交换机的名称publicstaticfinalStringNORMAL_EXCHANGE="normal_exchange";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 设置消息的过期时间 （TTL） 单位是 ms --> 设置消息的过期时间为 10s//        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();for(int i =1; i <11; i++){String message ="info"+ i;//            channel.basicPublish(NORMAL_EXCHANGE, "zhangsan", properties, message.getBytes());
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null, message.getBytes());}}}

c1 消费者 (启动之后关闭该消费者 模拟其接收不到消息)

packageorg.example.seven;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.HashMap;importjava.util.Map;importjava.util.Objects;importjava.util.concurrent.TimeoutException;publicclassConsumer01{// 普通交换机的名称publicstaticfinalStringNORMAL_EXCHANGE="normal_exchange";// 死信交换机的名称publicstaticfinalStringDEAD_EXCHANGE="dead_change";// 普通队列的名称publicstaticfinalStringNORMAL_QUEUE="normal_queue";// 死刑队列的名称publicstaticfinalStringDEAD_QUEUE="dead_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 声明死信和普通交换机 , 类型为 direct （直接交换机）// 普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE,BuiltinExchangeType.DIRECT);// 死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);// 声明普通队列Map<String,Object> arguments =newHashMap<>();// 过期时间 10s 由生产者指定 更加灵活//        arguments.put("x-message-ttl", 10000);// 正常的队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);// 设置死信 路由键 (routingKey)
        arguments.put("x-dead-letter-routing-key","lisi");// 设置队列的限制 , 例如 发送 10 个消息 , 6 个为正常 , 4 个为死信
        arguments.put("x-max-length",6);// 声明队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false, arguments);// 申明死刑队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);// 绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");// 绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");System.out.println("等待接受消息");DeliverCallback deliverCallback =(tag, message)->{System.out.println("C1 接收到的消息为: "+newString(message.getBody(),"UTF-8"));};
        channel.basicConsume(NORMAL_QUEUE,true, deliverCallback,(tag)->{});}}

注意:

这参数改变了(没有设置 ttl 时间，新增了 队列的 最大长度限制 为 6) ，所以 需要把原来队列删除

消费者c2 代码不变

packageorg.example.seven;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassConsumer02{// 死信队列的名称publicstaticfinalStringDEAD_QUEUE="dead_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();System.out.println("等待接受死信消息......");DeliverCallback deliverCallback =(tag, message)->{System.out.println("C2 接收到的消息： "+newString(message.getBody(),"UTF-8"));};

        channel.basicConsume(DEAD_QUEUE,true, deliverCallback,(tag)->{});}}

效果:

这里 之所以要启动 c1 后在关闭，是为了展示 6个消息放到 普通队列 ，4个消息放到死信队列， 如果不这么做，发送的 10个消息 都会被 c1 消费 ，（消息发送到 队列后 , 立马 转发给 c1 导致 队列就不会达到 6 个 ，队列不会满 ，也就不会将消息转化给 死信队列).

3.3 拒绝消息

消息生产者 和 消费者 c2 与上面的代码一样

这里我们 拒绝 info7 消息 ，想要 拒绝 info7 消息，我们可以采用手动应答.

消费者c1

packageorg.example.seven;importcom.rabbitmq.client.BuiltinExchangeType;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.HashMap;importjava.util.Map;importjava.util.Objects;importjava.util.concurrent.TimeoutException;publicclassConsumer01{// 普通交换机的名称publicstaticfinalStringNORMAL_EXCHANGE="normal_exchange";// 死信交换机的名称publicstaticfinalStringDEAD_EXCHANGE="dead_change";// 普通队列的名称publicstaticfinalStringNORMAL_QUEUE="normal_queue";// 死刑队列的名称publicstaticfinalStringDEAD_QUEUE="dead_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 声明死信和普通交换机 , 类型为 direct （直接交换机）// 普通交换机
        channel.exchangeDeclare(NORMAL_EXCHANGE,BuiltinExchangeType.DIRECT);// 死信交换机
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);// 声明普通队列Map<String,Object> arguments =newHashMap<>();// 过期时间 10s 由生产者指定 更加灵活//        arguments.put("x-message-ttl", 10000);// 正常的队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);// 设置死信 路由键 (routingKey)
        arguments.put("x-dead-letter-routing-key","lisi");// 设置队列的限制 , 例如 发送 10 个消息 , 6 个为正常 , 4 个为死信//        arguments.put("x-max-length", 6);// 声明队列
        channel.queueDeclare(NORMAL_QUEUE,false,false,false, arguments);// 申明死刑队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);// 绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");// 绑定死信的交换机与死信的队列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");System.out.println("等待接受消息");DeliverCallback deliverCallback =(tag, message)->{String msg =newString(message.getBody(),"UTF-8");if(msg.equals("info7")){System.out.println("C1 接收到消息为: "+ msg +" 此消息被 C1 拒绝");//requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);}else{System.out.println("C1 接收到的消息为: "+ msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);}};//        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback, (tag) -> {});// 开启 手动应答
        channel.basicConsume(NORMAL_QUEUE,false, deliverCallback,(tag)->{});}}