有用模板网官网,seo快速排名软件,云南省城乡住房建设厅官方网站,毕业设计微信小程序开发目录 一、简介1.1 定义1.2 消息丢失的场景 二、交换机的持久化方式一#xff1a;直接 new方式二#xff1a;channel.exchangeDeclare()方式三#xff1a;ExchangeBuilder【推荐】 三、队列的持久化方式一#xff1a;直接 new方式二#xff1a;channel.queueDeclare()方式三… 目录 一、简介1.1 定义1.2 消息丢失的场景 二、交换机的持久化方式一直接 new方式二channel.exchangeDeclare()方式三ExchangeBuilder【推荐】 三、队列的持久化方式一直接 new方式二channel.queueDeclare()方式三QueueBuilder【推荐】 四、消息的持久化方式一channel.basicPublish()方式二rabbitTemplate.convertAndSend()【推荐】 五、持久化问题 一、简介
1.1 定义
持久化 是为了提高 RabbitMQ 消息的可靠性防止在异常情况重启、宕机下数据的丢失。RabbitMQ 持久化分为三部分交换机的持久化、队列的持久化、消息的持久化。
1.2 消息丢失的场景
出现消息丢失的场景有
生产者发送消息丢失发送消息到 RabbitMQ Server 异常。 可能因为网络问题造成 RabbitMQ 服务端无法收到消息。——解决方案ConfirmCallback生产者发送消息丢失消息无法路由到指定队列。 可能由于代码层面或配置层面错误导致消息路由到指定队列失败。——解决方案ReturnCallbackRabbitMQ Server 存储消息丢失消息未完全持久化到磁盘。 可能因为 RabbitMQ 宕机导致消息未完全持久化或队列丢失从而导致消息丢失等持久化问题。——解决方案集群部署实现高可用消费者消费消息丢失消费者消费消息异常。 可能在消费者接收消息后还没来得及消费消息消费者就发生宕机、故障等问题。——解决方案消费端手动确认消息 二、交换机的持久化
方式一直接 new
直接实例化对应的 Exchange 实现类即可默认持久化的、非自动删除的。
Bean
public DirectExchange directExchange() {return new DirectExchange(directExchange);
}Exchange 源码
package org.springframework.amqp.core;public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {...public AbstractExchange(String name) {// 默认是持久化的、非自动删除的。this(name, true, false);}...
}方式二channel.exchangeDeclare()
在 RabbitMQ 的原生 API 中例如Java 客户端 API声明持久化交换机时需要将 durable 参数设置为 true
// 声明交换机(交换机名称交换机类型持久化)
channel.exchangeDeclare(myExchange, direct, true);方式三ExchangeBuilder【推荐】
在 Spring AMQP 中我们可以使用 ExchangeBuilder 来创建持久化的交换机
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;Exchange exchange ExchangeBuilder.directExchange(myExchange).durable(true) // 交换机持久化.build();三、队列的持久化
方式一直接 new
直接实例化 Queue 类即可默认持久化的、非独占的、非自动删除的。
Bean
public Queue myQueue() {return new Queue(myQueue);
}Queue 源码
package org.springframework.amqp.core;public Queue(String name) {// 默认是持久化的、非独占的、非自动删除的。this(name, true, false, false);
}方式二channel.queueDeclare()
在原生 RabbitMQ API 中例如 Java 客户端 API声明持久化队列时需要将 durable 参数设置为 true
// 声明队列(队列名称持久化非独占非自动删除可选队列参数)
channel.queueDeclare(myQueue, true, false, false, null);方式三QueueBuilder【推荐】
在 Spring AMQP 中可以使用 QueueBuilder 来创建持久化的队列
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;Queue queue QueueBuilder.durable(myQueue).build();四、消息的持久化
即使队列时持久化的发送到队列中的消息默认情况下 可能仍然是非持久化的。要使消息持久化需要在发送消息时设置消息属性为持久化。
方式一channel.basicPublish()
在原生 RabbitMQ API 中例如 Java 客户端 API声明持久化消息时需要将 deliveryMode 参数设置为 2
import com.rabbitmq.client.AMQP;AMQP.BasicProperties props new AMQP.BasicProperties.Builder().deliveryMode(2) // 这里2对应MessageDeliveryMode.PERSISTENT.build();// 参数(交换机路由键消息的其他参数消息体)
channel.basicPublish(myExchange, myRoutingKey, props. messageBytes);或者我们可以直接使用 MessageProperties.PERSISTENT_TEXT_PLAIN 来进行指定消息的持久化
import com.rabbitmq.client.MessageProperties;// 参数(交换机路由键消息的其他参数消息体)
channel.basicPublish(myExchange, myRoutingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());方式二rabbitTemplate.convertAndSend()【推荐】
在 Spring AMQP 中可以使用 rabbitTemplate.convertAndSend() 来创建持久化的消息
rabbitTemplate.convertAndSend(myQueue, body, message - {message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);return message;
});通过上述步骤消息会被持久化存储只要队列也被正确配置为持久化即使 RabbitMQ 服务器重启消息也将被保留下来。 补充 然而即使队列和消息都是持久化的也不能完全保证消息的 100% 不丢失。例如在消息尚未被刷写到磁盘时RabbitMQ 服务器突然崩溃这种情况下的消息仍然可能会丢失。此外RabbitMQ 不保证消息的立即持久化而是尽可能快地将消息保存到磁盘。 五、持久化问题 思考将交换机、队列、消息都设置持久化之后就能保证数据不会丢失吗 答当然不能需要从多方面考虑 场景1 如果消费者订阅队列的时候将 autoAck自动确认设置为 true虽然消费者接收到了消息但是没有来得及处理就宕机了那消息也会丢失。 解决方案 将消费者的消息确认机制改为手动确认带处理完消息之后手动删除消息。 场景2 在 RabbitMQ 服务器如果消息正确被发送但是 RabbitMQ 服务器中的消息还没来得及持久化即没有将数据写入磁盘时如果服务器发生异常消息也会丢失。 解决方案 可以 通过 RabbitMQ 集群的方式实现消息中间件的高可用。
因此还需要做好生产者和消费者的 消息确认机制 以及通过 集群 的方式来实现 RabbitMQ 的高可用。
整理完毕完结撒花~ 参考地址
1.RabbitMQ保证消息可靠性https://www.cnblogs.com/auguse/articles/17712620.html
