springboot中RabbitMQ死信队列的实现示例
作者:Java小生不才
死信队列是一种特殊的消息队列,用来存储无法被正常消费的消息,常被用来实现延迟处理,异常消息处理等,本文主要介绍了springboot中RabbitMQ死信队列的实现示例,感兴趣的可以了解一下
1. 死信队列
死信队列是一种特殊的消息队列,用来存储无法被正常消费的消息,常被用来实现延迟处理,异常消息处理等,提高了系统的可伸缩性和容错性,能够应对高并发和异常消息。
死信队列中的消息被称为死信消息,用来分发死信消息的交换机被称为死信交换机(Dead Letter Exchange,DLX)。
死信队列在实际项目中的应用场景有很多如:
- 订单超时未支付,将此消息放入死信队列中,等待后续处理(延迟等待)
- 消息消费失败将消息放入死信队列中进行重试(消息重试机制)
2.正常消息成为死信消息的条件
- 消息到了过期时间仍然未被消费者消费
- 队列已满无法保存新消息
- 消息被拒绝消费且未设置重新放入队列
3.消费者1
package com.hong.rabbitmq9; import com.hong.utils.RabbitMQUtil; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.CancelCallback; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.Map; /** * @Description: 死信队列消费者1 * @Author: hong * @Date: 2024-01-17 21:04 * @Version: 1.0 **/ public class Consumer1 { //正常交换机名称 public static final String NORMAL_EXCHANGE = "normal_exchange"; //死信交换机名称 public static final String DEAD_EXCHANGE = "dead_exchange"; //正常队列名称 public static final String NORMAL_QUEUE = "normal_queue"; //死信队列名称 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMQUtil.getChannel(); channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); channel.queueDeclare(DEAD_QUEUE,false,false,false,null); channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi"); //正常队列绑定死信队列信息 Map<String,Object> map = new HashMap<>(); map.put("x-dead-letter-exchange",DEAD_EXCHANGE); map.put("x-dead-letter-routing-key", "lisi"); //声明正常队列 channel.queueDeclare(NORMAL_QUEUE,false,false,false,map); channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan"); System.out.println("Consumer1等待接收消息:"); DeliverCallback deliverCallback = (comsumerTag, message) -> { System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8)); }; CancelCallback cancelCallback = var -> { }; channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback); } }
4.生产者
package com.hong.rabbitmq9; import com.hong.utils.RabbitMQUtil; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; /** * @Description: 死信队列消息生产者 * @Author: hong * @Date: 2024-01-17 20:49 * @Version: 1.0 **/ public class Producer { public static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] args) throws Exception { Channel channel = RabbitMQUtil.getChannel(); channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); //设置消息的 TTL 时间 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()); System.out.println("生产者发送消息:" + message); } } }
启动消费者1后马上关闭消费者1,模拟消费者1接收不到消息,再启动生产者
生产者发送10条正常队列中有10条消息
10s后正常队列中的消息由于没有消费者消费进入死信队列中
5.消费者2
package com.hong.rabbitmq9; import com.hong.utils.RabbitMQUtil; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.CancelCallback; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.Map; /** * @Description: 死信队列-死信消费者 * @Author: hong * @Date: 2024-01-17 21:31 * @Version: 1.0 **/ public class Consumer2 { public static final String DEAD_EXCHANGE = "dead_exchange"; public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMQUtil.getChannel(); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); channel.queueDeclare(DEAD_QUEUE,false,false,false,null); channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi"); System.out.println("Consumer2等待接收死信消息:"); DeliverCallback deliverCallback = (comsumerTag, message) -> { System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8)); }; CancelCallback cancelCallback = var -> { }; channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback); } }
启动消费者2,使其消费死信队列中的消息
6.队列达到最大长度
以上代码是TTL,队列达到最大长度只要将上述代码稍微改动一下即可
6.1.注释掉生产者代码中的ttl部分
package com.hong.rabbitmq9; import com.hong.utils.RabbitMQUtil; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.Channel; /** * @Description: 死信队列消息生产者 * @Author: hong * @Date: 2024-01-17 20:49 * @Version: 1.0 **/ public class Producer { public static final String NORMAL_EXCHANGE = "normal_exchange"; public static void main(String[] args) throws Exception { Channel channel = RabbitMQUtil.getChannel(); channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); //设置消息的 TTL 时间 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", null, message.getBytes()); System.out.println("生产者发送消息:" + message); } } }
6.2.消费者1代码中加最大长度
package com.hong.rabbitmq9; import com.hong.utils.RabbitMQUtil; import com.rabbitmq.client.BuiltinExchangeType; import com.rabbitmq.client.CancelCallback; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DeliverCallback; import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.Map; /** * @Description: 死信队列消费者1 * @Author: hong * @Date: 2024-01-17 21:04 * @Version: 1.0 **/ public class Consumer1 { //正常交换机名称 public static final String NORMAL_EXCHANGE = "normal_exchange"; //死信交换机名称 public static final String DEAD_EXCHANGE = "dead_exchange"; //正常队列名称 public static final String NORMAL_QUEUE = "normal_queue"; //死信队列名称 public static final String DEAD_QUEUE = "dead_queue"; public static void main(String[] args) throws Exception { Channel channel = RabbitMQUtil.getChannel(); channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT); channel.exchangeDeclare(DEAD_EXCHANGE, BuiltinExchangeType.DIRECT); channel.queueDeclare(DEAD_QUEUE,false,false,false,null); channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi"); //正常队列绑定死信队列信息 Map<String,Object> map = new HashMap<>(); map.put("x-dead-letter-exchange",DEAD_EXCHANGE); map.put("x-dead-letter-routing-key", "lisi"); map.put("x-max-length",8); //声明正常队列 channel.queueDeclare(NORMAL_QUEUE,false,false,false,map); channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan"); System.out.println("Consumer1等待接收消息:"); DeliverCallback deliverCallback = (comsumerTag, message) -> { System.out.println( "routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8)); }; CancelCallback cancelCallback = var -> { }; channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback); } }
启动消费者1后立马关闭,模拟队列已满
到此这篇关于springboot中RabbitMQ死信队列的实现示例的文章就介绍到这了,更多相关springboot RabbitMQ死信队列内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!