JAVA面试之MQ

文章目录RabbitMQ在什么场景下使用有什么好处如何保证消息的可靠传输如果消息丢了怎么办生产者丢失MQ中丢失消费者丢失兜底方案失败重试机制业务幂等性如何在 RabbitMQ中实现消息幂等性?如何保证消息的顺序性RabbitMQ的交换机有哪些什么是死信队列死信队列是如何导致的什么是死信队列死信队列是如何导致的RabbitMQ怎么实现延迟队列?ActiveMQ、RabbitMQ、RocketMQ、Kafka对比1.消息模型与路由2. 事务消息与分布式事务RocketMQ 的事务消息原理是分布式理论的绝佳实践3.顺序消息4.延时与死信队列选型指南面试总结Kafka 的“百万级吞吐”到底强在哪拉模式 vs 推模式一个你能够主动引导的“选型反例”RabbitMQ在什么场景下使用有什么好处1系统解耦使用RabbitMQ作为消息的中间传递者,不同的系统之间通过发送和接收消息来进行通信而不需要直接依赖于彼此。2异步处理RabbitMQ可以用于异步处理任务。生产者将任务消息发送到RabbitMQ然后消费者从队列中获取消息并处理。3负载均衡当有多个消费者订阅同一个队列时RabbitMQ会将消息均匀地发送给每个消费者从而实现负载均衡。4优先级队列RabbitMQ可以为不同的消息设置不同的优先级。5发布/订阅模式RabbitMQ支持发布/订阅模式可以同时将消息发送给多个消费者。生产者发布消息到交换机上多个消费者可以订阅该交换机的队列来接收消息。6持久化和可靠性传输消息可以被持久化到磁盘上即使RabbitMQ服务器出现故障消息也不会丢失。如何保证消息的可靠传输如果消息丢了怎么办数据的丢失问题可能出现在生产者、MQ、消费者中。1生产者发送消息时丢失①生产者发送消息时连接MQ失败②生产者发送消息到达MQ后未找到Exchange(交换机)③生产者发送消息到达MQ的Exchange后未找到合适的Queue(队列)④消息到达MQ后处理消息的进程发生异常2MQ导致消息丢失消息到达MQ保存到队列后尚未消费就突然宕机3消费者处理消息时①消息接收后尚未处理突然宕机②消息接收后处理过程中抛出异常生产者丢失1可以选择开启 RabbitMQ 提供的事务功能如果消息没有成功被RabbitMQ 接收到那么生产者会收到异常报错此时就可以回滚事务channel.txRollback然后重新发送消息如果收到了消息就可以提交事务channel.txCommit()。但是这样会导致吞吐量下来因为太耗性能。2另一种方案就是a.对于生产者发送消息时连接MQ失败可以开启生产者超时重试机制b.对于其他情况比如生产者发送消息到达MQ后未找到Exchange、生产者发送消息到达MQ的Exchange后未找到合适的Queue、消息到达MQ后处理消息的进程发生异常。我们可以使用RabbitMQ提供的Publisher Confirm和Publisher Return两种确认机制。在开启确认机制后MQ返回给生产者的结果有以下几种情况①当消息投递到MQ但是路由失败时通过Publisher Return返回异常原因同时返回ack的确认信息代表投递成功②临时消息投递到了MQ并且入队成功返回ACK告知投递成功③持久消息投递到了MQ并且入队完成持久化返回ACK 告知投递成功④其它情况都会返回NACK告知投递失败其中ack和nack属于Publisher Confirm机制ack是投递成功nack是投递失败。而return则属于Publisher Return机制。默认两种机制都是关闭状态需要通过配置文件来开启。3事务机制和cnofirm机制最大的不同在于事务机制是同步的你提交一个事务之后会阻塞在那儿但是confirm机制是异步的你发送个消息之后就可以发送下一个消息然后那个消息RabbitMQ 接收了之后会异步回调你一个接口通知你这个消息接收到了。所以一般在生产者这块避免数据丢失都是用confirm机制的。4补充(面试问了再回答)publisher-confirm-type有三种模式可选①none关闭confirm机制②simple同步阻塞等待MQ的回执③correlatedMQ异步回调返回回执一般我们推荐使用correlated回调机制。MQ中丢失为了提升性能默认情况下MQ的数据都是在内存存储的临时数据重启后就会消失。为了保证数据的可靠性必须配置数据持久化包括交换机持久化、队列持久化、消息持久化消费者丢失1消费者确认机制为了确认消费者是否成功处理消息RabbitMQ提供了消费者确认机制Consumer Acknowledgement。即当消费者处理完消息后应该向RabbitMQ发送一个回执告知RabbitMQ自己消息处理状态。回执有三种可选值ack成功处理消息RabbitMQ从队列中删除该消息nack消息处理失败RabbitMQ需要再次投递消息reject消息处理失败并拒绝该消息(比如消息格式有问题)RabbitMQ从队列中删除该消息2补充由于消息回执的处理代码比较统一因此SpringAMQP帮我们实现了消息确认。并允许我们通过配置文件设置ACK处理方式有三种模式none不处理。即消息投递给消费者后立刻ack消息会立刻从MQ删除。非常不安全不建议使用manual手动模式。需要自己在业务代码中调用api发送ack或reject存在业务入侵但更灵活auto自动模式。SpringAMQP利用AOP对我们的消息处理逻辑做了环绕增强当业务正常执行时则自动返回ack. 当业务出现异常时根据异常判断返回不同结果a.如果是业务异常会自动返回nackb.如果是消息处理或校验异常自动返回reject;我们选择auto自动模式即可兜底方案1虽然我们利用各种机制尽可能增加了消息的可靠性但也不好说能保证消息100%的可靠。万一真的MQ通知失败该怎么办呢有没有其它兜底方案能够确保订单的支付状态一致呢2其实思想很简单既然MQ通知不一定发送到交易服务那么交易服务就必须自己主动去查询支付状态。这样即便支付服务的MQ通知失败我们依然能通过主动查询来保证订单状态的一致失败重试机制1当消费者出现异常后消息会不断requeue重新入队到队列再重新发送给消费者然后再次异常再次requeue到队列无限循环导致mq的消息处理飙升带来不必要的压力2为了应对上述情况Spring又提供了消费者失败重试机制在消费者出现异常时利用本地重试而不是无限制的requeue到mq队列。3开启本地重试时消息处理过程中抛出异常不会requeue到队列而是在消费者本地重试重试达到最大次数后Spring会返回reject消息会被丢弃4Spring允许我们自定义重试次数耗尽后的消息处理策略这个策略是由MessageRecovery接口来定义的它有3个不同实现①RejectAndDontRequeueRecoverer重试耗尽后直接reject丢弃消息。默认就是这种方式②ImmediateRequeueMessageRecoverer重试耗尽后返回nack消息重新入队③RepublishMessageRecoverer重试耗尽后将失败消息投递到指定的交换机比较优雅的一种处理方案是RepublishMessageRecoverer失败后将消息投递到一个指定的专门存放异常消息的队列后续由人工集中处理。业务幂等性幂等是一个数学概念用函数表达式来描述是这样的f(x) f(f(x))例如求绝对值函数。在程序开发中则是指同一个业务执行一次或多次对业务状态的影响是一致的。方案一:给每个消息都设置一个唯一id利用id区分是否重复消息①每一条消息都生成一个唯一的id与消息一起投递给消费者②消费者接收到消息后处理自己的业务业务处理成功后将消息ID保存到数据库③如果下次又收到相同消息去数据库查询判断是否存在存在则为重复消息放弃处理④我们该如何给消息添加唯一ID呢其实很简单SpringAMQP的MessageConverter自带了MessageID的功能我们只要开启这个功能即可方案二业务判断业务判断就是基于业务本身的逻辑或状态来判断是否是重复的请求或消息不同的业务场景判断的思路也不一样。例如我们当前案例中处理消息的业务逻辑是把订单状态从未支付修改为已支付。因此我们就可以在执行业务时判断订单状态是否是未支付如果不是则证明订单已经被处理过无需重复处理。相比较而言消息ID的方案需要改造原有的数据库所以我更推荐使用业务判断的方案。如何在 RabbitMQ中实现消息幂等性?如何保证消息的顺序性1先看看顺序会错乱的场景RabbitMQ一个 queue多个 consumer这不明显乱了2解决RabbitMQ的交换机有哪些1Direct Exchange直连交换机根据Routing Key(路由键)进行投递到不同队列。2Fanout Exchange扇形交换机采用广播模式根据绑定的交换机路由到与之对应的所有队列。3Topic Exchange主题交换机对路由键进行模式匹配后进行投递符号#表示一个或多个词*表示一个词。4Header Exchange头交换机不处理路由键。而是根据发送的消息内容中的headers属性进行匹配。什么是死信队列死信队列是如何导致的什么是死信队列1死信队列本身也是一个普通的消息队列可以通过设置一些参数将其设置为死信队列2无法被消费的消息称为死信当一个消息在队列中变成死信之后它能被重新发送到另一个交换机中即死信交换机绑定到死信交换机的队列就称为死信队列。死信队列是如何导致的1消息过期当一个消息过期后它就会被发送到死信队列。这通常是由于消息的TTLTime To Live过期导致的。2消息被拒绝当一个消费者拒绝处理某个消息时这个消息就会被发送到死信队列。这通常是由于消息格式不正确等原因导致的。3队列满了当一个队列已经满了新的消息就无法进入该队列。这时新的消息就会被发送到死信队列。RabbitMQ怎么实现延迟队列?ActiveMQ、RabbitMQ、RocketMQ、Kafka对比对比维度ActiveMQRabbitMQRocketMQKafka开发语言JavaErlangJavaScala/Java单机吞吐量万级万级几万十万级百万级时效性毫秒级微秒级毫秒级毫秒级消息可靠性高 (Master/Slave)高 (Mirror Queue)极高(同步刷盘/主从)极高 (ISR复制)功能完备性较完备极强(Exchange路由)强 (事务/延迟/死信)偏弱 (聚焦流处理)社区生态老牌稳定庞大插件丰富阿里主导国内活跃极强大数据生态核心运维难度低低到中中 (组件多)中到高 (对硬件有要求)管理界面一般优秀(Web管理端)较好 (Dashboard)一般 (需外部工具)1.消息模型与路由ActiveMQ/RabbitMQ都是“智能代理”模型支持复杂的路由规则。RabbitMQ 的 Exchange 能实现几乎任意路由组合。RocketMQ/Kafka都是“傻代理”模型路由规则简单只发布到 Topic。区别在于RocketMQ 支持 Tag 和 SQL92 过滤在 Topic 之下再做一层子分类比 Kafka 灵活。备注1.“所谓任意路由组合是指 RabbitMQ 提供了 Direct、Fanout、Topic、Headers 四种交换机并且支持把多个交换机和队列通过不同的绑定键Binding Key串联形成复杂的拓扑。2.Topic 就是消息的主题或类别。生产者把消息发到某个 Topic消费者订阅这个 Topic 来接收消息。类比就像你订阅的微信公众号。公众号是 Topic发布文章就是发消息你关注它就能收到推送。RabbitMQ 的 Exchange 在消息投递入口这个角色上相当于 Kafka/RocketMQ 的 Topic。RabbitMQ 的 Exchange 只负责路由逻辑不存消息真正存储消息的是 Queue这和 Kafka 的 Topic 既是逻辑分类又是物理存储单元有本质区别。3.Tag标签消息的第二级分类标识。一个 Topic 下可以有多个 Tag生产者在发消息时指定。类比Topic 是邮件”Tag 是邮件标题。同一个 Topic 的所有消息都在一个快递车里但消费者可以根据标题决定是否处理。4… SQL92 过滤在 Tag 的基础上支持用 SQL 风格的语法对消息的用户属性Property做条件过滤。2. 事务消息与分布式事务RocketMQ 是唯一商业级支持事务消息的。它的“半消息”机制可以完美解决分布式事务的最终一致性问题。ActiveMQ/RabbitMQ基于 AMQP/JMS 标准有本地事务支持但分布式事务场景需要自研本地消息表等方案。Kafka后来加入事务但主要用于流处理内部的精确一次语义。RocketMQ 的事务消息原理是分布式理论的绝佳实践你提到了它是唯一商业级支持可以简短用几句话概括它的“半消息”机制体现你深入理解了RocketMQ 的分布式事务核心是事务反查机制。生产者先发一条对消费者不可见的“半消息”。然后执行本地事务根据结果向 Broker 提交Commit或回滚Rollback。关键是如果 Broker 因网络波动长时间没收到确认它会主动回调生产者接口反查本地事务的最终状态。这个反查机制巧妙解决了网络不确定性带来的悬挂问题。3.顺序消息1所有消息队列都保证分区队列内严格有序。2RocketMQ 在这方面做得最完善通过 MessageQueueSelector 可将业务标识相同的消息如同一订单ID发送到同一队列轻松实现全局部分有序。RocketMQ 的“完美”是有代价的它通过同一队列Queue串行投递来保证严格顺序。但在高并发下如果某个订单的消息处理阻塞会卡住整个队列导致其他订单也无法消费。这需要配合非阻塞消费逻辑和熔断降级。3Kafka也是把消息按 Key 哈希到分区保证顺序但扩容分区会破坏顺序这是架构硬伤。Kafka 的“硬伤”其实可以解你可以补充说Kafka 的顺序保证单位是分区扩容确实会破坏 Key 的分区映射。所以在数据架构设计时通常会对 Key 做预分区如初始就创建 64 或 128 个分区短期靠增大单分区吞吐量顶住长期的数据再平衡则需要上游配合重置 Offset成本很高。这体现的是架构师的预见性。4.延时与死信队列RabbitMQ通过插件和 TTLDLX 组合实现。RocketMQ原生支持定时/延时消息商业版支持任意延时且内置死信队列业务属性最强。选型指南面试总结ActiveMQ慎用避坑。适合维护遗留系统新项目别选。RabbitMQ企业应用与微服务解耦。适合对路由要求高、需要毫秒级/微秒级可靠投递的场景如订单通知、审批流并且开发运维友好。RocketMQ高并发核心在线业务。如阿里的电商、金融场景尤其当你有分布式事务和全链路顺序等刚性需求时。Kafka大数据和流处理管道。海量日志采集、用户行为埋点、实时计算的数据源是Hadoop/Flink生态的事实标准。Kafka 的“百万级吞吐”到底强在哪你说到了吞吐量数据如果再解释一下底层原理会非常加分顺序 I/OKafka 将消息以追加方式写入分区日志文件把磁盘的随机写变成了顺序写极致压榨了机械硬盘甚至 SATA SSD 的性能。零拷贝Zero-CopyKafka 消费消息时通过sendfile系统调用数据从磁盘缓存直接传输到网卡缓冲区绕过了用户态内核态的多次数据拷贝大幅降低 CPU 使用率。这是它比 RocketMQ基于 Netty 有多次内存拷贝吞吐量高一个量级的核心原因。拉模式 vs 推模式你可以点出它们在消费模型上的本质区别RabbitMQ 是纯“推”模式Broker 主动推送优点是实时性极高缺点是当消费者处理慢时容易产生消息堆积导致消费者内存溢出。可以通过设置prefetch做流量控制。RocketMQ 和 Kafka 是“拉”模式消费者根据自身处理能力主动拉取天然具备消费削峰能力更适配高吞吐场景。RocketMQ 为平衡实时性做了“长轮询”不至于空转。一个你能够主动引导的“选型反例”如果面试官让你说一个反常识的案例这会让你脱颖而出“提到 Kafka 不是只等于大数据。在现代事件驱动架构EDA和DDD领域驱动设计微服务中Kafka 常被用作事件中心利用其持久化和重播能力实现事件溯源Event Sourcing。这个场景下Kafka 不是管道而是事实来源Source of Truth。”