1、什么是消息队列？

Message Queue 简称 MQ，是一种应用间的通信方式，由生产者Producer、代理Broker、消费者Consumer三者组成。

场景：

1. 异步，实时性要求不严格的场景，注册发送验证码、下单通知、发送优惠券等。不需要等待消息服务返回结果。
2. 应用解耦，将相关但耦合度不高的系统联系起来。解决了各个系统可以采用不同的架构、语言来实现，大大增加系统的灵活性。
3. 流量削峰，应用在大流量入口且短时间内业务需求处理不完的服务中心，为权衡高可用，将大量并行业务发送到MQ，起到大流量缓冲的作用。

类型：ActiveMQ、RabbitMQ、Kafka、RocketMQ

中小型公司、低吞吐量一般用ActiveMQ、RabbitMQ。

大数据高吞吐量的大型公司一般选用用Kafka和RocketMQ。

2、消息队列有哪些使用场景？（为什么使用消息队列）

2.1 应用解耦

1. 将相关但耦合度不高的系统联系起来。解决了各个系统可以采用不同的架构、语言来实现，大大增加系统的灵活性。
2. 扩充下游系统时不需要做大的调整。

2.2 流量削峰

场景特点：应用在大流量入口且短时间内业务需求处理不完的服务中心

比如做秒杀，需要避免流量暴涨打垮应用的风险。假设系统每秒最多处理2k个请求，但实际每秒有5k个请求，此时引入消息队列，每秒从队列中拉取2k个请求处理就可以了。

消息积压：

• 秒杀活动不可能每时每刻大请求量，高峰期过去，积压请求可以慢慢处理。
• 队列长度超过最大数量，可以直接抛弃用户请求，走兜底业务

2.3 异步处理

将按顺序执行的业务，同步执行，大大减少响应速度。

2.4 消息通讯

内置了高效的通讯机制，可以实现点对点消息队列、聊天室等。

2.5 远程调用

3、消息队列如何解决消息丢失问题？

生产者产生消息-》Broker代理存储消息-》消费者消费消息

3.1 生产者保证消息不丢失

如果使用RocketMQ消息中间件，生产者提供了三种发送方式：同步、异步、单向

• 采用同步发送，send消息返回成功状态，则表示成功存储
• send消息异常或返回非成功状态，可以重试
• 可以使用事务消息，RocketMQ事务消息机制就是为了保证零丢失设计的

3.2 存储段不丢消息

刷盘机制：

生产者发送消息，只有持久化到磁盘，RocketMQ存储端才会返回一个成功ACK响应。但影响性能。

异步刷盘：

只要消息写入PageCache缓存，就返回一个成功ACK响应。提高了性能，但一单及其断电，就会丢失消息。

Broker一般集群部署，有master主节点和slave从节点。

同步复制：主节点和从节点都收到消息，才返回成功ACK，保证了消息不丢失，但降低性能。

异步复制：只要消息写入主节点，就返回成功ACK，速度快，但会有丢失问题。

3.3 消费者不丢消息

消费者执行完业务逻辑，在反馈给Broker消费成功。

4、消息队列如何保证消息的顺序性？

4.1 单机：

发送至同一个服务的MQ上，发送至同一个服务的消费者，且等到M1消费端ACK成功后，M2再发送。

4.2 集群：

生产者有序存储、消费者有序消费。

4.2.1 生产者有序消费

普通发送消息的模式下，生产者采用轮询的方式均匀发送至不同队列中，被不同的消费者消费。此时无法使用队列有序特性保证有序性。

解决方案：投放消息支持自定义投放策略，顺序消息必须使用同步发送的方式，才能保证发送有序。

实现一个**MessageQueueSelector接口**，使用Hash取模法来保证同一个订单在同一个队列中就行了

即通过订单ID%队列数量得到该ID的订单所投放的队列在队列列表中的索引，然后该订单的所有消息都会被投放到这个队列中。

4.2.2 消费者有序消费

RockerMQ的MessageListener回调函数提供了两种消费模式，有序消费模式MessageListenerOrderly和并发消费模式MessageListenerConcurrently。

在消费的时候，还需要保证消费者注册MessageListenerOrderly类型的回调接口实现顺序消费，如果消费者采用Concurrently并行消费，则仍然不能保证消息消费顺序。

实际上，每一个消费者的的消费端都是采用线程池实现多线程消费的模式，即消费端是多线程消费。虽然MessageListenerOrderly被称为有序消费模式，但是仍然是使用的线程池去消费消息。

MessageListenerConcurrently是拉取到新消息之后就提交到线程池去消费，而MessageListenerOrderly则是通过加分布式锁和本地锁保证同时只有一条线程去消费一个队列上的数据。

messageQueue的本地synchronized锁：

1. 执行消费任务的开头获取本地锁对象objLock，通过synchronized实现锁定。
2. 锁对象存储在MessageQueueLock.mqLockTable属性中，结构为ConcurrentMap<MessageQueue, Object>，一个messageQueue对应一个锁。
3. 这个锁保证同一时刻对于同一个队列只有一个线程去消费他。

ProcessQueue的本地consumeLock：

执行真正的消费之前，会获取ProcessQueue的本地consumeLock，这个本地锁是一个ReentrantLock。

作用：防止在消费过程中，该消息队列因负载均衡而被分配给其他客户端，导致两个客户端重复消费。

5、如何避免消息重复消费？

生产端为保证消息可靠性，可能想MQ重复发送消息，直到拿到成功的ACK。

消费端流程：拉取消息-》业务逻辑执行-》提交消费位移。

假设更新位移时，消费者挂了，这时另一个消费者就会拉取到重复的消息。

解决方案：

5.1 提高消费端的处理性能，避免触发Balance

• 多线程处理消息，缩短单个消息消费时长。
• 调整消息处理的超时时间。
• 减少一次性从Broker上拉取数据的条数。

5.2 使用ConsumerRebalanceListener，再均衡监听器

设定发生再均衡动作前后的一些准备工作和收尾工作。

5.3 使用消息幂等性

• 开启KafKa的幂等性功能
• 将消息生成md5，带唯一业务标记，保存到Mysql或Redis中，在处理消息之前先查Mysql或Redis，进行判断是否已消费。

6、如何解决幂等性问题？

6.1 概念

一个方法无论被多少次重复执行，所期望的结果和第一次执行所期望的结果保持一致。

6.2 问题

• 用户重复提交、恶意攻击
• 分布式系统中，为避免数据丢失，采用的超时重试机制

6.3 如何解决？

本质：保证接口的执行结果只影响一次，后续再次调用，不能对数据产生影响。

1. 使用数据库唯一约束实现。
2. 使用Redis提供的setNX指令
3. 使用状态机实现，即一条数据的完整运行状态的转换流程，状态只会向前变更

7、如何处理消息队列消息积压问题？

起因：消费速度原小于生产速度

临时解决：紧急扩容，先保证消息都消费完。

1. 修复consumer消费者问题，确保恢复消费速度，然后将现有consumer都停掉。
2. 新建一个topic，partition，临时建立好原先10倍的queue数量。
3. 写一个临时分发数据的consumer程序，部署上去消费积压的数据，消费之后不做耗时处理，直接轮询写入临时创建好的10倍数量的queue中。
4. 临时征用10倍数量的机器部署consumer，每一批consumer消费一个临时queue数量。相当于临时将queue资源和consumer资源扩大10倍，10倍速度消费。
5. 快速消费积压数据后，恢复原先部署的程序。

8、MQ技术选型

指标	Kafka	RocketMQ	RabbitMQ
单机吞吐量	17.3w/s	11.6w/s	2.6w/s(消息做持久化)
开发语言	Scala/Java	Java	Erlang
维护	Apache	Alibaba	Spring
订阅形式	基于topic，进行正则匹配	基于topic、messageTag，根据类型和属性正则匹配	支持direct、topic、Headers、fanout四种模式
持久化	支持大量堆积	支持大量堆积	支持少量堆积
顺序消息	支持	支持	不支持
集群方式	天然Leader-Slave，无状态集群，每台服务器既是Master也是Slave	常用“多对Master-Slave”，开源版需要手动切换从节点变主节点	支持简单集群，“复制模式”，对高级集群支持不好
性能稳定性	较差	一般	好

• 大数据领域实时计算、日志采集等，用Kafka为业内标准
• RocketMQ，阿里出品
• RabbitMQ开源，稳定支持、活跃度高，但不是Java开发

9、如何保证数据一致性，事务消息如何实现？

1. 生产者发送消息到MQ服务器
2. MQ收到消息后，持久化到存储系统，状态为代发送
3. MQ服务器返回ACK确认到生产者，此时还不触发推送事件
4. 生产者执行本地事务
5. 本地事务执行成功，提交结果到MQ服务器，执行失败发送rollback
6. 正常提交，MQ更新消息状态为可送达，rollback回滚则删除消息
7. 如果消息更新为可送达，则MQ服务器push消息给消费者，消费完成就回ACK。
8. 如果MQ服务器长时间没有收到生产者的commit或rollback，会反查生产者，根据查询结果执行最终状态

10、RabbitMQ的消息可靠传输如何保证？

丢失的三种情况：

• 生产者发送消息到MQ 服务的过程中丢失
• MQ服务收到消息后，在持久化之前宕机，导致数据丢失
• 消费端收到消息还未处理

对于生产者：生产者发送之后，MQ提供了一个消息确认机制，客户端根据消息的处理结果决定是否需要重新发送。

MQ 服务端：开启消息持久化机制，存入磁盘。

• 创建Queue设置持久化
• 发送消息的时候，将消息投递模式设置为持久化投递。

消费端：将消息的自动确认机制修改为手动确认，只会手动用消息确认方法才表示消息已被签收。

11、RabbitMQ的消息如何实现路由？

本质：一个基于AMQP协议实现的分布式消息中间件。

AMQP工作机制：

• 生产者将消息发送到RabbitMQ Broker上的Exchange交换机上。
• Exchange交换机把收到的消息根据路由规则发给绑定的Queue队列。
• 再把消息投递给订阅了这个队列的消费者，完成消息的异步通讯。

Exchange ：消息交换机，定义了消息路由的规则，规定消息与队列的关系。

Queue：消息的载体，每个消息可以根据路由规则到一个或多个队列中。

RoutingKey：路由键，发送消息时声明，Exchange拿到路由键，根据它和路由表里面的BindingKey进行匹配。

在RabbitMQ中，有三种类型的Exchange：direct ，fanout和topic。

direct：完整匹配，是Routing key和Binding Key完全一致，点对点发送。

fanout：广播机制，不急于路由键匹配，将消息广播给绑定到当前交换机上的所有队列。

topic：正则表达式匹配，根据Routing key使用正则表达式进行匹配，发送消息给符合规则的Queue。