做淘客网站怎么,手机海外代理ip,河南省建设银行网站年报,找到网站后台地址#xff08;一#xff09;消息确认
1.概念
生产者发送消息后#xff0c;到达消费端会有以下情况#xff1a;
1.消息处理成功
2.消息处理异常 如果RabbitMQ把消息发送给消费者后就把消息删除#xff0c;那么就可能会导致#xff0c;消息处理异常想要再获取这条消息的时…一消息确认
1.概念
生产者发送消息后到达消费端会有以下情况
1.消息处理成功
2.消息处理异常 如果RabbitMQ把消息发送给消费者后就把消息删除那么就可能会导致消息处理异常想要再获取这条消息的时候造成消息丢失但如果消息处理成功不需要这条消息了就可以进行删除。此时就不会有问题 所以如何保证消费者成功接收并正确处理我们RabbitMQ就给我们提供了消息确认机制 消费者在订阅队列的时候我们可以指定autoAck参数根据这个参数我们可以把确认机制大致分为两种Spring boot有三种但是本质也是这两种 自动确认当autoAck为true时RabbitMQ只要发送了这个消息就会从内存中删除不会管消费者是否收到了这些消息所以自动确认的可靠性不高 手动确认当autoAck为false时RabbitMQ会等待消费者调用Basic.Ack回复确认信号后RabbitMQ才会从队列中删除消息所以手动确认的可靠性是比较高的
当我们设置为手动确认后队列中的消息就分为了两个部分
1.等待投递给消费者的消息
2.已经投递给消费者但是没有收到确认信号的消息 也就是Ready和Unacked
如果RabbitMQ一致没有收到消费者的确认信号并且消息对应的消费者断开连接RabbitMQ就会安排消息重新入队列等待投递给下一个消费者 2.Spring Boot的三种策略和代码演示
Springboot给我们提供了三种策略本质上还是自动确认和手动确认
1AcknowledgeMode.NONE 这种模式下就是标准的自动确认消息一旦投递给消费者不管消费者是否正确处理了消息RabbitMQ就会自动确认消息并且从队列中移除所以消息是很有可能丢失的 2AcknowledgeMode.AUTO默认
这种模式下消息成功处理时会自动确认消息如果消息过程中抛出了异常就不会确认消息
3AckniwledgeMode.MANUAL
这种模式下就是标准的手动确认模式消费者必须在成功处理消息后调用basicAck方法来确认消息如果消息没有被确认RabbitMQ就会认为消息没有被成功处理会重新投递该消息这种模式会提高消息的可靠性因为消息不会丢失而是重新入队
代码演示
配置文件代码
spring:rabbitmq:addresses: amqp://student:student62.234.46.219:5672/testlistener:simple:acknowledge-mode: NONE
声明交换机和队列代码
Component
public class Config {Bean(ackExchange)public Exchange ackExchange(){return ExchangeBuilder.directExchange(Constants.ACK_EXCHANGE).durable(true).build();}Bean(ackQueue)public Queue ackQueue(){return QueueBuilder.durable(Constants.ACK_QUEUE).build();}Beanpublic Binding ackBind(Qualifier(ackExchange) Exchange ackExchange,Qualifier(ackQueue) Queue queue){return BindingBuilder.bind(queue).to(ackExchange).with(ack).noargs();}
}
生产者代码 RequestMapping(producer)
RestController
public class AckProducer {Autowiredprivate RabbitTemplate rabbitTemplate;RequestMapping(ack)public String ackPro(){rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,ack,ack test);return 发送成功;}
}然后我们发送一条消息看现象此时我们还没写消费者代码所以没有自动确认 消费者代码
Component
public class AckConsumer {RabbitListener(queues Constants.ACK_QUEUE)public void ListenerQueue(Message message, Channel channel){System.out.println(接收到消息: new String(message.getBody()) TagID: message.getMessageProperties().getDeliveryTag());
// int num3/0; //模拟失败System.out.println(处理完成);}
}此时我们没有出错我们看能否正确接收消息已经队列中消息是否存在 我们发现能够准确的接收消息
如果我们此时抛出异常呢 我们发现抛出了异常后我们后端并没有接收到消息但是队列的消息也消失了所以就会导致我们的消息不见了这也就是自动确认的弊端
此时我们把确认策略改成AUTO
我们这里只需要改配置文件即可
截图体现不出来但是结果是一直在循环因为我们如果设置确认策略为AUTO在运行正常时会自动确认并且从队列中删除如果抛出异常就不会确认消息造成循环 我们再来演示下正确情况 此时我们再来说一下手动确认策略
首先更改配置文件 其次和我们刚刚演示AUTO和NONE时代码是没有改动的但是手动确认因为要我们手动做一些处理所以代码也要有一定的改变 那我们先来讲一下手动确认方法
手动确认方法
1.肯定确认 RabbitMQ已经知道该消息并且成功的处理消息可以丢弃了
我们来解释下这个方法的参数
deliveryTag消息的唯一标识是一个单调递增的64位长整型每个channel之间是独立的所以在每个channel上是唯一的当消费者确认一条消息时需要用对应的信道上进行确认
multiple是否批量确认如果为false就只确认当前值如果为true就会确定小于等于的值 2.否定确认
有两个方法 他们两个的区别不大唯一区别就是是否支持一次性批量拒绝消息
我们来看一些这个requeue这个参数这个参数表示拒绝后消息要如何处理如果为trueRabbitMQ就会把他重新入队发给下一个消费者如果为falseRabbitMQ就会把他从队列中删除不会把他发送给信道消费者
消费者代码
Component
public class AckConsumer {RabbitListener(queues Constants.ACK_QUEUE)public void ListenerQueue(Message message, Channel channel) throws IOException {long Tagmessage.getMessageProperties().getDeliveryTag();try {System.out.println(接收到消息: new String(message.getBody()) TagID: Tag);int num3/0; //模拟失败channel.basicAck(Tag,false);System.out.println(处理完成);}catch (Exception e){channel.basicNack(Tag,false,true);channel.basicReject(Tag,true);}}
}出现异常时的现象
因为我们是设置出现异常重新入队给下一个消费者所以会循环 如果消息成功接收 二持久化 我们刚刚做的消息确认和之前说过七大工作模式中的发送确认都是为了保证消息不丢失能够正确的接收但是我们如何保证RabbitMQ服务停掉以后生产者发送过的消息不丢失存储在RabbitMQ中的消息
1.RabbitMQ的持久化 RabbitMQ的持久化分为三个部分交换机持久化队列持久化和消息持久化
1交换机持久化 交换机持久化是通过在声明交换机的时候将durable参数设置为true实现的相当于将交换机的属性在服务器中存储M服务关闭后再次重新RabbitMQ会自动重新建立交换机此时就相当于一直存在
上面是我们把交换机设置为持久化的代码
其实我们不手动设置默认交换机也是持久的我们来看源码 我们发现就算不设置他默认也是true在我们创建的时候也是持久化的
2队列持久化
队列持久化也是在声明队列时调用方法来实现的
如果队列不设置持久化MQ重启时队列就会被删掉与此同时消息是存在队列中的所以队列上的消息也会小时
所以我们要设置消息持久化那么一定要设置队列持久化否则没用
这是设置为持久化的代码 这是设置为非持久化的代码
那我们还是来看一眼源码 然后我们点进这个QueueBuilder看看 我们发现这个就是给名字赋值用的
再来看看durable中的setDurable 我们发现是设置持久化的因为是boolean类型默认都是false所以非持久化就不需要做处理了
3消息持久化
消息持久化
消息实现持久化就需要把消息的投递模式进行更改 设置了队列和消息的持久化RabbitMQ服务器重启后消息才会存在其他情况下重新MQ都武器消息都会丢失
消息持久化代码
RequestMapping(producer)
RestController
public class AckProducer {Autowiredprivate RabbitTemplate rabbitTemplate;RequestMapping(ack)public String ackPro(){String s1ack test;Message messagenew Message(s1.getBytes(StandardCharsets.UTF_8),new MessageProperties());message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,ack,message);return 发送成功;}
}
注意我们这里传的不是一个单独字符串了而是一个消息 RabbitMQ会默认将消息视为持久化的除非队列被声明为非持久化或者发送消息时被标记为非持久化
问题 注如果我们将所有消息都设置为持久化会严重影响RabbitMQ的性能写入磁盘的速度比写入内存的速度慢很多所以我们在选择消息持久化的时候要做一个衡量
如果我们将交换机队列消息都设置了持久化后就能保证百分百不丢失数据了吗
答案是错误的
1.在持久化消息存入RabbitMQ时还有一段时间消息是在缓存上的还没有写入硬盘因为RabbitMQ并不会为每条消息都同步存盘因为这样会严重影响性能所以会有缓存等待数据一起写入硬盘。如果在这是MQ重启了消息还没有写入硬盘那这些消息就会丢失
2.消费者设置自动确认然后消费者接收消息后宕机此时消息也会丢失此时我们只需要设置为手动确认即可
第一个问题的解决方案我们下一篇博客在讲
