网站建设可以自己弄吗知乎,自己怎么做商城网站视频教程,网上推广怎么拉客户,网站开发实践报告专栏导航 RabbitMQ入门指南 从零开始了解大数据 目录 专栏导航 前言 一、消息丢失的可能性 1.发送消息时丢失#xff1a; 2.MQ导致消息丢失#xff1a; 3.消费者处理消息时消息丢失#xff1a; 二、生产者可靠性 1.生产者重试机制 2.生产者确认机制 总结 前言 Ra… 专栏导航 RabbitMQ入门指南 从零开始了解大数据 目录 专栏导航 前言 一、消息丢失的可能性 1.发送消息时丢失 2.MQ导致消息丢失 3.消费者处理消息时消息丢失 二、生产者可靠性 1.生产者重试机制 2.生产者确认机制 总结 前言 RabbitMQ是一个高效、可靠的开源消息队列系统广泛用于软件开发、数据传输、微服务等领域。本文主要介绍了消息丢失的可能性、生产者可靠性中的生产者重试机制和生产者确认机制等内容。 一、消息丢失的可能性 流程图如下 1.发送消息时丢失 生产者发送消息时连接MQ失败当生产者尝试与RabbitMQ建立连接但失败时发送消息的操作将无法完成。这可能是由于网络问题、RabbitMQ服务未运行或配置错误导致的。 生产者发送消息到达MQ后未找到Exchange在RabbitMQ中生产者将消息发送到ExchangeExchange再路由到一个或多个Queue。如果生产者发送消息到达MQ后未找到正确的Exchange那么消息将无法被路由到正确的Queue从而丢失。 生产者发送消息到达MQ的Exchange后未找到合适的Queue即使消息找到了Exchange如果Exchange没有正确配置或没有找到合适的Queue来路由消息那么消息也会丢失。这可能是由于配置错误或Queue不存在导致的。 消息到达MQ后处理消息的进程发生异常在某些情况下当消息到达MQ后处理消息的进程可能会发生异常导致消息无法被处理。这可能是由于代码错误、资源限制或其他系统问题导致的。 2.MQ导致消息丢失 消息到达MQ保存到队列后尚未消费就突然宕机在某些情况下如果RabbitMQ服务突然宕机而消息已经保存到队列中但尚未被消费者消费那么这些消息可能会丢失。为了防止这种情况可以使用持久化配置来确保消息在MQ宕机后仍然保留在磁盘上。 3.消费者处理消息时消息丢失 消息接收后尚未处理突然宕机如果消费者在接收消息后尚未处理就突然宕机那么这些消息可能会丢失。为了防止这种情况可以使用确认机制来确保消息在被消费者成功处理之前一直保留在队列中。 消息接收后处理过程中抛出异常在处理消息的过程中如果消费者代码出现异常并抛出错误那么可能会导致消息处理失败。为了防止这种情况可以对消费者的代码进行充分的测试和异常处理以确保其能够正确、稳定地处理消息。 二、生产者可靠性 1.生产者重试机制 由于网络波动可能会出现客户端连接MO失败的情况为了解决这个问题Spring AMQP提供了消息发送时的重试机制 。 在publisher服务的application.yml文件中添加配置 spring:rabbitmq:connection-timeout: 1s # 设置MQ的连接超时时间template:retry:enabled: true # 开启超时重试机制initial-interval: 1000ms # 失败后的初始等待时间multiplier: 1 # 失败后下次的等待时长倍数下次等待时长initial-interval * multipliermax-attempts: 3 # 最大重试次数 当网络不稳定时重试机制可以有效提高消息发送的成功率。然而Spring AMQP提供的重试机制是阻塞式的这意味着在等待重试的过程中当前线程会被阻塞。对于对业务性能有要求的应用建议禁用重试机制。如果仍需使用建议合理配置等待时长和重试次数并考虑使用异步线程来执行发送消息的代码以避免阻塞主线程。 2.生产者确认机制 RabbitMQ提供了Publisher Confirm和Publisher Return两种确认机制。开启确机制认后当生产者发送消息给MQ后MQ会根据消息处理的情况返回不同的回执。返回的结果有以下几种情况 消息投递到了MQ但是路由失败。此时会通过PublisherReturn返回路由异常原因然后返回ACK告知投递成功。临时消息投递到了MQ并且入队成功返回ACK告知投递成功。持久消息投递到了MQ并且入队完成持久化返回ACK告知投递成功。其他情况都会返回NACK告知投递失败。 ACK和NACK属于Publisher Confirm机制ACK是投递成功NACK是投递失败。return属于Publisher Return机制。默认两种机制都是关闭状态需要通过配置文件来开启。 开启生产者确认 在publisher服务的application.yml文件中添加配置 spring:rabbitmq:publisher-confirm-type: correlated # 开启publisher confirm机制并设置confirm类型publisher-returns: true # 开启publisher return机制 publisher-confirm-type有三种模式一般推荐使用correlated回调机制 none关闭confirm机制simple同步阻塞等待MQ的回执correlatedMQ异步回调返回回执 定义ReturnCallback 在publisher服务定义一个配置类每个RabbitTemplate只能配置一个ReturnCallback Slf4j Configuration public class MqConfirmConfig implements ApplicationContextAware {Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {RabbitTemplate rabbitTemplate applicationContext.getBean(RabbitTemplate.class);// 配置回调rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {Overridepublic void returnedMessage(ReturnedMessage returned) {log.info(消息的return callback:);log.info(exchange:{}, key:{}, msg:{}, code:{}, text:{},returned.getExchange(), returned.getRoutingKey(), returned.getMessage(), returned.getReplyCode(), returned.getReplyText());}});} } 定义ConfirmCallback 由于每条消息的处理逻辑可能不同因此需要在每次发送消息时定义ConfirmCallback当调用RabbitTemplate的convertAndSend方法时需要多传递一个参数。 这个参数是一个CorrelationData对象包含两个核心内容 id这是消息的唯一标识MQ会使用它来判断不同的消息的回执避免混淆。SettableListenableFuture这是一个Future对象MQ将通过它返回回执结果。可以提前为CorrelationData中的Future添加回调函数来处理消息回执。 在publisher服务中的测试类添加一个测试方法实现消息发送并且添加ConfirmCallback Testvoid testConfirmCallback() throws InterruptedException {// 1.创建CorrelationDataCorrelationData cd new CorrelationData(UUID.randomUUID().toString());// 2.添加ConfirmCallbackcd.getFuture().addCallback(new ListenableFutureCallbackCorrelationData.Confirm() {Overridepublic void onFailure(Throwable ex) {log.error(消息回调失败, ex);}Overridepublic void onSuccess(CorrelationData.Confirm result) {log.debug(收到confirm callback回执);if (result.isAck()) {// 消息发送成功log.debug(消息发送成功收到ACK);} else {// 消息发送失败log.error(消息发送失败收到NACK原因{}, result.getReason());}}});rabbitTemplate.convertAndSend(test.direct, red, test, cd);} 生产者确认会带来额外的网络和系统资源开销因此应尽量避免使用。如果确实需要使用则无需开启Publisher-Return机制因为一般路由失败是业务问题。对于NACK消息可以有限次数重试如果仍然失败则记录异常消息。 总结 RabbitMQ是一个开源的消息队列软件旨在提供可靠的消息传递和消息队列功能。本文主要介绍了消息丢失的可能性、生产者可靠性中的生产者重试机制和生产者确认机制等内容希望对大家有所帮助。