惠普网站建设的目标,网页搜索快捷键是什么,做网站内容,梅州做网站文章目录 1. 消息丢失问题1.1 发送者消息丢失1.2 MQ消息丢失1.3 消费者消息丢失1.3.1 消费失败重试机制 总结 2. 死信交换机2.1 TTL 3. 惰性队列3.1 总结#xff1a; 4. MQ集群 消息队列在使用过程中#xff0c;面临着很多实际问题需要思考#xff1a; 1. 消息丢失问题
1.1… 文章目录 1. 消息丢失问题1.1 发送者消息丢失1.2 MQ消息丢失1.3 消费者消息丢失1.3.1 消费失败重试机制 总结 2. 死信交换机2.1 TTL 3. 惰性队列3.1 总结 4. MQ集群 消息队列在使用过程中面临着很多实际问题需要思考 1. 消息丢失问题
1.1 发送者消息丢失
解决方式生产者确认机制 RabbitMQ提供了publisher confirm机制来避免消息发送到MQ过程中丢失。这种机制必须给每个消息指定一个唯一ID。消息发送到MQ以后会返回一个结果给发送者表示消息是否处理成功。
返回结果有两种方式
publisher-confirm发送者确认 消息成功投递到交换机返回ack消息未投递到交换机返回nack publisher-return发送者回执 消息投递到交换机了但是没有路由到队列。返回ACK及路由失败原因。 案例
修改配置 首先修改publisher服务中的application.yml文件添加下面的内容 说明
publish-confirm-type开启publisher-confirm这里支持两种类型 simple同步等待confirm结果直到超时correlated异步回调定义ConfirmCallbackMQ返回结果时会回调这个ConfirmCallback publish-returns开启publish-return功能同样是基于callback机制不过是定义ReturnCallbacktemplate.mandatory定义消息路由失败时的策略。true则调用ReturnCallbackfalse则直接丢弃消息
定义Return回调 每个RabbitTemplate只能配置一个ReturnCallback因此需要在项目加载时配置 修改publisher服务添加一个 定义ConfirmCallback ConfirmCallback可以在发送消息时指定因为每个业务处理confirm成功或失败的逻辑不一定相同。
1.2 MQ消息丢失
解决方式 交换机持久化、队列持久化、消息持久化 交换机持久化RabbitMQ中交换机默认是非持久化的mq重启后就丢失。SpringAMQP中可以通过代码指定交换机持久化 事实上默认情况下由SpringAMQP声明的交换机都是持久化的。
队列持久化RabbitMQ中队列默认是非持久化的mq重启后就丢失。SpringAMQP中可以通过代码指定交换机持久化 事实上默认情况下由SpringAMQP声明的队列都是持久化的。 消息持久化利用SpringAMQP发送消息时可以设置消息的属性MessageProperties指定delivery-mode 默认情况下SpringAMQP发出的任何消息都是持久化的不用特意指定。
1.3 消费者消息丢失
解决方式消费者消息确认机制消费者在成功处理了消息后告诉MQ成功处理
而SpringAMQP则允许配置三种确认模式 manual手动ack需要在业务代码结束后调用api发送ack。 auto自动ack由spring监测listener代码是否出现异常aop没有异常则返回ack抛出异常则返回nack none关闭ackMQ假定消费者获取消息后会成功处理因此消息投递后立即被删除
由此可知
none模式下消息投递是不可靠的可能丢失auto模式类似事务机制出现异常时返回nack消息回滚到mq没有异常返回ackmanual自己根据业务情况判断什么时候该ack
一般我们都是使用默认的auto即可。 案例
1.3.1 消费失败重试机制
当消费者出现异常后消息会不断requeue重入队到队列再重新发送给消费者然后再次异常再次requeue无限循环导致mq的消息处理飙升带来不必要的压力 解决办法: 本地重试我们可以利用Spring的retry机制在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。 修改consumer服务的application.yml文件添加内容 结论
开启本地重试时消息处理过程中抛出异常不会requeue到队列而是在消费者本地重试重试达到最大次数后Spring会返回ack消息会被丢弃
在开启重试模式后重试次数耗尽如果消息依然失败则需要有MessageRecovery接口来处理它包含三种不同的实现
RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认就是这种方式ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机
比较优雅的一种处理方案是RepublishMessageRecoverer失败后将消息投递到一个指定的专门存放异常消息的队列后续由人工集中处理。 1在consumer服务中定义处理失败消息的交换机和队列 2定义一个RepublishMessageRecoverer关联队列和交换机自己定义的bean会覆盖原先默认的
总结
如何确保RabbitMQ消息的可靠性
开启生产者确认机制确保生产者的消息能到达队列开启持久化功能确保消息未消费前在队列中不会丢失开启消费者确认机制为auto由spring确认消息处理成功后完成ack开启消费者失败重试机制并设置MessageRecoverer多次重试失败后将消息投递到异常交换机交由人工处理
2. 死信交换机
当一个队列中的消息满足下列情况之一时可以成为死信dead letter
消费者使用basic.reject或 basic.nack声明消费失败并且消息的requeue参数设置为false消息是一个过期消息超时无人消费要投递的队列消息满了无法投递
如果这个包含死信的队列配置了dead-letter-exchange属性指定了一个交换机那么队列中的死信就会投递到这个交换机中而这个交换机称为死信交换机Dead Letter Exchange检查DLX。 在失败重试策略中默认的RejectAndDontRequeueRecoverer会在本地重试次数耗尽后发送reject给RabbitMQ消息变成死信被丢弃。 我们可以给simple.queue添加一个死信交换机给死信交换机绑定一个队列。这样消息变成死信后也不会丢弃而是最终投递到死信交换机路由到与死信交换机绑定的队列。
我们在consumer服务中定义一组死信交换机、死信队列 死信交换机的使用场景是什么
如果队列绑定了死信交换机死信会投递到死信交换机可以利用死信交换机收集所有消费者处理失败的消息死信交由人工处理进一步提高消息队列的可靠性。
2.1 TTL
一个队列中的消息如果超时未消费则会变为死信超时分为两种情况
消息所在的队列设置了超时时间消息本身设置了超时时间 只需要在发消息中设置消息的超时时间ttl.queue设置队列的超时时间。 总结 消息超时的两种方式是
给队列设置ttl属性进入队列后超过ttl时间的消息变为死信给消息设置ttl属性队列接收到消息超过ttl时间后变为死信
如何实现发送一个消息20秒后消费者才收到消息
给消息的目标队列指定死信交换机将消费者监听的队列绑定到死信交换机发送消息时给消息设置超时时间为20秒
延迟队列 利用TTL结合死信交换机我们实现了消息发出后消费者延迟收到消息的效果。这种消息模式就称为延迟队列Delay Queue模式。
延迟队列的使用场景包括
延迟发送短信用户下单如果用户在15 分钟内未支付则自动取消预约工作会议20分钟后自动通知所有参会人员
因为延迟队列的需求非常多所以RabbitMQ的官方也推出了一个插件原生支持延迟队列效果。 这个插件就是DelayExchange插件。
DelayExchange需要将一个交换机声明为delayed类型。当我们发送消息到delayExchange时流程如下
接收消息判断消息是否具备x-delay属性如果有x-delay属性说明是延迟消息持久化到硬盘读取x-delay值作为延迟时间返回routing not found结果给消息发送者x-delay时间到期后重新投递消息到指定队列
使用DelayExchange 插件的使用也非常简单声明一个交换机交换机的类型可以是任意类型只需要设定delayed属性为true即可然后声明队列与其绑定即可。
3. 惰性队列
消息堆积问题当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。之后发送的消息就会成为死信可能会被丢弃这就是消息堆积问题。
解决消息堆积有三种思路
增加更多消费者提高消费速度。在消费者内开启线程池加快出消息处理速度扩大队列容积提高堆积上限
惰性队列扩大队列容积
接收到消息后直接存入磁盘而非内存消费者要消费消息时才会从磁盘中读取并加载到内存支持数百万条的消息存储
基于RabbitListener声明LazyQueue
3.1 总结
消息堆积问题的解决方案
队列上绑定多个消费者提高消费速度使用惰性队列可以在mq中保存更多消息
惰性队列的优点有哪些
基于磁盘存储消息上限高没有间歇性的page-out性能比较稳定
惰性队列的缺点有哪些
基于磁盘存储消息时效性会降低性能受限于磁盘的IO
4. MQ集群 普通集群是一种分布式集群将队列分散到集群的各个节点从而提高整个集群的并发能力。 镜像集群是一种主从集群普通集群的基础上添加了主从备份功能提高集群的数据可用性。 仲裁队列镜像集群虽然支持主从但主从同步并不是强一致的某些情况下可能有数据丢失的风险。因此在RabbitMQ的3.8版本以后推出了新的功能仲裁队列来代替镜像集群底层采用Raft协议确保主从的数据一致性。
Java代码创建仲裁队列
SpringAMQP连接MQ集群:
