关于图书网站建设的书籍,镇江详情设计,宁波seo公司网站推广,网站名词解释目录
一、介绍
1、什么是死信队列(延迟队列)
2、应用场景
3、死信队列(延迟队列)的使用
4、死信消息来源
二、案例实践
1、案例一
2、案例二#xff08;消息接收确认 #xff09;
3、总结 一、介绍
1、什么是死信队列(延迟队列) 死信#xff0c;在官网中对应的单词…目录
一、介绍
1、什么是死信队列(延迟队列)
2、应用场景
3、死信队列(延迟队列)的使用
4、死信消息来源
二、案例实践
1、案例一
2、案例二消息接收确认
3、总结 一、介绍
1、什么是死信队列(延迟队列) 死信在官网中对应的单词为“Dead Letter”,它是 RabbitMQ 的一种消息机制。 死信队列Dead Letter Queue和延迟队列Delay Queue是两种不同的队列类型但在实际应用中它们可以结合使用。 死信队列是当消息在队列中因为过期、被拒绝等原因无法正常处理时会被重新发送到另一个交换机上这个交换机就是死信交换机。死信队列可以用于实现重试机制、日志审计等特殊应用逻辑。 延迟队列则是一种特殊的队列类型它允许将消息延迟指定的时间后才能被消费者消费。这种队列通常用于处理那些需要在特定时间点被处理的任务例如定时任务、限时优惠等。在RabbitMQ中可以通过设置消息的TTL生存时间来实现延迟队列的功能。当消息在队列中超过了TTL它就会被移除并被发送到指定的死信交换机进而被路由到死信队列中。 结合使用死信队列和延迟队列可以实现一些复杂的应用逻辑。 例如: 可以将某个需要延迟处理的消息发送到延迟队列中并在消息过期之前将其存储在死信队列中。这样当消息从延迟队列中移除时它会被自动发送到死信队列中然后由消费者消费并执行相应的操作。这种结合使用的方式可以提供更高的灵活性和可靠性使得系统能够更好地应对各种异常情况。 2、应用场景
死信队列Dead Letter Queue和延迟队列Delay Queue在以下应用场景中表现优异 这些场景都是通过结合使用死信队列和延迟队列来提高系统的可靠性、鲁棒性和灵活性。通过合理地设置死信队列和延迟队列的参数可以实现各种复杂的业务逻辑和异常处理机制。
库存解锁服务例如当一个商品被锁定且无法被购买时可以将其放入死信队列中并在一定时间后重新发送到原始队列进行处理从而解锁库存。定时关单功能例如用户在商城下单成功并点击去支付后如果在指定时间内未支付系统可以自动将订单放入死信队列中并在一定时间后重新发送到原始队列进行处理。保证订单业务的消息数据不丢失当消息消费发生异常时可以将消息投入死信队列中以便后续等到环境好了之后再消费死信队列中的消息。实现重试机制当某个消息处理失败时可以将它放入死信队列中并在一段时间后重新发送到交换机进行重试。这样可以提高系统的鲁棒性确保消息能够被正确处理。处理定时任务和秒杀活动使用延迟队列可以将消息延迟到特定的时间后进行处理例如定时任务、秒杀活动等。这样可以确保在特定的时间点执行相应的操作。日志处理和审计将日志消息发送到死信队列中可以在日志发生异常时进行记录和审计以便分析和排查问题。 3、死信队列(延迟队列)的使用
使用死信队列Dead Letter Queue和延迟队列Delay Queue可以提高系统的可靠性和灵活性特别是在处理异常情况、重试机制和定时任务等方面。
需要注意的是在使用死信队列和延迟队列时需要考虑系统的可用性和性能。如果大量的消息被放入死信队列中可能会导致系统资源的过度消耗。因此需要合理配置死信队列和延迟队列的大小和数量以及监控和管理系统的性能和资源使用情况。
定义死信队列和延迟队列在RabbitMQ中可以在定义队列时指定队列类型为死信队列或延迟队列。例如使用RabbitMQ的命令行工具或管理界面创建死信队列或延迟队列。配置交换机和队列属性在绑定交换机和队列时可以设置一些属性来控制消息的路由和消费。例如可以设置消息的TTL生存时间来实现延迟队列的功能或者设置消息的优先级和延迟时间等属性。发送消息到队列使用RabbitMQ的生产者将消息发送到定义的队列中。如果需要将消息放入延迟队列中可以在发送消息时设置相应的延迟时间。处理消息消费者从队列中获取消息并进行处理。如果消息无法正常处理可以将它放入死信队列中。在处理消息时可以根据需要设置重试机制以便在消息处理失败时重新发送消息到队列中进行重试。监控和管理使用RabbitMQ的管理界面或命令行工具监控和管理死信队列和延迟队列的状态和消息。可以查看队列的大小、消息的延迟时间、消费者数量等信息并根据需要进行调整和管理。
4、死信消息来源 消息 TTL 过期队列满了无法再次添加数据消息被拒绝reject 或 nack并且 requeue false
二、案例实践 1、案例一
生产者 在生产者的Config里面添加队列 。 // 死信队列// 定义QueueA Bean返回QueueA实例Beanpublic Queue QueueA() {MapString, Object config new HashMap();//message在该队列queue的存活时间最大为5秒config.put(x-message-ttl, 5000);//x-dead-letter-exchange参数是设置该队列的死信交换器DLXconfig.put(x-dead-letter-exchange, deadExchange);//x-dead-letter-routing-key参数是给这个DLX指定路由键config.put(x-dead-letter-routing-key, deadQueue);// 返回QueueA实例return new Queue(QueueA, true, true, false, config);}// 定义DirectExchangeA Bean返回DirectExchangeA实例Beanpublic DirectExchange DirectExchangeA() {// 返回DirectExchangeA实例return new DirectExchange(DirectExchangeA);}// 定义bindingA Bean将QueueA绑定到DirectExchangeA并设置路由键为A.ABeanpublic Binding bindingA() {// 将QueueA绑定到DirectExchangeAreturn BindingBuilder// 设置路由键为A.A.bind(QueueA()).to(DirectExchangeA()).with(A.A);}// 定义QueueB Bean返回QueueB实例Beanpublic Queue QueueB() {// 返回QueueB实例return new Queue(QueueB);}// 定义DirectExchangeB Bean返回DirectExchangeB实例Beanpublic DirectExchange DirectExchangeB() {// 返回DirectExchangeB实例return new DirectExchange(DirectExchangeB);}// 定义bindingB Bean将QueueB绑定到DirectExchangeB并设置路由键为B.BBeanpublic Binding bindingB() {// 将QueueB绑定到DirectExchangeBreturn BindingBuilder// 设置路由键为B.B.bind(QueueB()).to(DirectExchangeB()).with(B.B);}
消费者 在消费者里面添加QueueA和QueueB QueueA
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
SuppressWarnings(all)
Slf4j
RabbitListener(queues QueueA)
public class ReceiverQA {// 接收directExchange01交换机中Queue02队列消息的方法RabbitHandlerpublic void Queue02(String msg) {log.warn(QueueA接收到信息: msg);}}
QueueB
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
SuppressWarnings(all)
Slf4j
RabbitListener(queues QueueB)
public class ReceiverQB {// 接收directExchange01交换机中Queue02队列消息的方法RabbitHandlerpublic void Queue02(String msg) {log.warn(QueueB接收到信息: msg);}} 编写Controller层 RequestMapping(send5)public String send5() {rabbitTemplate.convertAndSend(DirectExchangeA, A.A, Hello,A);return ;} 结果 2、案例二消息接收确认 如果某个服务忘记 ACK 了则 RabbitMQ 不会再发送数据给它因为 RabbitMQ 认为该服务的处理能力有限 ACK 机制还可以起到限流作用比如在接收到某条消息时休眠几秒钟 消息确认模式有 AcknowledgeMode.NONE自动确认 AcknowledgeMode.AUTO根据情况确认 AcknowledgeMode.MANUAL手动确认 配置文件 默认情况下消息消费者是自动 ack 确认消息的如果要手动 ack确认则需要修改确认模式为 manual spring:rabbitmq:listener:simple:acknowledge-mode: manual 在消费者里的更改ReceiverQA import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import java.io.IOException;Component
SuppressWarnings(all)
Slf4j
RabbitListener(queues QueueA)
public class ReceiverQA {// 接收directExchange01交换机中Queue02队列消息的方法RabbitHandlerpublic void QueueA(String msg, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {log.warn(QueueA接收到信息: msg);channel.basicAck(tag, false);}} 需要注意的 basicAck 方法需要传递两个参数 deliveryTag唯一标识 ID当一个消费者向 RabbitMQ 注册后会建立起一个 Channel RabbitMQ 会用 basic.deliver 方法向消费者推送消息这个方法携带了一个 delivery tag 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID是一个单调递增的正整数delivery tag 的范围仅限于 Channel multiple为了减少网络流量手动确认可以被批处理当该参数为 true 时则可以一次性确认 delivery_tag 小于等于传入值的所有消息 3、总结 持久化 exchange要持久化 queue要持久化 message要持久化 消息确认 启动消费返回ReturnList注解生产者就可以知道哪些消息没有发出去 生产者和Serverbroker之间的消息确认 消费者和Serverbroker之间的消息确认
