消息中间件如何选型

RabbitMQ

erlang开发，对消息堆积的支持并不好，当大量消息积压的时候，会导致 RabbitMQ 的性能急剧下降。每秒钟可以处理几万到十几万条消息。

RocketMQ

java开发，面向互联网集群化功能丰富，对在线业务的响应时延做了很多的优化，大多数情况下可以做到毫秒级的响应，每秒钟大概能处理几十万条消息。

Kafka

Scala开发，面向日志功能丰富，性能最高。当你的业务场景中，每秒钟消息数量没有那么多的时候，Kafka 的时延反而会比较高。所以，Kafka 不太适合在线业务场景。 ActiveMQ java开发，简单，稳定，性能不如前面三个。小型系统用也ok，但是不推荐。推荐用互联网主流的。

消息中间件的作用

因为项目比较大，做了分布式系统，所有远程服务调用请求都是同步执行经常出问题，所以引入了mq

RocketMQ角色

主要是发布订阅模式，主要组件：消息发送者、消息服务器（消息存储）、消息消费、。

NameServer

NameServer 是整个 RocketMQ 的“大脑”，它是 RocketMQ 的服务注册中心,所以 RocketMQ需要先启动 NameServer 再启动 Rocket 中的 Broker。

Broker 在启动时向所有 NameServer 注册（主要是服务器地址等），生产者在发送消息之前先从 NameServer 获取 Broker 服务器地址列表（消费者一 样），然后根据负载均衡算法从列表中选择一台服务器进行消息发送。

NameServer与每台 Broker 服务保持长连接， NameServer每间隔 30S检查 Broker 是否存活，如果检测到 Broker 宕机，则从路由注册表中将其移除。这样就可以实现RocketMQ的高可用。

Broker和NameServer的关系

Broker会向所有的NameServer上注册自己的信息，而不是某一个，是每一个，全部！

RocketMQ中的Topic和queue区别？

queue就是来源于数据结构的FIFO队列。而Topic是个抽象的概念，每个Topic底层对应N个queue，而数据也真实存在queue上的。

消息分类Topic和Tag

消息消费是，消费者可以对同一主题下的消息按照规则只消费自己感兴趣的消息，可以支持在服务端与消费端的消息过滤机制。

消息标签，二级消息类型，用来进一步区分某个 Topic 下的消息分类。 Topic 与 Tag 都是业务上用来归类的标识，区分在于 Topic 是一级分类，而 Tag 可以理解为是二级分类。 以天猫交易平台为例，订单消息和支付消息属于不同业务类型的消息，分别创建 Topic_Order 和 Topic_Pay，其中订单消息根据商品品类以不同的 Tag 再进行细分，如电器类、男装类、女装类、化妆品类，最后他们都被各个不同的系统所接收。 通过合理的使用 Topic 和 Tag，可以让业务结构清晰，更可以提高效率。

您可能会有这样的疑问：到底什么时候该用 Topic，什么时候该用 Tag？

1）消息类型是否一致：如普通消息、事务消息、定时（延时）消息、顺序消息，不同的消息类型使用不同的 Topic，无法通过 Tag 进行区分。

2）业务是否相关联：没有直接关联的消息，如淘宝交易消息，京东物流消息使用不同的 Topic 进行区分；而同样是天猫交易消息，电器类订单、女 装类订单、化妆品类订单的消息可以用 Tag 进行区分。

3）消息优先级是否一致：如同样是物流消息，盒马必须小时内送达，天猫超市 24 小时内送达，淘宝物流则相对会慢一些，不同优先级的消息用不同的 Topic 进行区分。

4）消息量级是否相当：有些业务消息虽然量小但是实时性要求高，如果跟某些万亿量级的消息使用同一个 Topic，则有可能会因为过长的等待时间 而“饿死”，此时需要将不同量级的消息进行拆分，使用不同的 Topic。

消息存储

一般 MQ 核心就是消息的存储，对存储一般来说两个维度:消息堆积能力和消息存储性能。RocketMQ 追求消息存储的高性能，引入内存映射机制，所有的主题消息顺序存储在同一个文件中。同时为了防止无限堆积，引入消息文件过期机制和文件存储空间报警机制。

消息高可用

Rocket 关机、断电等情况下，Rokcet 可以确保不丢失消息（同步刷盘机制不丢失，异步刷盘会丢失少量）。

另外如果 Rocket 服务器因为 CPU、内存、主板、磁盘等关键设备损坏导致无法开机，这个属于单点故障，该节点上的消息全部丢失，如果开启了 异步复制机制，Rocket 可以确保只丢失很少量消息。

如果引入双写机制，这样基本上可以满足消息可靠性要求极高的场景（毕竟两台主服务器同时故障的可能性还是非常小）

消息消费低延迟

RocketMQ在消息不发生消息堆积时，以长轮询模式实现准实时的消息推送模式。

确保消息必须被消费一次

消息确认机制(ACK)来确保消息至少被消费一次，一般 ACK 机制只能做到消息只被消费一次，有重复消费的可能。

消息回溯

已经消费完的消息，可以根据业务要求重新消费消息。

消息堆积

消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的 消息堆积能力，RocketMQ 采用磁盘文件存储，所以堆积能力比较强，同时提供文件过期删除机制。

定时消息

定时消息，定时消息是指消息发送到 Rocket Broker 上之后，不被消费者理解消费，要到等待一定的时间才能进行消费，apache 的版本目 前只支持等待指定的时间才能被消费，不支持任意精度的定时消息消费。（一个说法是任意精度的定时消息会带来性能损耗，但是阿里云版本的 RocketMQ 却提供这样的功能，充值收费优先策略？）

消息重试机制

消息重试是指在消息消费时，如果发送异常，那么消息中间件需要支持消息重新投递，RocketMQ 支持消息重试机制。

Message Key

Key 一般用于消息在业务层面的唯一标识。对发送的消息设置好 Key，以后可以根据这个 Key 来查找消息。比如消息异常，消息丢失，进行查找会很 方便。

RocketMQ 会创建专门的索引文件，用来存储 Key 与消息的映射，由于是 Hash 索引，应尽量使 Key 唯一，避免潜在的哈希冲突。

Tag 和 Key 的主要差别是使用场景不同，Tag 用在 Consumer 代码中，用于服务端消息过滤，Key 主要用于通过命令进行查找消息 RocketMQ 并不能保证 message id 唯一，在这种情况下，生产者在 push 消息的时候可以给每条消息设定唯一的 key, 消费者可以通过 message key 保证对消息幂等处理。

消息被消费后会立即删除吗？

不会，RocketMQ每条消息都会持久化到CommitLog中，每个Consumer连接到Broker后会维持消费进度信息，当有消息消费后只是当前Consumer的消费进度（CommitLog的offset）更新了。

消息会堆积吗？什么时候清理过期消息？

先简单给大家说一下，其实默认broker会启动后台线程，这个后台线程会自动去检查CommitLog、ConsumeQueue文件，因为这些文件都是多个的，比如CommitLog会有多个，ConsumeQueue也会有多个。 然后如果是那种比较旧的超过72小时的文件，就会被删除掉，也就是说，默认来说，broker只会给你把数据保留3天而已，当然你也可以自己通过fileReservedTime来配置这个时间，要保留几天的时间。 这个定时检查过期数据文件的线程代码，在DefaultMessageStore这个类里，他的start0方法中会调用一个addScheduleTask0方法，里面会每隔10s定时调度执行一个后台检查任务，我们看下面的源码片段

上面就可以看到了，其实他是每隔10s，就会执行一个调度任务

这个调度任务里就会执行DefaultMessageStore,this.dleanFilesPeriodically0方法，其实就是会去周期性的清理掉磁盘上的数据文件，也就是超过72小时的CommitLog、ConsumeQueue文件，接着我们具体看看这里的清理逻辑，他其实里面包含了清理CommitLo和ConsumeQueue的清理逻辑，如下面源码片段。

在清理文件的时候，他会具体判断一下，如果当前时间是预先设置的凌晨4点，就会触发删除文件的逻辑，这个时间是默认的;或者是如果磁盘空间不足了，就是超过了85%的使用率了，立马会触发删除文件逻辑.

上面两个条件，第一个是说如果磁盘没有满，那么每天就默认一次会删除磁盘文件，默认就是凌晨4点执行，那个时候必然是业务低峰期，因为凌晨4点大部分人都睡觉了，无论什么业务都不会有太高业务量的。

第二个是说，如果磁盘使用率超过85%了，那么此时可以允许继续写入数据，但是此时会立马触发删除文件的逻辑如果磁盘使用率超过90%了，那么此时不允许在磁盘里写入新数悟，立马删除文件。这是因为，一旦磁盘满了，那么你写入磁盘会失败，此时你MO就彻底故障了 所以一且磁盘满了，也会立马删除文件的 在删除文件的时候，无非就是对文件进行遍历，如果一个文件超过72小时都没修改过了，此时就可以删除了，哪怕有的消息你可能还没消费过，但是此时也不会再让你消费了，就直接删除掉。这就是RocketMQ的一整套文件删除的逻辑和和机制

4.6版本默认72小时后会删除不再使用的CommitLog文件

• 检查这个文件最后访问时间
• 判断是否大于过期时间
• 指定时间删除，默认凌晨4点

RocketMQ消费模式有几种？

消费模型由Consumer决定，消费维度为Topic。

• 集群消费

1.一条消息只会被同Group中的一个Consumer消费

2.多个Group同时消费一个Topic时，每个Group都会有一个Consumer消费到数据

• 广播消费

消息将对一 个Consumer Group 下的各个 Consumer 实例都消费一遍。即即使这些 Consumer 属于同一个Consumer Group ，消息也会被 Consumer Group 中的每个 Consumer 都消费一次。

消费消息是push还是pull？-重点看

RocketMQ没有真正意义的push，都是pull，虽然有push类，但实际底层实现采用的是长轮询机制，即拉取方式

broker端属性 longPollingEnable 标记是否开启长轮询。默认开启

// {@link org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl#pullMessage()}
// 看到没，这是一只披着羊皮的狼，名字叫PushConsumerImpl，实际干的确是pull的活。
 
// 拉取消息，结果放到pullCallback里
this.pullAPIWrapper.pullKernelImpl(pullCallback);

实时性和消息堆积如何取舍

PULL方式

由消费者客户端主动向消息中间件（MQ消息服务器代理）拉取消息；采用Pull方式，如何设置Pull消息的拉取频率需要重点去考虑，举个例子来说，可能1分钟内连续来了1000条消息，然后2小时内没有新消息产生（概括起来说就是“消息延迟与忙等待”）

如果每次Pull的时间间隔比较久，会增加消息的延迟，即消息到达消费者的时间加长，MQ中消息的堆积量变大；若每次Pull的时间间隔较短，但是在一段时间内MQ中并没有任何消息可以消费，那么会产生很多无效的Pull请求的RPC开销，影响MQ整体的网络性能；

PUSH方式

由消息中间件（MQ消息服务器代理）主动地将消息推送给消费者；采用Push方式，可以尽可能实时地将消息发送给消费者进行消费。但是，在消费者的处理消息的能力较弱的时候(比如，消费者端的业务系统处理一条消息的流程比较复杂，其中的调用链路比较多导致消费时间比较久。概括起来地说就是“慢消费问题”)，而MQ不断地向消费者Push消息，消费者端的缓冲区可能会溢出或者产生消息堆积，导致异常；

代码@1:如果开启了长轮询模式，则每次只挂起 5s，然后就去尝试拉取。

代码@2:如果不开启长轮询模式，则只挂起一次，挂起时间为 shortPollingTimeMills，然后去尝试查找消息。

代码@3:遍历 pullRequestTable，如果拉取任务的待拉取偏移量小于当前队列的最大偏移量时执行拉取，否则如果没有超过最大等待时间则等待，否则返回未拉取到消息，返回给消息拉取客户端

参考链接：blog.csdn.net/prestigedin…

为什么主动拉取不使用事件监听方式？

事件驱动方式是建立好长连接，由事件（发送数据）的方式来实时推送。

如果broker主动推送消息的话有可能push速度快，消费速度慢的情况，那么就会造成消息在consumer端堆积过多，同时又不能被其他consumer消费的情况。

而pull的方式可以根据当前自身情况来pull，不会造成过多的压力而造成瓶颈。

所以采取了pull的方式。消费者端的缓冲区可能会溢出，导致异常；

broker如何处理拉取请求的？

Consumer首次请求Broker

• Broker中是否有符合条件的消息

• 有 ->

  • 响应Consumer
  • 等待下次Consumer的请求

• 没有

  • 挂起consumer的请求，即不断开连接，也不返回数据

  • 使用consumer的offset

    • DefaultMessageStore#ReputMessageService#run方法

      • 每隔1ms检查commitLog中是否有新消息，有的话写入到pullRequestTable
      • 当有新消息的时候返回请求

    • PullRequestHoldService 来Hold连接，每隔5s执行一次检查pullRequestTable有没有消息，有的话立即推送

RocketMQ如何做负载均衡？

通过Topic在多Broker中分布式存储实现。

producer端

发送端指定message queue发送消息到相应的broker，来达到写入时的负载均衡

• 提升写入吞吐量，当多个producer同时向一个broker写入数据的时候，性能会下降
• 消息分布在多broker中，为负载消费做准备

默认策略是平均选择：

• producer维护一个自增index
• 每次取节点会自增
• index向所有broker个数取余
• 自带容错策略

其他实现：

• SelectMessageQueueByHash

  • hash的是传入的args

• SelectMessageQueueByRandom

• SelectMessageQueueByMachineRoom 没有实现

也可以自定义实现MessageQueueSelector接口中的select方法

MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);

consumer端

采用的是平均分配算法来进行负载均衡。

其他负载均衡算法

平均分配策略(默认)(AllocateMessageQueueAveragely) 环形分配策略(AllocateMessageQueueAveragelyByCircle) 手动配置分配策略(AllocateMessageQueueByConfig) 机房分配策略(AllocateMessageQueueByMachineRoom) 一致性哈希分配策略(AllocateMessageQueueConsistentHash) 靠近机房策略(AllocateMachineRoomNearby)

consumer和queue不对等的时候会发生什么？

Consumer和queue会优先平均分配，如果Consumer少于queue的个数，则会存在部分Consumer消费多个queue的情况，如果Consumer等于queue的个数，那就是一个Consumer消费一个queue，如果Consumer个数大于queue的个数，那么会有部分Consumer空余出来，白白的浪费了。