在国外做网站网站犯法吗,商城网站开发视频教程,场口一站式建站哪家公司好,网页制作工具按其制作方式分为什么是死信队列
由于特定的**原因导致 Queue 中的某些消息无法被消费#xff0c;**这类消费异常的数据将会保存在死信队列中防止消息丢失#xff0c;例如用户在商城下单成功并点击支付后#xff0c;在指定时间未支付时的订单自动失效死信队列只不过是绑定在死信交换机上的队…什么是死信队列
由于特定的**原因导致 Queue 中的某些消息无法被消费**这类消费异常的数据将会保存在死信队列中防止消息丢失例如用户在商城下单成功并点击支付后在指定时间未支付时的订单自动失效死信队列只不过是绑定在死信交换机上的队列。死信交换机只不过是用来接受死信的交换机可以为任何类型【Direct、Fanout、Topic】。一般来说会为每个业务队列分配一个独有的路由key并对应的配置一个死信队列进行监听也就是说一般会为每个重要的业务队列配置一个死信队列。
来源
消息的存活时间到了ttl过期队列积压的消息达到最大长度在队列中等待时间最久的消息会成为死信消息被拒消费方返回nack进行否定应答且不重新加入队列requeuefalse)
演示
模拟一条正常应该被C1消费者接收的消息由于出现消费异常情况进入死信队列被C2消费者进行消费的案例
架构图 TTL过期
在生产者方进行指定为当前发送消息的过期时间缺点是消息即使过期也不一定会被马上丢弃因为消息是否过期是在即将投递到消费者之前判定的** 如果当前队列有严重的消息积压情况已过期的消息依旧会被积压在队列中如果队列配置了消息积压上限**将导致后续应当正常消费的消息全部进入死信队列 在队列指定为所有到达该队列的消息的过期时间时间从消息入队列开始计算只要超过了队列的超时时间配置消息会自动清除
该案例为在生产者方进行指定配置类
package com.example.rabbitmqqoslimiting.demos;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class RabbitUtils {private static final ConnectionFactory connectionFactory; //放到静态代码块中在类加载时执行只执行一次。达到工厂只创建一次每次获取是新连接的效果static {//创建连接工厂connectionFactory new ConnectionFactory();connectionFactory.setHost(101.133.141.74); //设置MQ的主机地址connectionFactory.setPort(5672); //设置MQ服务端口connectionFactory.setVirtualHost(study); //设置Virtual Hosts(虚拟主机)connectionFactory.setUsername(guest); //设置MQ管理人的用户名要在Web版先配置保证该用户可以管理设置的虚拟主机connectionFactory.setPassword(guest); //设置MQ管理人的密码}//定义提供连接对象的方法封装public static Connection getConnection(){try {//创建连接对象并返回return connectionFactory.newConnection();} catch (Exception e) {e.printStackTrace();}return null;}//关闭通道和关闭连接工具类的方法public static void closeConnectionAndChannel(Channel channel, Connection connection){try {if (channel!null) {channel.close();}if (connection!null){connection.close();}} catch (Exception e) {e.printStackTrace();}}
}生产者
package com.dmbjz.one;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.nio.charset.StandardCharsets;/* 死信队列TTL案例 生产者 */
public class Provider {private static final String EXCHANGE_NAME normal_exchange; //正常交换机名称private static final String KEY zhangsan; //普通队列 RoutingKeypublic static void main(String[] args) throws Exception {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT); //声明交换机/*死信队列 设置TTL消息过期时间 单位毫秒*/AMQP.BasicProperties properties new AMQP.BasicProperties().builder().expiration(20000).build();/*模拟消息循环发送*/for(int i 1; i 11; i) {String message INFO i;channel.basicPublish(EXCHANGE_NAME,KEY,properties,message.getBytes(StandardCharsets.UTF_8)); }}
}消费者C1
package com.dmbjz.one;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;/* 死信队列TTL案例 消费者C1 */
public class ConsumerC1 {private static final String EXCHANGE_NAME normal_exchange; //正常交换机名称private static final String DEAD_EXCHANGE_NAME dead_exchange; //死信队列交换机名称private static final String KEY zhangsan; //普通队列 RoutingKeyprivate static final String DEAD_KEY lisi; //死信队列 RoutingKeyprivate static final String QUEUE_NAME normal-queue; //普通队列名称private static final String DEAD_QUEUE_NAME dead-queue; //死信队列名称public static void main(String[] args) throws IOException {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();/*声明死信和普通交换机正常交换机已被生产者声明实际可以省略第一行代码*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);/*创建队列* 通过额外参数实现什么情况下转发到死信队列 ,key都是固定的* 1、TTL过期时间设置(一般由生产者指定)* 2、死信交换机的名称* 3、死信交换机的RoutingKey* */MapString,Object arguments new HashMap(8);arguments.put(x-dead-letter-exchange,DEAD_EXCHANGE_NAME); //死信交换机的名称arguments.put(x-dead-letter-routing-key,DEAD_KEY); //死信交换机的RoutingKey// arguments.put(x-dead-letter-ttl,10000); 指定消息的有效时间为20秒一般为生产者指定channel.queueDeclare(QUEUE_NAME,false,false,false,arguments);channel.queueDeclare(DEAD_QUEUE_NAME,false,false,false,null);/*绑定队列*/channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,KEY);channel.queueBind(DEAD_QUEUE_NAME,DEAD_EXCHANGE_NAME,DEAD_KEY);DeliverCallback successBack (consumerTag, message) - {System.out.println(C1用户接收到的信息为:new String(message.getBody()));};CancelCallback cnaelBack a-{System.out.println(C1用户进行取消消费操作!);};channel.basicConsume(QUEUE_NAME,true,successBack,cnaelBack);}
}消费者C2
package com.dmbjz.one;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.*;import java.io.IOException;/* 死信队列TTL案例 消费者C2 */
public class ConsumerC2 {private static final String DEAD_EXCHANGE_NAME dead_exchange; //死信队列交换机名称private static final String DEAD_KEY lisi; //死信队列 RoutingKeyprivate static final String DEAD_QUEUE_NAME dead-queue; //死信队列名称public static void main(String[] args) throws IOException {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();/*声明队列和普通交换机并进行绑定,由于消费者C1已经声明过了这里实际可以省略这三行代码*/channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);channel.queueDeclare(DEAD_QUEUE_NAME,false,false,false,null);channel.queueBind(DEAD_QUEUE_NAME,DEAD_EXCHANGE_NAME,DEAD_KEY);DeliverCallback successBack (consumerTag, message) - {System.out.println(C2用户接收到的信息为:new String(message.getBody()));};CancelCallback cnaelBack a-{System.out.println(C2用户进行取消消费操作!);};channel.basicConsume(DEAD_QUEUE_NAME,true,successBack,cnaelBack);}}效果演示
执行消费者C1创建出所有交换机和队列绑定后停止运行。执行生产者等待10秒钟后查看控制台消息全部通过交换机进入死信队列运行消费者C2死信队列消息被成功消费 队列消息积压达到最大长度
在绑定死信队列的消费者端添加队列的消息积压长度限制即可核心代码为TTL案例的消费者C1基础上再添加Map参数
//指定队列能够积压消息的大小超出该范围的消息将进入死信队列
arguments.put(x-max-length,6); 生产者
package com.dmbjz.maxlength;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.nio.charset.StandardCharsets;/* 死信队列 队列达到最大长度案例 生产者 */
public class Provider {private static final String EXCHANGE_NAME normal_exchange; //正常交换机名称private static final String KEY zhangsan; //普通队列 RoutingKeypublic static void main(String[] args) throws Exception {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT); //声明交换机/*循环消息发送*/for(int i 1; i 11; i) {String message INFO i;channel.basicPublish(EXCHANGE_NAME,KEY,null,message.getBytes(StandardCharsets.UTF_8)); //发送超级VIP消息}}}消费者C1消费者C2的代码与TTL案例中保持一致
package com.dmbjz.maxlength;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;/* 死信队列 队列达到最大长度案例 消费者C1 */
public class ConsumerC1 {private static final String EXCHANGE_NAME normal_exchange; //正常交换机名称private static final String DEAD_EXCHANGE_NAME dead_exchange; //死信队列交换机名称private static final String KEY zhangsan; //普通队列 RoutingKeyprivate static final String DEAD_KEY lisi; //死信队列 RoutingKeyprivate static final String QUEUE_NAME normal-queue; //普通队列名称private static final String DEAD_QUEUE_NAME dead-queue; //死信队列名称public static void main(String[] args) throws IOException {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();/*声明死信和普通交换机正常交换机已被生产者声明实际可以省略第一行代码*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);/*创建队列* 通过额外参数实现什么情况下转发到死信队列 ,key都是固定的* 1、TTL过期时间设置(一般由生产者指定)* 2、死信交换机的名称* 3、死信交换机的RoutingKey* */MapString,Object arguments new HashMap(8);arguments.put(x-dead-letter-exchange,DEAD_EXCHANGE_NAME); //死信交换机的名称arguments.put(x-dead-letter-routing-key,DEAD_KEY); //死信交换机的RoutingKeyarguments.put(x-max-length,6); //指定正常队列的长度超出该范围的消息将进入死信队列channel.queueDeclare(QUEUE_NAME,false,false,false,arguments);channel.queueDeclare(DEAD_QUEUE_NAME,false,false,false,null);/*绑定队列*/channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,KEY);channel.queueBind(DEAD_QUEUE_NAME,DEAD_EXCHANGE_NAME,DEAD_KEY);DeliverCallback successBack (consumerTag, message) - {System.out.println(C1用户接收到的信息为:new String(message.getBody()));};CancelCallback cnaelBack a-{System.out.println(C1用户进行取消消费操作!);};channel.basicConsume(QUEUE_NAME,true,successBack,cnaelBack);}
}效果演示
执行消费者C1创建出所有交换机和队列绑定后停止运行。执行生产者由于通道积压的六条消息从未被消费。剩余消息进入死信队列运行消费者C2死信队列消息被成功消费 消息拒绝应答
在绑定死信队列的消费者端添加需要拒绝应答的消息判断即可核心代码为消息拒绝应答
/* requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中,未配置则进行丢弃操作*/
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);生产者
package com.dmbjz.noack;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;import java.nio.charset.StandardCharsets;/* 死信队列 队列达到最大长度案例 生产者 */
public class Provider {private static final String EXCHANGE_NAME normal_exchange; //正常交换机名称private static final String KEY zhangsan; //普通队列 RoutingKeypublic static void main(String[] args) throws Exception {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.DIRECT); //声明交换机/*循环消息发送*/for(int i 1; i 11; i) {String message INFO i;channel.basicPublish(EXCHANGE_NAME,KEY,null,message.getBytes(StandardCharsets.UTF_8)); //发送超级VIP消息}}
}消费者C1消费者C2的代码和TTL案例保持一致
package com.dmbjz.noack;import com.dmbjz.utils.RabbitUtils;
import com.rabbitmq.client.*;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;/* 死信队列 队列达到最大长度案例 消费者C1 */
public class ConsumerC1 {private static final String EXCHANGE_NAME normal_exchange; //正常交换机名称private static final String DEAD_EXCHANGE_NAME dead_exchange; //死信队列交换机名称private static final String KEY zhangsan; //普通队列 RoutingKeyprivate static final String DEAD_KEY lisi; //死信队列 RoutingKeyprivate static final String QUEUE_NAME normal-queue; //普通队列名称private static final String DEAD_QUEUE_NAME dead-queue; //死信队列名称public static void main(String[] args) throws IOException {Connection connection RabbitUtils.getConnection();Channel channel connection.createChannel();/*声明死信和普通交换机正常交换机已被生产者声明实际可以省略第一行代码*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);channel.exchangeDeclare(DEAD_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);/*创建队列* 通过额外参数实现什么情况下转发到死信队列 ,key都是固定的* 1、TTL过期时间设置(一般由生产者指定)* 2、死信交换机的名称* 3、死信交换机的RoutingKey* */MapString,Object arguments new HashMap(8);arguments.put(x-dead-letter-exchange,DEAD_EXCHANGE_NAME); //死信交换机的名称arguments.put(x-dead-letter-routing-key,DEAD_KEY); //死信交换机的RoutingKeychannel.queueDeclare(QUEUE_NAME,false,false,false,arguments);channel.queueDeclare(DEAD_QUEUE_NAME,false,false,false,null);/*绑定队列*/channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,KEY);channel.queueBind(DEAD_QUEUE_NAME,DEAD_EXCHANGE_NAME,DEAD_KEY);DeliverCallback successBack (consumerTag, message) - {String info new String(message.getBody(),UTF-8);if(info.equals(INFO 5)){System.out.println(C1用户拒绝的信息为:new String(message.getBody()));/* requeue 设置为 false 代表拒绝重新入队 该队列如果配置了死信交换机将发送到死信队列中,未配置则进行丢弃操作*/channel.basicReject(message.getEnvelope().getDeliveryTag(),false);}else{System.out.println(C1用户接收到的信息为:new String(message.getBody()));channel.basicAck(message.getEnvelope().getDeliveryTag(),false);}};CancelCallback cnaelBack a-{System.out.println(C1用户进行取消消费操作!);};channel.basicConsume(QUEUE_NAME,false,successBack,cnaelBack);}
}效果演示
执行消费者C1执行生产者等待10秒钟后查看控制台消息5被拒绝接收进入死信队列运行消费者C2死信队列消息被成功消费
