一、基本概念
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) Mapparams = 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) Mapparams = 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) Mapparams = 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服务的死信队列当中,等待被其他的消费者进行处理!
需要注意的是:
只有针对消息的设置会放在消息发布方进行,队列等操作,因为发布方无法自己决定消息被路由到哪个队列,只能决定把消息交给哪个交换机,以及给定路由规则;
对于消息消费方而言,需要确定交换机、消息队列,已经完成 交换机和队列的绑定操作,所以针对于队列的设置都是放在消费方完成的!