网站制作公司要文化建设费,做商铺最好的网站,wordpress lapa,用凡科做的手机网站版死信队列
死信的概念
先从概念解释上搞清楚这个定义#xff0c;死信#xff0c;顾名思义就是无法被消费的消息#xff0c;字面意思可以这样理解#xff0c;一般来说#xff0c;producer 将消息投递到 broker 或者直接到queue 里了#xff0c;consumer 从 queue 取出消息…死信队列
死信的概念
先从概念解释上搞清楚这个定义死信顾名思义就是无法被消费的消息字面意思可以这样理解一般来说producer 将消息投递到 broker 或者直接到queue 里了consumer 从 queue 取出消息进行消费但某些时候由于特定的原因导致queue 中的某些消息无法被消费这样的消息如果没有后续的处理就变成了死信有死信自然就有了死信队列.
应用场景:为了保证订单业务的消息数据不丢失需要使用到 RabbitMQ 的死信队列机制当消息消费发生异常时将消息投入死信队列中.还有比如说: 用户在商城下单成功并点击去支付后在指定时间未支付时自动失效
造成死信的原因
消息 TTL 过期队列达到最大长度(队列满了无法再添加数据到 MQ 中)消息被拒绝(basic.reject 或 basic.nack)并且 requeuefalse.
死信架构图 代码实战 TTL过期 package com.vmware.rabbit.demo8;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.vmware.rabbit.utils.RabbitUtil;import java.util.HashMap;public class Consumer {//普通交换机private static final String NORMAL_EXCHANGE normal_exchange;//死信交换机private static final String DEAD_EXCHANGE dead_exchange;//普通队列private static final String NORMAL_QUEUE normal_queue;//死信队列private static final String DEAD_QUEUE dead_queue;public static void main(String[] args) throws Exception {Connection connection RabbitUtil.getConnection();Channel channel connection.createChannel();//创建普通队列死信分发参数HashMapString,Object arguments new HashMap();arguments.put(x-dead-letter-exchange,DEAD_EXCHANGE);arguments.put(x-dead-letter-routing-key,lisi);//创建普通交换机与队列并绑定channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,zhangsan);//创建死信交换机和队列并绑定channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);channel.queueDeclare(DEAD_QUEUE,false,false,false,null);channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,lisi);DeliverCallback deliverCallback (tag,msg)-{String message new String(msg.getBody());System.out.println(接收到消息:message);}};//创建消费者channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,(tag)-{});}
}package com.vmware.rabbit.demo8;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.vmware.rabbit.utils.RabbitUtil;public class Producer {private static final String EXCHANGE_NAME normal_exchange;private static final String ROUTING_KEY zhangsan;public static void main(String[] args) throws Exception {Connection connection RabbitUtil.getConnection();System.out.println(已连接到RabbitMQ服务器....);Channel channel connection.createChannel();//设置超时为10秒AMQP.BasicProperties basicProperties new AMQP.BasicProperties().builder().expiration(10000).build();for (int i 0; i 10; i) {String message msg i;channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, basicProperties, message.getBytes());System.out.println(消息:message发送成功!);}}
}队列达到最大长度 package com.vmware.rabbit.demo8;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.vmware.rabbit.utils.RabbitUtil;import java.util.HashMap;public class Consumer {//普通交换机private static final String NORMAL_EXCHANGE normal_exchange;//死信交换机private static final String DEAD_EXCHANGE dead_exchange;//普通队列private static final String NORMAL_QUEUE normal_queue;//死信队列private static final String DEAD_QUEUE dead_queue;public static void main(String[] args) throws Exception {Connection connection RabbitUtil.getConnection();Channel channel connection.createChannel();//创建普通队列死信分发参数HashMapString,Object arguments new HashMap();arguments.put(x-dead-letter-exchange,DEAD_EXCHANGE);arguments.put(x-dead-letter-routing-key,lisi);//设置队列最大长度arguments.put(x-max-length,5);//创建普通交换机与队列并绑定channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,zhangsan);//创建死信交换机和队列并绑定channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);channel.queueDeclare(DEAD_QUEUE,false,false,false,null);channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,lisi);DeliverCallback deliverCallback (tag,msg)-{String message new String(msg.getBody());System.out.println(消息:message被拒绝);};//创建消费者channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,(tag)-{});}
}package com.vmware.rabbit.demo8;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.vmware.rabbit.utils.RabbitUtil;public class Producer {private static final String EXCHANGE_NAME normal_exchange;private static final String ROUTING_KEY zhangsan;public static void main(String[] args) throws Exception {Connection connection RabbitUtil.getConnection();System.out.println(已连接到RabbitMQ服务器....);Channel channel connection.createChannel();for (int i 0; i 10; i) {String message msg i;channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes());System.out.println(消息:message发送成功!);}}
}消息被拒 package com.vmware.rabbit.demo8;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.vmware.rabbit.utils.RabbitUtil;import java.util.HashMap;public class Consumer {//普通交换机private static final String NORMAL_EXCHANGE normal_exchange;//死信交换机private static final String DEAD_EXCHANGE dead_exchange;//普通队列private static final String NORMAL_QUEUE normal_queue;//死信队列private static final String DEAD_QUEUE dead_queue;public static void main(String[] args) throws Exception {Connection connection RabbitUtil.getConnection();Channel channel connection.createChannel();//创建普通队列死信分发参数HashMapString,Object arguments new HashMap();arguments.put(x-dead-letter-exchange,DEAD_EXCHANGE);arguments.put(x-dead-letter-routing-key,lisi);//创建普通交换机与队列并绑定channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,zhangsan);//创建死信交换机和队列并绑定channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);channel.queueDeclare(DEAD_QUEUE,false,false,false,null);channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,lisi);DeliverCallback deliverCallback (tag,msg)-{String message new String(msg.getBody());if (message.equals(msg5)){System.out.println(消息:message被拒绝);channel.basicReject(msg.getEnvelope().getDeliveryTag(),false);}else {System.out.println(接收到消息:message);channel.basicAck(msg.getEnvelope().getDeliveryTag(),false);}};//创建消费者channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,(tag)-{});}
}package com.vmware.rabbit.demo8;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import com.vmware.rabbit.utils.RabbitUtil;public class Consumer2 {private static final String DEAD_QUEUE_NAME dead_queue;public static void main(String[] args)throws Exception {Connection connection RabbitUtil.getConnection();Channel channel connection.createChannel();DeliverCallback deliverCallback(tag,msg)-{String message new String(msg.getBody());System.out.println(队列:DEAD_QUEUE_NAME\t收到消息:message);};channel.basicConsume(DEAD_QUEUE_NAME,true,deliverCallback,(tag)-{});}
}package com.vmware.rabbit.demo8;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.vmware.rabbit.utils.RabbitUtil;public class Producer {private static final String EXCHANGE_NAME normal_exchange;private static final String ROUTING_KEY zhangsan;public static void main(String[] args) throws Exception {Connection connection RabbitUtil.getConnection();System.out.println(已连接到RabbitMQ服务器....);Channel channel connection.createChannel();for (int i 0; i 10; i) {String message msg i;channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, message.getBytes());System.out.println(消息:message发送成功!);}}
}
