宝塔装wordpress,seo怎么赚钱,小程序应用开发,平凉网站开发:::info 使用场景
异步发送#xff08;验证码、短信、邮件。。。#xff09;MySQL 和 Redis、ES 之间的数据同步分布式事务削峰填谷… :::
如何保证消息不丢失 上图是消息正常发送的一个过程#xff0c;那在哪个环节中消息容易丢失#xff1f;在哪一个环节都可能丢失
生… :::info 使用场景
异步发送验证码、短信、邮件。。。MySQL 和 Redis、ES 之间的数据同步分布式事务削峰填谷… :::
如何保证消息不丢失 上图是消息正常发送的一个过程那在哪个环节中消息容易丢失在哪一个环节都可能丢失
生产者宕机消息就可能到达不了交换机或者消息未到达队列消息发送成功后消费者还没有消费前MQ 宕机就有可能导致队列中消息丢失消费者宕机导致消费者未接收到消息
生产者队列消费者三个层面都有可能导致消息丢失所以保证消息不丢失需要从这三个层面解决
生产者确认机制
RabbitMQ 提供了 publisher confirm 机制来避免消息发送到 MQ 过程中丢失。消息发送到 MQ 后会返回一个结果给发送者表示消息是否处理成功。 如果消息发送成功就会返回 publish-confirm ack如果发送到交换机失败就会返回 publish-confirm nack如果发送到队列失败就会返回 publish-return ack。 消息失败之后如何处理呢
回调方法即时重发记录日志保存到数据库然后定时重发成功发送后即刻删除表中的数据
消息持久化
MQ 默认是内存存储消息开启持久化功能可以确保缓存在 MQ 中的消息不丢失。
交换机持久化
Bean
public DirectExchange simpleExchange(){// 三个参数交换机名称、是否持久化、当没有queue与其绑定时是否自动删除return new DirectExchange(simple.direct,true,false);
}队列持久化
Bean
public Queue simpleQueue(){// 使用QueueBuilder构建队列durable就是持久的return QueueBuilder.durable(simple.queue).build();
}消息持久化SpringAMQP 中的消息默认是持久的可以通过 MessageProperties 中的 DeliveryMode 来指定
Message msg MessageBuilder.withBody(message.getBytes(StandardCharsets.UTF_8)) // 消息体.setDeliveryMode(MessageDeliveryMode.PERSISTENE) // 持久化.build();消费者确认
RabbitMQ 支持消费者确认机制即消费者处理消息后可以向 MQ 发送 ack 回执MQ 收到 ack 回执后才会删除消息。 SpringAMQP 则允许配置三种确认模式
manual手动 ack需要在业务代码结束后调用 api 发送 ack。auto自动 ack由 spring 监测 listener 代码是否出现异常没有异常则返回 ack抛出异常则返回 nack。none关闭 ackMQ 假定消费者获取消息后会成功处理因此消息投递后立即被删除
我们可以利用 Spring 的 retry 机制在消费者出现异常时利用本地重试设置重试次数当次数达到了以后如果消息依然失败将消息投递到异常交换机交由人工处理。 RabbitMQ 如何保证消息不丢失
开启生产者确认机制确保生产者的消息能到达队列开启持久化功能确保消息未消费前在队列中不会丢失开启消费者确认机制为 auto由 spring 确认消息处理成功后完成 ack开启消费者失败重试机制多次重试失败后将消息投递到异常交换机交由人工处理
消息的重复消费问题如何解决的
为什么会出现重复消费的问题
网络抖动消费者挂了 消费者已经处理完消息还没来得及给 MQ 发送确认这时网络发生了抖动或者消费者挂了等网络恢复之后或者消费者重启之后因为队列没有收到确认所以消息还在 MQ 中因为我们设置了重试机制消费者就会重新消费消息。 解决方案
每条消息设置一个唯一的标识 id消费者收到消息后去业务 id 是否存在幂等方案分布式锁、数据库锁悲观锁、乐观锁
RabbitMQ 死信交换机(RabbitMQ 延迟队列有了解过嘛)
延迟队列进入队列的消息会被延迟消费的队列场景超时订单、限时优惠、定时发布
延迟队列死信交换机TTL消息的生存时间
死信交换机
当一个队列中的消息满足下列情况之一时可以成为死信dead letter:
消费者使用 basic.reject 或 basic.nack 声明消费失败并且消息的 requeue 参数为 false消息是一个过期消息超时无人消费要投递的队列消息堆积满了最早的消息可能成为死信
如果该队列配置了 dead-letter-exchange 属性指定了一个交换机那么队列中的死信就会投递到这个交换机中而这个交换机称为死信交换机Dead Letter Exchange简称 DLX。
Bean
public Queue ttlQueue(){return QueueBuilder.durable(simple.queue) // 指定队列名称并持久化.ttl(10000) // 设置队列的超时时间10秒.deadLetterExchange(dl.direct) // 指定死信交换机.build();
}TTL
TTL也就是 Time-To-Live。如果一个队列中的消息 TTL 结束仍未消费则会变成死信ttl 超时分为两种情况
消息所在的队列设置了存活时间消息本身设置了存活时间 哪个 ttl 短以哪个为准。
// 创建消息
Message message MessageBuilder.withBody(hello, ttl message.getBytes(StandardCharsets.UTF_8)).setExpiration(5000).build();
// 消息ID需要封装到CorrelationData中
CorrelationData correlationData new correlationData(UUID.randomUUID().toString());
// 发送消息
rabbitTemplate.convertAndSend(ttl.direct,ttl,message,correlationData);延迟队列插件
实现延迟队列还可以使用官方提供的插件。DelayExchange 插件需要安装在 RabbitMQ 中 RabbitMQ 有一个官方的插件社区地址为https://www.rabbitmq.com/community-plugins.html
RabbitListener(bindings QueueBinding(value Queue(name delay.queue, durabletrue),exchange Exchange(namedelay.direct,delayedtrue)keydelay
))
public void listenDelayedQueue(String msg){log.info(接收到 delay.queue的延迟消息{},msg);
}
// 创建消息
Message message MessageBuilder.withBody(hello,delayed message.getBytes(StandardCharsets.UTF_8)).setHeader(x-delay,10000).build();
// 消息ID需要封装到CorrelationData中
CorrelationData correlationData new CorrelationData(UUID.randomUUID().toString());
// 发送消息
rabbitTemplate.convertAndSend(delay.direct,delay,message,correlationData);RabbitMQ 死信交换机RabbitMQ 延迟队列有了解过嘛
我们当时一个业务使用到了延迟队列超时订单、限时优惠、定时发布…其中延迟队列就用到了死信交换机和 TTL 实现的消息超时未消费就会变成死信死信的其他情况拒绝被消费队列满了
延迟队列插件实现延迟队列 DelayExchange
声明一个交换机添加 delayed 属性为 true发送消息时添加 x-delay 头值为超时时间
如果有 100 万消息堆积在 MQ如何解决消息堆积怎么解决
当生产者发送消息的速度超过了消费者处理消息的速度就会导致队列中的消息堆积直到队列存储消息达到上限。之后发送的消息就会成为死信可能会被丢弃这就是消息堆积问题。 解决消息堆积有三种思路
增加更多消费者提高消费速度在消费者内开启线程池加快消息处理速度扩大队列容积提高堆积上限
惰性队列
惰性队列的特征如下
接收到消息后直接存入磁盘而非内存消费者要消费消息时才会从磁盘中读取并加载到内存支持百万条的消息存储
Bean
public Queue lazyQueue(){return QueueBuilder.durable(lazy.queue).lazy() // 开启x-queue-mode为lazy.build();
}
// 或者
RabbitListener(queuesToDeclare Queue(namelazy.queue,durabletrue,argumentsArgument(namex-queue-mode,valuelazy)
))
public void listenLazyQueue(String msg){log.info(接收到lazy.queue的消息{},msg);
}如果有 100 万消息堆积在 MQ如何解决 解决消息堆积有三种思路
增加更多的消费者提高消费速度在消费者内开启线程池加快消息处理速度扩大队列容积提高堆积上限采用惰性队列 在声明队列的时候可以设置属性 x-queue-mode 为 lazy即为惰性队列基于磁盘存储消息上限高性能比较稳定但基于磁盘存储受限与磁盘 IO时效性会降低
RabbitMQ 高可用机制
在生产环境下使用集群来保证高可用性普通集群、镜像集群、仲裁队列
普通集群
普通集群或者叫标准集群classic cluster具备下列特征
会在集群的各个节点共享部分数据包括交换机、队列元信息。不包含队列中的消息。当访问集群某节点时如果队列不在该节点会从数据所在节点传递到当前节点并返回。队列所在节点宕机队列中的消息就会丢失。 镜像集群
镜像集群本质是主从模式具备下面的特征
交换机、队列、队列中的消息会在各个 mq 的镜像节点之间同步备份。创建队列的节点被称为该队列的主节点备份到其他节点叫做该队列的镜像节点。一个队列的主节点可能是另一个队列的镜像节点。所有操作都是在主节点完成然后同步给镜像节点。主宕机后镜像节点会替代成新的主。
仲裁队列
仲裁队列仲裁队列是 3.8 版本以后才有的新功能用来替代镜像队列具备下列特征
与镜像队列一样都是主从模式支持主从数据同步使用非常简单没有复杂的配置主从同步基于 Raft 协议强一致
Bean
public Queue quorumQueue(){return QueueBuilder.durable(quorum.queue) // 持久化.quorum() //仲裁队列.build();
}RabbitMQ 的高可用机制有了解过嘛
在生产环境下我们当时采用的镜像模式搭建的集群共有 3 个节点。镜像队列结构是一主多从从就是镜像所有操作都是主节点完成然后同步给镜像节点。主节点宕机后镜像节点会替代成新的主如果在主从同步完成前主就已经宕机可能出现数据丢失
出现数据丢失怎么解决 我们可以采用仲裁队列与镜像队列一样都是主从模式支持主从数据同步主从同步基于 Raft 协议强一致。并且使用起来也非常简单不需要额外的配置在声明队列的时候只要指定这个是仲裁队列即可。
