cloud面试:| 组件 | 作用 | 替代品 || -------------------------- | ------------ | ----------------- || **Eureka / Nacos** | 服务注册与发现 | Consul、Zookeeper || **Ribbon / LoadBalancer** | 客户端负载均衡?? | — || **Feign / OpenFeign** | 声明式 HTTP 客户端?? | — || **Hystrix / Resilience4j** | 熔断、降级、限流 | Sentinel || ** Gateway** | API 网关 | Kong、APISIX || **Config / Nacos** | 分布式配置中心 | Apollo || **Sleuth Zipkin** | 分布式链路追踪 | SkyWalking、Jaeger || **Bus** | 消息总线配置刷新广播 | — || **Stream** | 消息驱动微服务 | — |客户端负载均衡? LoadBalancer,已成为官方推荐的默认实现.调用? OpenFeign .熔断、降级、限流? Sentinel.分布式链路追踪? Zipkin?| 对比维度 | Eureka | Nacos || ---------- | ---------------------------- | ----------------- || **CAP 理论** | AP高可用容忍数据不一致 | 同时支持 AP 和 CP可切换 || **健康检查** | 客户端心跳默认30s | 客户端心跳 服务端主动探测 || **配置中心** | 不支持需配合 Spring Cloud Config | 内置配置中心 || **数据一致性** | 对等复制Peer to Peer | 支持 Raft 协议CP模式 || **生态** | Netflix 开源已进入维护模式 | 阿里开源活跃迭代支持 K8s || **性能** | 大规模集群下可能出现复制延迟 | 性能更好支持百万级实例 |Nacos 的 AP/CP 切换通过 nacos.core.protocol.raft 配置实现。临时实例走 APDistro协议持久化实例走 CPRaft协议。注册中心nacos如果挂了, 会发生什么?已有服务间调用不受影响因为使用本地缓存。但新服务无法注册服务上下线无法感知配置无法更新。集群模式下单节点故障会自动切换到其他节点。┌─────────────────────────────────────────────────────────────┐│ Nacos Server 宕机 │├─────────────────────────────────────────────────────────────┤│ ││ 场景1已有服务实例之间的调用 ││ ├── 客户端本地有缓存的服务列表 → 继续正常调用 ✅ ││ ├── 调用不受任何影响直到缓存过期 ││ └── 这是 AP 架构的核心优势 ││ ││ 场景2新服务实例启动 ││ ├── 无法注册到 Nacos → 注册失败 ❌ ││ ├── 其他服务无法通过 Nacos 发现它 ││ └── 但服务本身可以正常启动取决于是否开启失败容忍 ││ ││ 场景3服务实例下线/宕机 ││ ├── Nacos 无法感知 → 不会从列表中剔除 ❌ ││ ├── 其他客户端缓存中仍有该实例 ││ └── 调用方会调用到已宕机的实例 → 触发熔断降级 ││ ││ 场景4配置中心功能 ││ ├── 无法推送新配置 ❌ ││ ├── 客户端本地缓存的配置仍然有效 ││ └── 应用继续用旧配置运行 ││ │└─────────────────────────────────────────────────────────────┘Nacos Client 维护了一份本地内存缓存这是 Nacos 挂了还能继续调用的核心原因.private ConcurrentHashMapString, ListInstance serviceInfoMap;缓存更新时机启动时从 Nacos Server 全量拉取 → 写入内存 磁盘运行时长轮询接收变更 → 更新内存 磁盘Nacos 挂了使用内存缓存 → 内存丢了读磁盘缓存nacos服务注册和发现流程?面试标准回答1注册流程 服务启动时向 Nacos Server 发送 HTTP POST 请求携带 IP、端口、服务名等信息。Nacos 将实例信息存储在内存中临时实例或内存MySQL持久实例。注册成功后客户端每 5 秒发送一次心跳维持注册状态。2发现流程 消费者启动时从 Nacos 全量拉取服务列表缓存到本地内存和磁盘。然后持续监听服务变更有变更时 Nacos 立即推送(主动推送给所有订阅该服务的客户端实时性可达毫秒级)。实际调用时直接从本地缓存选择实例减少网络IO开销。LoadBalancer,负载均衡策略有哪些如何自定义轮询策略,随机策略, 当内置的轮询或随机策略无法满足业务需求如基于用户 ID 哈希、灰度发布、同机房优先、开发环境隔离等时可以通过实现 ReactorServiceInstanceLoadBalancer 接口来自定义算法。feign和openfeign区别.Feign 是 Netflix 开源的原生声明式 HTTP 客户端2019 年后已停止新功能迭代OpenFeign 是 Spring Cloud 社区在其基础上 fork 演进的增强版本深度适配 Spring 生态是当前微服务项目的主流选择.Feign仅支持原生自定义注解如 RequestLine、Param语法与 Spring 开发习惯完全割裂需要重新学习 API 规范 。OpenFeign原生兼容 Spring MVC 全套注解直接使用 GetMapping、PostMapping、PathVariable 等日常开发常用注解无需额外记忆新语法大幅降低学习成本 。熔断、降级、限流? Sentinel.使用 SentinelResource 注解标记核心业务方法实现限流与降级的解耦RestControllerpublic class OrderController {GetMapping(/create)SentinelResource(value createOrder,blockHandler handleBlock, // 限流/熔断兜底fallback handleFallback // 业务异常兜底)public String createOrder(RequestParam String userId) {// 模拟业务逻辑if (error.equals(userId)) {throw new RuntimeException(Business Error);}return Order Created;}// 限流/熔断后的处理public String handleBlock(String userId, BlockException ex) {return System Busy, Please Try Later;}// 业务异常后的处理public String handleFallback(String userId, Throwable ex) {return Service Unavailable: ex.getMessage();}}Q2: Sentinel 的底层原理是什么如何实现高性能统计滑动窗口算法Sentinel 基于滑动时间窗口Sliding Window统计指标。将时间划分为多个小的 Bucket默认 500ms每个 Bucket 记录该时间段内的请求数、异常数、RT 等。原子操作使用 LongAdder 或 AtomicInteger 保证高并发下计数的线程安全避免锁竞争。LeapArray核心数据结构负责窗口的复用和过期清理确保内存占用稳定。优势相比固定窗口滑动窗口解决了“临界点突变”问题统计更平滑准确。Q3: 如何实现分布式限流Sentinel 支持吗默认单机Sentinel 默认是单机限流每个实例独立统计。集群模式Sentinel 支持集群限流但需要部署独立的 Token Server令牌服务器。客户端向 Token Server 申请令牌Token Server 全局控制配额。替代方案如果不想部署 Token Server可使用 Redis Lua如 Redisson RateLimiter实现全局限流或在网关层Spring Cloud Gateway做全局限流。限流后用户体验很差怎么优化快速失败太生硬不要直接返回 500 或空白页。友好提示通过 blockHandler 返回统一的 JSON 格式错误码如 code: 429, msg: 系统繁忙请稍后重试。排队等待对于秒杀等场景使用“匀速排队”模式让用户等待几秒后成功而不是直接拒绝。前端优化前端按钮置灰、增加 loading 动画防止用户重复点击。Spring Cloud Gateway--,,,,--RabbitMQ核心概念Producer / Consumer生产者与消费者模型Queue消息队列存储消息Exchange交换机决定消息如何路由到队列Binding队列与交换机的绑定关系Routing Key / Binding Key消息路由规则消息确认机制自动 / 手动 ACK消息持久化防止消息丢失MQ 的常见问题有消息的顺序问题消息的重复问题消息的可靠传输消息的顺序问题消息的重复问题消息的可靠传输消息不可靠的情况可能是消息丢失劫持等原因丢失又分为生产者丢失消息、消息列表丢失消息、消费者丢失消息生产者丢失消息从生产者弄丢数据这个角度来看RabbitMQ提供transaction和confirm模式来确保生产者不丢消息transaction机制就是说发送消息前开启事务channel.txSelect()然后发送消息如果发送过程中出现什么异常事务就会回滚channel.txRollback()如果发送成功则提交事务channel.txCommit()。然而这种方式有个缺点吞吐量下降confirm模式用的居多一旦channel进入confirm模式所有在该信道上发布的消息都将会被指派一个唯一的ID从1开始一旦消息被投递到所有匹配的队列之后rabbitMQ就会发送一个ACK给生产者包含消息的唯一ID这就使得生产者知道消息已经正确到达目的队列了如果rabbitMQ没能处理该消息则会发送一个Nack消息给你生产者可以进行重试操作。发送方确认机制先说配置和使用配置文件spring: rabbitmq: publisher-confirm-type: correlated # 开启发送方确认机制配置属性有三种分别为none表示禁用发送方确认机制correlated表示开启发送方确认机制simple表示开启发送方确认机制并支持waitForConfirms()和waitForConfirmsOrDie()的调用。这里一般使用correlated开启发送方确认机制即可至于simple的waitForConfirms()方法调用是指串行确认方法即生产者发送消息后调用该方法等待 RabbitMQ Server 确认如果返回 false 或超时未返回则进行消息重传。由于串行性能较差这里一般都是用异步 confirm 模式。通过调用setConfirmCallback()实现异步 confirm 模式感知消息发送结果/** * 消息业务实现类 * * author 单程车票 */ Service public class RabbitMQServiceImpl { Autowired private RabbitTemplate rabbitTemplate; Override public void sendMessage() { // 发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); // 设置消息确认回调方法 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { /** * MQ确认回调方法 * param correlationData 消息的唯一标识 * param ack 消息是否成功收到 * param cause 失败原因 */ Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { // 记录日志 log.info(ConfirmCallback...correlationData[correlationData]ack:[ack]cause:[cause]); if (!ack) { // 出错处理 ... } } }); } }3. 保证消息在 RabbitMQ Server 中的持久化对于消息的持久化只需要在发送消息时将消息持久化并且在创建交换机和队列时也保证持久化即可。配置如下/** * 消息队列 */ Bean public Queue queue() { // 四个参数name队列名、durable持久化、exclusive独占、autoDelete自动删除 return new Queue(MESSAGE_QUEUE, true); } /** * 直接交换机 */ Bean public DirectExchange exchange() { // 四个参数name交换机名、durable持久化、autoDelete自动删除、arguments额外参数 return new DirectExchange(Direct_Exchange, true, false); }在创建交换机和队列时通过构造方法将持久化的参数都设置为 true 即可实现交换机和队列的持久化。Override public void sendMessage() { // 构造消息将消息持久化 Message message MessageBuilder.withBody(单程车票.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build(); // 向MQ发送消息消息内容都为消息表记录的id rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); }在发送消息前通过调用MessageBuilder的setDeliveryMode(MessageDeliveryMode.PERSISTENT)在构造消息时设置消息持久化MessageDeliveryMode.PERSISTENT即可实现对消息的持久化。生产者发送消息后通过调用setConfirmCallback()可以将信道设置为 confirm 模式所有消息会被指派一个消息唯一标识当消息被发送到 RabbitMQ Server 后Server 确认消息后生产者会回调设置的方法从而实现生产者可以感知到消息是否正确无误的投递从而实现发送方确认机制。并且该模式是异步的发送消息的吞吐量会得到很大提升。上面就是发送放确认机制的配置和使用使用这种机制可以保证生产者的消息可靠性投递并且性能较好。消息队列丢数据消息持久化。处理消息队列丢数据的情况一般是开启持久化磁盘的配置。这个持久化配置可以和confirm机制配合使用你可以在消息持久化磁盘后再给生产者发送一个Ack信号。这样如果消息持久化磁盘之前rabbitMQ阵亡了那么生产者收不到Ack信号生产者会自动重发。那么如何持久化呢这里顺便说一下吧其实也很容易就下面两步将queue的持久化标识durable设置为true则代表是一个持久的队列发送消息的时候将deliveryMode2这样设置以后即使rabbitMQ挂了重启后也能恢复数据消费者丢失消息消费者丢数据一般是因为采用了自动确认消息模式改为手动确认消息即可消费者在收到消息之后处理消息之前会自动回复RabbitMQ已收到消息如果这时处理消息失败就会丢失该消息解决方案处理消息成功后手动回复确认消息。4. 保证消费者消费的消息不丢失在保证发送方和 RabbitMQ Server 的消息可靠性的前提下只需要保证消费者在消费消息时异常消息不丢失即可保证消息的可靠性。RabbitMQ 提供了消费者应答机制来使 RabbitMQ 能够感知到消费者是否消费成功消息默认情况下消费者应答机制是自动应答的也就是RabbitMQ 将消息推送给消费者便会从队列删除该消息如果消费者在消费过程失败时消息就存在丢失的情况。所以需要将消费者应答机制设置为手动应答只有消费者确认消费成功后才会删除消息从而避免消息的丢失。下面来看看如何配置消费者手动应答spring: rabbitmq: publisher-confirm-type: correlated # 开启发送方确认机制 publisher-returns: true # 开启消息返回 template: mandatory: true # 消息投递失败返回客户端 listener: simple: acknowledge-mode: manual # 开启手动确认消费机制通过listener.simple.acknowledge-mode manual即可将消费者应答机制设置为手动应答。之后只需要在消费消息时通过调用channel.basicAck()与channel.basicNack()来根据业务的执行成功选择是手动确认消费还是手动丢弃消息。/** * 监听消费队列的消息 */ RabbitListener(queues RabbitMQConfig.MESSAGE_QUEUE) public void onMessage(Message message, Channel channel) { // 获取消息索引 long index message.getMessageProperties().getDeliveryTag(); // 解析消息 byte[] body message.getBody(); ... try { // 业务处理 ... // 业务执行成功则手动确认 channel.basicAck(index, false); }catch (Exception e) { // 记录日志 log.info(出现异常{}, e.getMessage()); try { // 手动丢弃信息 channel.basicNack(index, false, false); } catch (IOException ex) { log.info(丢弃消息异常); } } }这里说明一下basicAck()与basicNack()的参数说明:void basicAck(long deliveryTag, boolean multiple)方法会抛异常deliveryTag该消息的indexmultiple是否批量处理true 表示将一次性ack所有小于deliveryTag的消息void basicNack(long deliveryTag, boolean multiple, boolean requeue)方法会抛异常deliveryTag该消息的indexmultiple是否批量处理true 表示将一次性ack所有小于deliveryTag的消息requeue被拒绝的是否重新入队列true 表示添加在队列的末端false 表示丢弃通过设置手动确认消费者应答机制即可保证消费者在消费信息时的消息可靠性。
中间件面试题
cloud面试:| 组件 | 作用 | 替代品 || -------------------------- | ------------ | ----------------- || **Eureka / Nacos** | 服务注册与发现 | Consul、Zookeeper || **Ribbon / LoadBalancer** | 客户端负载均衡?? | — || **Feign / OpenFeign** | 声明式 HTTP 客户端?? | — || **Hystrix / Resilience4j** | 熔断、降级、限流 | Sentinel || ** Gateway** | API 网关 | Kong、APISIX || **Config / Nacos** | 分布式配置中心 | Apollo || **Sleuth Zipkin** | 分布式链路追踪 | SkyWalking、Jaeger || **Bus** | 消息总线配置刷新广播 | — || **Stream** | 消息驱动微服务 | — |客户端负载均衡? LoadBalancer,已成为官方推荐的默认实现.调用? OpenFeign .熔断、降级、限流? Sentinel.分布式链路追踪? Zipkin?| 对比维度 | Eureka | Nacos || ---------- | ---------------------------- | ----------------- || **CAP 理论** | AP高可用容忍数据不一致 | 同时支持 AP 和 CP可切换 || **健康检查** | 客户端心跳默认30s | 客户端心跳 服务端主动探测 || **配置中心** | 不支持需配合 Spring Cloud Config | 内置配置中心 || **数据一致性** | 对等复制Peer to Peer | 支持 Raft 协议CP模式 || **生态** | Netflix 开源已进入维护模式 | 阿里开源活跃迭代支持 K8s || **性能** | 大规模集群下可能出现复制延迟 | 性能更好支持百万级实例 |Nacos 的 AP/CP 切换通过 nacos.core.protocol.raft 配置实现。临时实例走 APDistro协议持久化实例走 CPRaft协议。注册中心nacos如果挂了, 会发生什么?已有服务间调用不受影响因为使用本地缓存。但新服务无法注册服务上下线无法感知配置无法更新。集群模式下单节点故障会自动切换到其他节点。┌─────────────────────────────────────────────────────────────┐│ Nacos Server 宕机 │├─────────────────────────────────────────────────────────────┤│ ││ 场景1已有服务实例之间的调用 ││ ├── 客户端本地有缓存的服务列表 → 继续正常调用 ✅ ││ ├── 调用不受任何影响直到缓存过期 ││ └── 这是 AP 架构的核心优势 ││ ││ 场景2新服务实例启动 ││ ├── 无法注册到 Nacos → 注册失败 ❌ ││ ├── 其他服务无法通过 Nacos 发现它 ││ └── 但服务本身可以正常启动取决于是否开启失败容忍 ││ ││ 场景3服务实例下线/宕机 ││ ├── Nacos 无法感知 → 不会从列表中剔除 ❌ ││ ├── 其他客户端缓存中仍有该实例 ││ └── 调用方会调用到已宕机的实例 → 触发熔断降级 ││ ││ 场景4配置中心功能 ││ ├── 无法推送新配置 ❌ ││ ├── 客户端本地缓存的配置仍然有效 ││ └── 应用继续用旧配置运行 ││ │└─────────────────────────────────────────────────────────────┘Nacos Client 维护了一份本地内存缓存这是 Nacos 挂了还能继续调用的核心原因.private ConcurrentHashMapString, ListInstance serviceInfoMap;缓存更新时机启动时从 Nacos Server 全量拉取 → 写入内存 磁盘运行时长轮询接收变更 → 更新内存 磁盘Nacos 挂了使用内存缓存 → 内存丢了读磁盘缓存nacos服务注册和发现流程?面试标准回答1注册流程 服务启动时向 Nacos Server 发送 HTTP POST 请求携带 IP、端口、服务名等信息。Nacos 将实例信息存储在内存中临时实例或内存MySQL持久实例。注册成功后客户端每 5 秒发送一次心跳维持注册状态。2发现流程 消费者启动时从 Nacos 全量拉取服务列表缓存到本地内存和磁盘。然后持续监听服务变更有变更时 Nacos 立即推送(主动推送给所有订阅该服务的客户端实时性可达毫秒级)。实际调用时直接从本地缓存选择实例减少网络IO开销。LoadBalancer,负载均衡策略有哪些如何自定义轮询策略,随机策略, 当内置的轮询或随机策略无法满足业务需求如基于用户 ID 哈希、灰度发布、同机房优先、开发环境隔离等时可以通过实现 ReactorServiceInstanceLoadBalancer 接口来自定义算法。feign和openfeign区别.Feign 是 Netflix 开源的原生声明式 HTTP 客户端2019 年后已停止新功能迭代OpenFeign 是 Spring Cloud 社区在其基础上 fork 演进的增强版本深度适配 Spring 生态是当前微服务项目的主流选择.Feign仅支持原生自定义注解如 RequestLine、Param语法与 Spring 开发习惯完全割裂需要重新学习 API 规范 。OpenFeign原生兼容 Spring MVC 全套注解直接使用 GetMapping、PostMapping、PathVariable 等日常开发常用注解无需额外记忆新语法大幅降低学习成本 。熔断、降级、限流? Sentinel.使用 SentinelResource 注解标记核心业务方法实现限流与降级的解耦RestControllerpublic class OrderController {GetMapping(/create)SentinelResource(value createOrder,blockHandler handleBlock, // 限流/熔断兜底fallback handleFallback // 业务异常兜底)public String createOrder(RequestParam String userId) {// 模拟业务逻辑if (error.equals(userId)) {throw new RuntimeException(Business Error);}return Order Created;}// 限流/熔断后的处理public String handleBlock(String userId, BlockException ex) {return System Busy, Please Try Later;}// 业务异常后的处理public String handleFallback(String userId, Throwable ex) {return Service Unavailable: ex.getMessage();}}Q2: Sentinel 的底层原理是什么如何实现高性能统计滑动窗口算法Sentinel 基于滑动时间窗口Sliding Window统计指标。将时间划分为多个小的 Bucket默认 500ms每个 Bucket 记录该时间段内的请求数、异常数、RT 等。原子操作使用 LongAdder 或 AtomicInteger 保证高并发下计数的线程安全避免锁竞争。LeapArray核心数据结构负责窗口的复用和过期清理确保内存占用稳定。优势相比固定窗口滑动窗口解决了“临界点突变”问题统计更平滑准确。Q3: 如何实现分布式限流Sentinel 支持吗默认单机Sentinel 默认是单机限流每个实例独立统计。集群模式Sentinel 支持集群限流但需要部署独立的 Token Server令牌服务器。客户端向 Token Server 申请令牌Token Server 全局控制配额。替代方案如果不想部署 Token Server可使用 Redis Lua如 Redisson RateLimiter实现全局限流或在网关层Spring Cloud Gateway做全局限流。限流后用户体验很差怎么优化快速失败太生硬不要直接返回 500 或空白页。友好提示通过 blockHandler 返回统一的 JSON 格式错误码如 code: 429, msg: 系统繁忙请稍后重试。排队等待对于秒杀等场景使用“匀速排队”模式让用户等待几秒后成功而不是直接拒绝。前端优化前端按钮置灰、增加 loading 动画防止用户重复点击。Spring Cloud Gateway--,,,,--RabbitMQ核心概念Producer / Consumer生产者与消费者模型Queue消息队列存储消息Exchange交换机决定消息如何路由到队列Binding队列与交换机的绑定关系Routing Key / Binding Key消息路由规则消息确认机制自动 / 手动 ACK消息持久化防止消息丢失MQ 的常见问题有消息的顺序问题消息的重复问题消息的可靠传输消息的顺序问题消息的重复问题消息的可靠传输消息不可靠的情况可能是消息丢失劫持等原因丢失又分为生产者丢失消息、消息列表丢失消息、消费者丢失消息生产者丢失消息从生产者弄丢数据这个角度来看RabbitMQ提供transaction和confirm模式来确保生产者不丢消息transaction机制就是说发送消息前开启事务channel.txSelect()然后发送消息如果发送过程中出现什么异常事务就会回滚channel.txRollback()如果发送成功则提交事务channel.txCommit()。然而这种方式有个缺点吞吐量下降confirm模式用的居多一旦channel进入confirm模式所有在该信道上发布的消息都将会被指派一个唯一的ID从1开始一旦消息被投递到所有匹配的队列之后rabbitMQ就会发送一个ACK给生产者包含消息的唯一ID这就使得生产者知道消息已经正确到达目的队列了如果rabbitMQ没能处理该消息则会发送一个Nack消息给你生产者可以进行重试操作。发送方确认机制先说配置和使用配置文件spring: rabbitmq: publisher-confirm-type: correlated # 开启发送方确认机制配置属性有三种分别为none表示禁用发送方确认机制correlated表示开启发送方确认机制simple表示开启发送方确认机制并支持waitForConfirms()和waitForConfirmsOrDie()的调用。这里一般使用correlated开启发送方确认机制即可至于simple的waitForConfirms()方法调用是指串行确认方法即生产者发送消息后调用该方法等待 RabbitMQ Server 确认如果返回 false 或超时未返回则进行消息重传。由于串行性能较差这里一般都是用异步 confirm 模式。通过调用setConfirmCallback()实现异步 confirm 模式感知消息发送结果/** * 消息业务实现类 * * author 单程车票 */ Service public class RabbitMQServiceImpl { Autowired private RabbitTemplate rabbitTemplate; Override public void sendMessage() { // 发送消息 rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); // 设置消息确认回调方法 rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { /** * MQ确认回调方法 * param correlationData 消息的唯一标识 * param ack 消息是否成功收到 * param cause 失败原因 */ Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { // 记录日志 log.info(ConfirmCallback...correlationData[correlationData]ack:[ack]cause:[cause]); if (!ack) { // 出错处理 ... } } }); } }3. 保证消息在 RabbitMQ Server 中的持久化对于消息的持久化只需要在发送消息时将消息持久化并且在创建交换机和队列时也保证持久化即可。配置如下/** * 消息队列 */ Bean public Queue queue() { // 四个参数name队列名、durable持久化、exclusive独占、autoDelete自动删除 return new Queue(MESSAGE_QUEUE, true); } /** * 直接交换机 */ Bean public DirectExchange exchange() { // 四个参数name交换机名、durable持久化、autoDelete自动删除、arguments额外参数 return new DirectExchange(Direct_Exchange, true, false); }在创建交换机和队列时通过构造方法将持久化的参数都设置为 true 即可实现交换机和队列的持久化。Override public void sendMessage() { // 构造消息将消息持久化 Message message MessageBuilder.withBody(单程车票.getBytes(StandardCharsets.UTF_8)).setDeliveryMode(MessageDeliveryMode.PERSISTENT).build(); // 向MQ发送消息消息内容都为消息表记录的id rabbitTemplate.convertAndSend(RabbitMQConfig.Direct_Exchange, routingKey, message); }在发送消息前通过调用MessageBuilder的setDeliveryMode(MessageDeliveryMode.PERSISTENT)在构造消息时设置消息持久化MessageDeliveryMode.PERSISTENT即可实现对消息的持久化。生产者发送消息后通过调用setConfirmCallback()可以将信道设置为 confirm 模式所有消息会被指派一个消息唯一标识当消息被发送到 RabbitMQ Server 后Server 确认消息后生产者会回调设置的方法从而实现生产者可以感知到消息是否正确无误的投递从而实现发送方确认机制。并且该模式是异步的发送消息的吞吐量会得到很大提升。上面就是发送放确认机制的配置和使用使用这种机制可以保证生产者的消息可靠性投递并且性能较好。消息队列丢数据消息持久化。处理消息队列丢数据的情况一般是开启持久化磁盘的配置。这个持久化配置可以和confirm机制配合使用你可以在消息持久化磁盘后再给生产者发送一个Ack信号。这样如果消息持久化磁盘之前rabbitMQ阵亡了那么生产者收不到Ack信号生产者会自动重发。那么如何持久化呢这里顺便说一下吧其实也很容易就下面两步将queue的持久化标识durable设置为true则代表是一个持久的队列发送消息的时候将deliveryMode2这样设置以后即使rabbitMQ挂了重启后也能恢复数据消费者丢失消息消费者丢数据一般是因为采用了自动确认消息模式改为手动确认消息即可消费者在收到消息之后处理消息之前会自动回复RabbitMQ已收到消息如果这时处理消息失败就会丢失该消息解决方案处理消息成功后手动回复确认消息。4. 保证消费者消费的消息不丢失在保证发送方和 RabbitMQ Server 的消息可靠性的前提下只需要保证消费者在消费消息时异常消息不丢失即可保证消息的可靠性。RabbitMQ 提供了消费者应答机制来使 RabbitMQ 能够感知到消费者是否消费成功消息默认情况下消费者应答机制是自动应答的也就是RabbitMQ 将消息推送给消费者便会从队列删除该消息如果消费者在消费过程失败时消息就存在丢失的情况。所以需要将消费者应答机制设置为手动应答只有消费者确认消费成功后才会删除消息从而避免消息的丢失。下面来看看如何配置消费者手动应答spring: rabbitmq: publisher-confirm-type: correlated # 开启发送方确认机制 publisher-returns: true # 开启消息返回 template: mandatory: true # 消息投递失败返回客户端 listener: simple: acknowledge-mode: manual # 开启手动确认消费机制通过listener.simple.acknowledge-mode manual即可将消费者应答机制设置为手动应答。之后只需要在消费消息时通过调用channel.basicAck()与channel.basicNack()来根据业务的执行成功选择是手动确认消费还是手动丢弃消息。/** * 监听消费队列的消息 */ RabbitListener(queues RabbitMQConfig.MESSAGE_QUEUE) public void onMessage(Message message, Channel channel) { // 获取消息索引 long index message.getMessageProperties().getDeliveryTag(); // 解析消息 byte[] body message.getBody(); ... try { // 业务处理 ... // 业务执行成功则手动确认 channel.basicAck(index, false); }catch (Exception e) { // 记录日志 log.info(出现异常{}, e.getMessage()); try { // 手动丢弃信息 channel.basicNack(index, false, false); } catch (IOException ex) { log.info(丢弃消息异常); } } }这里说明一下basicAck()与basicNack()的参数说明:void basicAck(long deliveryTag, boolean multiple)方法会抛异常deliveryTag该消息的indexmultiple是否批量处理true 表示将一次性ack所有小于deliveryTag的消息void basicNack(long deliveryTag, boolean multiple, boolean requeue)方法会抛异常deliveryTag该消息的indexmultiple是否批量处理true 表示将一次性ack所有小于deliveryTag的消息requeue被拒绝的是否重新入队列true 表示添加在队列的末端false 表示丢弃通过设置手动确认消费者应答机制即可保证消费者在消费信息时的消息可靠性。
