rabbitmq之消息确认（二）

文章目录概念生产者消息确认的策略RabbitMq工作原理图消费者消息确认的策略发布确认进阶篇前言Work Queues的基本特性请看另一篇文章https://editor.csdn.net/md/?articleId119969813概念消息确认就是用来保证生产者在发送消息到mq的过程中防止消息丢失的一种机制。发送方发消息给接受方一旦接收方接到消息后通知发送方如果发送方设置了持久化那么接受方会在消息成功持久化后在通知发送方。生产者消息确认的策略消息的确认做有很多法其中包括单条确认、批量确认、异步确认等。单条确认它是一种同步确认发布的方式也就是发布一个消息之后只有它被确认发布后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回如果在指定时间范围内这个消息没有被确认那么它将抛出异常。这种确认方式有一个最大的缺点就是:发布速度特别的慢因为如果没有确认发布的消息就会阻塞所有后续消息的发布批量确认先发布一批消息然后一起确认可以极大地提高吞吐量当然这种方式的缺点就是:当发生故障导致发布出现问题时不知道是哪个消息出现问题了所以重新发布时就会将这整批消息重新发送但这样就会造成消息重复消费反而性能下降。该方案仍然是同步的也一样阻塞消息的发布异步确认性价比最高无论可靠性还是效率都是最高效的它是利用回调函数来达到消息可靠性传递的无论接收方接受成功还是接受失败都会通知发送方。下面就让我们来详细讲解异步确认是怎么实现的。RabbitMq工作原理图每一个颜色块之间都存在着消息的确认机制我们大概分为两大类发送方确认和消费方确认其中发送方确认又分为生产者到交换器的确认和交换器到队列的确认。接下来主要讨论消费方的确认消费者消息确认的策略消费者应答发生在监听队列的消费者处理业务失败如发生了异常不符合要求的数据……这些场景我们就需要手动处理比如重新发送或者丢弃。我们知道ACK是默认是自动的自动确认会在消息发送给消费者后立即确认但存在丢失消息的可能如果消费端消费逻辑抛出异常假如你用回滚了也只是保证了数据的一致性但是消息还是丢了也就是消费端没有处理成功这条消息那么就相当于丢失了消息。消息应答模式有AcknowledgeMode.NONE自动应答AcknowledgeMode.AUTO根据情况应答AcknowledgeMode.MANUAL手动应答一般情况下我们都采用手动应答spring.rabbitmq.listener.direct.acknowledge-modeMANUAL需要注意的 basicAck 方法需要传递两个参数deliveryTag唯一标识 ID当一个消费者向 RabbitMQ 注册后会建立起一个 Channel RabbitMQ 会用 basic.deliver 方法向消费者推送消息这个方法携带了一个 delivery tag 它代表了 RabbitMQ 向该 Channel 投递的这条消息的唯一标识 ID是一个单调递增的正整数delivery tag 的范围仅限于 Channelmultiple为了减少网络流量手动确认可以被批处理当该参数为 true 时则可以一次性确认 delivery_tag 小于等于传入值的所有消息basicNack方法需要传递三个参数deliveryTag唯一标识 ID:multiplerequeue true 重回队列false 丢弃我们在nack方法中一般设置 false因为重发没有意义。basicReject方法需要传递两个参数deliveryTag唯一标识 ID上面已经解释了。requeue在reject方法里必须设置true。还要说明一下建议大家不要重发重发后基本还是失败因为出现问题一般都是异常导致的出现异常的话我的观点是丢弃这个消息然后在catch里做补偿操作。到此,可以进行测试。RestControllerpublicclassAsyncConfirmController{AutowiredprivateRabbitTemplaterabbitTemplate;GetMapping(/async/{id})publicStringAETest(PathVariableIntegerid){OrderordernewOrder(id,胖虎);rabbitTemplate.convertAndSend(confirm_exchange,,order);return成功;}---------------------------------ConfigurationpublicclassAsyncConfirmListener{BeanpublicQueueconfirmQueue(){returnnewQueue(confirm_queue);}BeanpublicFanoutExchangeconfirmExchange(){returnnewFanoutExchange(confirm_exchange);}//交换器绑定队列BeanBindingbindingExchangeConfirm(QueueconfirmQueue,FanoutExchangeconfirmExchange){returnBindingBuilder.bind(confirmQueue).to(confirmExchange);}}发布确认进阶篇通过上面的列子我们探讨了如何应对队列在发送消息给消费者的过程中出现了问题导致消费者无法消费的问题那么生产者发送消息给交换机的过程中出现了问题以及交换机发送消息给队列的过程中出现了问题应该如何应对具体请看我的一下篇文章生产方发布确认https://blog.csdn.net/qq_42872629/article/details/120123218?spm1001.2014.3001.5501