• 中小型公司，技术实力较为一般，技术挑战不是特别高，用 RabbitMQ 是不错的选择；
• 大型公司，基础架构研发实力较强，用 RocketMQ 是很好的选择。
• 大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，几乎是全世界这个领域的事实性规范。

可靠性: RabbitMQ使用一些机制来保证可靠性， 如持久化、传输确认及发布确认等。

灵活的路由 : 在消息进入队列之前，通过交换器来路由消息。对于典型的路由功能， RabbitMQ 己经提供了一些内置的交换器来实现。针对更复杂的路由功能，可以将多个 交换器绑定在一起， 也可以通过插件机制来实现自己的交换器。

扩展性: 多个RabbitMQ节点可以组成一个集群，也可以根据实际业务情况动态地扩展 集群中节点。

高可用性 : 队列可以在集群中的机器上设置镜像，使得在部分节点出现问题的情况下队 列仍然可用。

多种协议: RabbitMQ除了原生支持AMQP协议，还支持STOMP， MQTT等多种消息 中间件协议。

多语言客户端 :RabbitMQ 几乎支持所有常用语言，比如 Java、 Python、 Ruby、 PHP、 C#、 JavaScript 等。

管理界面 : RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息、集 群中的节点等。

令插件机制: RabbitMQ 提供了许多插件 ， 以实现从多方面进行扩展，当然也可以编写自 己的插件。

RabbitMQ就是 AMQP 协议的 Erlang 的实现(当然 RabbitMQ 还支持 STOMP2、 MQTT3 等协议 ) AMQP 的模型架构 和 RabbitMQ 的模型架构是一样的，生产者将消息发送给交换器，交换器和队列绑定 。

RabbitMQ 中的交换器、交换器类型、队列、绑定、路由键等都是遵循的 AMQP 协议中相 应的概念。目前 RabbitMQ 最新版本默认支持的是 AMQP 0-9-1。

Module Layer:协议最高层，主要定义了一些客户端调用的命令，客户端可以用这些命令实现自己的业务逻辑。

Session Layer:中间层，主要负责客户端命令发送给服务器，再将服务端应答返回客户端，提供可靠性同步机制和错误处理。

TransportLayer:最底层，主要传输二进制数据流，提供帧的处理、信道服用、错误检测和数据表示等。

• Broker可以看做RabbitMQ的服务节点。一般请下一个Broker可以看做一个RabbitMQ服务器。
• Queue:RabbitMQ的内部对象，用于存储消息。多个消费者可以订阅同一队列，这时队列中的消息会被平摊（轮询）给多个消费者进行处理。
• Exchange:生产者将消息发送到交换器，由交换器将消息路由到一个或者多个队列中。当路由不到时，或返回给生产者或直接丢弃。

1. Producer先连接到Broker,建立连接Connection,开启一个信道(Channel)。

2. Producer声明一个交换器并设置好相关属性。

3. Producer声明一个队列并设置好相关属性。

4. Producer通过路由键将交换器和队列绑定起来。

5. Producer发送消息到Broker,其中包含路由键、交换器等信息。

6. 相应的交换器根据接收到的路由键查找匹配的队列。

7. 如果找到，将消息存入对应的队列，如果没有找到，会根据生产者的配置丢弃或者退回给生产者。

8. 关闭信道。

9. 管理连接。

1. Producer先连接到Broker,建立连接Connection,开启一个信道(Channel)。

2. 向Broker请求消费响应的队列中消息，可能会设置响应的回调函数。

3. 等待Broker回应并投递相应队列中的消息，接收消息。

4. 消费者确认收到的消息,ack。

5. RabbitMq从队列中删除已经确定的消息。

6. 关闭信道。

7. 关闭连接。

1. Client发送消息给MQ

2. MQ将消息持久化后，发送Ack消息给Client，此处有可能因为网络问题导致Ack消息无法发送到Client，那么Client在等待超时后，会重传消息；

3. Client收到Ack消息后，认为消息已经投递成功。

1. MQ将消息push给Client（或Client来pull消息）

2. Client得到消息并做完业务逻辑

3. Client发送Ack消息给MQ，通知MQ删除该消息，此处有可能因为网络问题导致Ack失败，那么Client会重复消息，这里就引出消费幂等的问题；

4. MQ将已消费的消息删除。

RabbitMQ 有三种模式：单机模式，普通集群模式，镜像集群模式。

单机模式：就是demo级别的，一般就是你本地启动了玩玩儿的，没人生产用单机模式

普通集群模式：意思就是在多台机器上启动多个RabbitMQ实例，每个机器启动一个。

镜像集群模式：这种模式，才是所谓的RabbitMQ的高可用模式，跟普通集群模式不一样的是，你创建的queue，无论元数据(元数据指RabbitMQ的配置数据)还是queue里的消息都会存在于多个实例上，然后每次你写消息到queue的时候，都会自动把消息到多个实例的queue里进行消息同步。

• 是一个队列模型的消息中间件，具有高性能、高可靠、高实时、分布式等特点
• Producer、Consumer、队列都可以分布式
• Producer 向一些队列轮流发送消息，队列集合称为 Topic，Consumer 如果做广播消费，则一个 Consumer 实例消费这个 Topic 对应的所有队列，如果做集群消费，则多个 Consumer 实例平均消费这个 Topic 对应的队列集合
• 能够保证严格的消息顺序
• 支持拉（pull）和推（push）两种消息模式
• 高效的订阅者水平扩展能力
• 实时的消息订阅机制
• 亿级消息堆积能力
• 支持多种消息协议，如 JMS、OpenMessaging 等
• 较少的依赖

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

1、集群消费

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

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

2、广播消费

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

1.1.14. RocketMQ消费消息是push还是pull？

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

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

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

如果broker主动推送消息的话有可能push速度快，消费速度慢的情况，那么就会造成消息在consumer端堆积过多，同时又不能被其他consumer消费的情况。而pull的方式可以根据当前自身情况来pull，不会造成过多的压力而造成瓶颈。所以采取了pull的方式。

Consumer首次请求Broker

• Broker中是否有符合条件的消息
  • 有
    • 响应Consumer
    • 等待下次Consumer的请求
  • 没有
    • DefaultMessageStore#ReputMessageService#run方法
    • PullRequestHoldService 来Hold连接，每个5s执行一次检查pullRequestTable有没有消息，有的话立即推送
    • 每隔1ms检查commitLog中是否有新消息，有的话写入到pullRequestTable
    • 当有新消息的时候返回请求
    • 挂起consumer的请求，即不断开连接，也不返回数据
    • 使用consumer的offset

首先在如下三个部分都可能会出现丢失消息的情况：

• Producer端
• Broker端
• Consumer端

1 、Producer端如何保证消息不丢失

• 采取send()同步发消息，发送结果是同步感知的。

• 发送失败后可以重试，设置重试次数。默认3次。

• 集群部署，比如发送失败了的原因可能是当前Broker宕机了，重试的时候会发送到其他Broker上。

2、Broker端如何保证消息不丢失

• 修改刷盘策略为同步刷盘。默认情况下是异步刷盘的。

• 集群部署，主从模式，高可用。

3、Consumer端如何保证消息不丢失

• 完全消费正常后在进行手动ack确认。

首先要找到是什么原因导致的消息堆积，是Producer太多了，Consumer太少了导致的还是说其他情况，总之先定位问题。

然后看下消息消费速度是否正常，正常的话，可以通过上线更多consumer临时解决消息堆积问题

• 准备一个临时的topic
• queue的数量是堆积的几倍
• queue分布到多Broker中
• 上线一台Consumer做消息的搬运工，把原来Topic中的消息挪到新的Topic里，不做业务逻辑处理，只是挪过去
• 上线N台Consumer同时消费临时Topic中的数据
• 改bug
• 恢复原来的Consumer，继续消费之前的Topic

RocketMQ中的消息只会在commitLog被删除的时候才会消失，不会超时。也就是说未被消费的消息不会存在超时删除这情况。

不会，消息在消费失败后会进入重试队列（%RETRY%+ConsumerGroup），18次（默认18次，网上所有文章都说是16次，无一例外。但是我没搞懂为啥是16次，这不是18个时间吗 ？）才会进入死信队列（%DLQ%+ConsumerGroup）。

1. RocketMQ只需要一个轻量级的维护元数据信息的组件，为此引入zk增加维护成本还强依赖另一个中间件了。
2. RocketMQ追求的是AP，而不是CP，也就是需要高可用。
   • zk是CP，因为zk节点间通过zap协议有数据共享，每个节点数据会一致，但是zk集群当挂了一半以上的节点就没法使用了。
   • nameserver是AP，节点间不通信，这样会导致节点间数据信息会发生短暂的不一致，但每个broker都会定时向所有nameserver上报路由信息和心跳。当某个broker下线了，nameserver也会延时30s才知道，而且不会通知客户端（生产和消费者），只能靠客户端自己来拉，rocketMQ是靠消息重试机制解决这个问题的，所以是最终一致性。但nameserver集群只要有一个节点就可用。

Kafka 将消息以 topic 为单位进行归纳

将向 Kafka topic 发布消息的程序成为 producers.

将预订 topics 并消费消息的程序成为 consumer.

Kafka 以集群的方式运行，可以由一个或多个服务组成，每个服务叫做一个 broker.

producers 通过网络将消息发送到 Kafka 集群，集群向消费者提供消息

Kafka 的基本架构组成是：由多个 broker 组成一个集群，每个 broker 是一个节点；当创建一个 topic 时，这个 topic 会被划分为多个 partition，每个 partition 可以存在于不同的 broker 上，每个 partition 只存放一部分数据。

这就是天然的分布式消息队列，就是说一个 topic 的数据，是分散放在多个机器上的，每个机器就放一部分数据。

在 Kafka 0.8 版本之前，是没有 HA 机制的，当任何一个 broker 所在节点宕机了，这个 broker 上的 partition 就无法提供读写服务，所以这个版本之前，Kafka 没有什么高可用性可言。

在 Kafka 0.8 以后，提供了 HA 机制，就是 replica 副本机制。每个 partition 上的数据都会同步到其它机器，形成自己的多个 replica 副本。所有 replica 会选举一个 leader 出来，消息的生产者和消费者都跟这个 leader 打交道，其他 replica 作为 follower。写的时候，leader 会负责把数据同步到所有 follower 上去，读的时候就直接读 leader 上的数据即可。Kafka 负责均匀的将一个 partition 的所有 replica 分布在不同的机器上，这样才可以提高容错性。

img

拥有了 replica 副本机制，如果某个 broker 宕机了，这个 broker 上的 partition 在其他机器上还存在副本。如果这个宕机的 broker 上面有某个 partition 的 leader，那么此时会从其 follower 中重新选举一个新的 leader 出来，这个新的 leader 会继续提供读写服务，这就有达到了所谓的高可用性。

写数据的时候，生产者只将数据写入 leader 节点，leader 会将数据写入本地磁盘，接着其他 follower 会主动从 leader 来拉取数据，follower 同步好数据了，就会发送 ack 给 leader，leader 收到所有 follower 的 ack 之后，就会返回写成功的消息给生产者。

消费数据的时候，消费者只会从 leader 节点去读取消息，但是只有当一个消息已经被所有 follower 都同步成功返回 ack 的时候，这个消息才会被消费者读到。

img

生产者使用push模式将消息发布到Broker，消费者使用pull模式从Broker订阅消息。

push模式很难适应消费速率不同的消费者，如果push的速度太快，容易造成消费者拒绝服务或网络拥塞；如果push的速度太慢，容易造成消费者性能浪费。但是采用pull的方式也有一个缺点，就是当Broker没有消息时，消费者会陷入不断地轮询中，为了避免这点，kafka有个参数可以让消费者阻塞知道是否有新消息到达。

• Kafka 持久化日志，这些日志可以被重复读取和无限期保留
• Kafka 是一个分布式系统：它以集群的方式运行，可以灵活伸缩，在内部通过复制数据提升容错能力和高可用性
• Kafka 支持实时的流式处理

消费者组是Kafka独有的概念，即消费者组是Kafka提供的可扩展且具有容错性的消费者机制。

但实际上，消费者组（Consumer Group）其实包含两个概念，作为队列，消费者组允许你分割数据处理到一组进程集合上（即一个消费者组中可以包含多个消费者进程，他们共同消费该topic的数据），这有助于你的消费能力的动态调整；作为发布-订阅模型（publish-subscribe），Kafka允许你将同一份消息广播到多个消费者组里，以此来丰富多种数据使用场景。

需要注意的是：在消费者组中，多个实例共同订阅若干个主题，实现共同消费。同一个组下的每个实例都配置有相同的组ID，被分配不同的订阅分区。当某个实例挂掉的时候，其他实例会自动地承担起它负责消费的分区。 因此，消费者组在一定程度上也保证了消费者程序的高可用性。

目前，Kafka使用ZooKeeper存放集群元数据、成员管理、Controller选举，以及其他一些管理类任务。之后，等KIP-500提案完成后，Kafka将完全不再依赖于ZooKeeper。

• “存放元数据”是指主题分区的所有数据都保存在 ZooKeeper 中，且以它保存的数据为权威，其他 “人” 都要与它保持对齐。
• “成员管理” 是指 Broker 节点的注册、注销以及属性变更，等等。
• “Controller 选举” 是指选举集群 Controller，而其他管理类任务包括但不限于主题删除、参数配置等。

KIP-500 思想，是使用社区自研的基于Raft的共识算法，替代ZooKeeper，实现Controller自选举。

在Kafka中，每个主题分区下的每条消息都被赋予了一个唯一的ID数值，用于标识它在分区中的位置。这个ID数值，就被称为位移，或者叫偏移量。一旦消息被写入到分区日志，它的位移值将不能被修改。

• Cache Filesystem Cache PageCache缓存
• 顺序写：由于现代的操作系统提供了预读和写技术，磁盘的顺序写大多数情况下比随机写内存还要快。
• Zero-copy：零拷技术减少拷贝次数
• Batching of Messages：批量量处理。合并小的请求，然后以流的方式进行交互，直顶网络上限。
• Pull 拉模式：使用拉模式进行消息的获取消费，与消费端处理能力相符。

• 1（默认） 数据发送到Kafka后，经过leader成功接收消息的的确认，就算是发送成功了。在这种情况下，如果leader宕机了，则会丢失数据。
• 0 生产者将数据发送出去就不管了，不去等待任何返回。这种情况下数据传输效率最高，但是数据可靠性确是最低的。
• -1producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成，可靠性最高。当ISR中所有Replica都向Leader发送ACK时，leader才commit，这时候producer才能认为一个请求中的消息都commit了。

首先需要弄明白消息为什么会丢失，对于一个消息队列，会有 生产者、MQ、消费者 这三个角色，在这三个角色数据处理和传输过程中，都有可能会出现消息丢失。

消息丢失的原因以及解决办法：

1. 消费者异常导致的消息丢失

消费者可能导致数据丢失的情况是：消费者获取到了这条消息后，还未处理，Kafka 就自动提交了 offset，这时 Kafka 就认为消费者已经处理完这条消息，其实消费者才刚准备处理这条消息，这时如果消费者宕机，那这条消息就丢失了。

消费者引起消息丢失的主要原因就是消息还未处理完 Kafka 会自动提交了 offset，那么只要关闭自动提交 offset，消费者在处理完之后手动提交 offset，就可以保证消息不会丢失。但是此时需要注意重复消费问题，比如消费者刚处理完，还没提交 offset，这时自己宕机了，此时这条消息肯定会被重复消费一次，这就需要消费者根据实际情况保证幂等性。

1. 生产者数据传输导致的消息丢失

对于生产者数据传输导致的数据丢失主常见情况是生产者发送消息给 Kafka，由于网络等原因导致消息丢失，对于这种情况也是通过在 producer 端设置 acks=all 来处理，这个参数是要求 leader 接收到消息后，需要等到所有的 follower 都同步到了消息之后，才认为本次写成功了。如果没满足这个条件，生产者会自动不断的重试。

1. Kafka 导致的消息丢失

Kafka 导致的数据丢失一个常见的场景就是 Kafka 某个 broker 宕机，，而这个节点正好是某个 partition 的 leader 节点，这时需要重新重新选举该 partition 的 leader。如果该 partition 的 leader 在宕机时刚好还有些数据没有同步到 follower，此时 leader 挂了，在选举某个 follower 成 leader 之后，就会丢失一部分数据。

对于这个问题，Kafka 可以设置如下 4 个参数，来尽量避免消息丢失：

• 给 topic 设置 replication.factor 参数：这个值必须大于 1，要求每个 partition 必须有至少 2 个副本；
• 在 Kafka 服务端设置 min.insync.replicas 参数：这个值必须大于 1，这个参数的含义是一个 leader 至少感知到有至少一个 follower 还跟自己保持联系，没掉队，这样才能确保 leader 挂了还有一个 follower 节点。
• 在 producer 端设置 acks=all，这个是要求每条数据，必须是写入所有 replica 之后，才能认为是写成功了；
• 在 producer 端设置 retries=MAX（很大很大很大的一个值，无限次重试的意思）：这个参数的含义是一旦写入失败，就无限重试，卡在这里了。

在某些业务场景下，我们需要保证对于有逻辑关联的多条MQ消息被按顺序处理，比如对于某一条数据，正常处理顺序是新增-更新-删除，最终结果是数据被删除；如果消息没有按序消费，处理顺序可能是删除-新增-更新，最终数据没有被删掉，可能会产生一些逻辑错误。对于如何保证消息的顺序性，主要需要考虑如下两点：

• 如何保证消息在 Kafka 中顺序性；

• 如何保证消费者处理消费的顺序性。

• 如何保证消息在 Kafka 中顺序性

对于 Kafka，如果我们创建了一个 topic，默认有三个 partition。生产者在写数据的时候，可以指定一个 key，比如在订单 topic 中我们可以指定订单 id 作为 key，那么相同订单 id 的数据，一定会被分发到同一个 partition 中去，而且这个 partition 中的数据一定是有顺序的。消费者从 partition 中取出来数据的时候，也一定是有顺序的。通过制定 key 的方式首先可以保证在 kafka 内部消息是有序的。

1. 如何保证消费者处理消费的顺序性

对于某个 topic 的一个 partition，只能被同组内部的一个 consumer 消费，如果这个 consumer 内部还是单线程处理，那么其实只要保证消息在 MQ 内部是有顺序的就可以保证消费也是有顺序的。但是单线程吞吐量太低，在处理大量 MQ 消息时，我们一般会开启多线程消费机制，那么如何保证消息在多个线程之间是被顺序处理的呢？对于多线程消费我们可以预先设置 N 个内存 Queue，具有相同 key 的数据都放到同一个内存 Queue 中；然后开启 N 个线程，每个线程分别消费一个内存 Queue 的数据即可，这样就能保证顺序性。当然，消息放到内存 Queue 中，有可能还未被处理，consumer 发生宕机，内存 Queue 中的数据会全部丢失，这就转变为上面提到的如何保证消息的可靠传输的问题了。

• ISR：In-Sync Replicas 副本同步队列
• AR:Assigned Replicas 所有副本

ISR是由leader维护，follower从leader同步数据有一些延迟（包括延迟时间replica.lag.time.max.ms和延迟条数replica.lag.max.messages两个维度，当前最新的版本0.10.x中只支持replica.lag.time.max.ms这个维度），任意一个超过阈值都会把follower剔除出ISR，存入OSR（Outof-Sync Replicas）列表，新加入的follower也会先存放在OSR中。

AR=ISR+OSR。

Kafka副本当前分为领导者副本和追随者副本。只有Leader副本才能对外提供读写服务，响应Clients端的请求。Follower副本只是采用拉（PULL）的方式，被动地同步Leader副本中的数据，并且在Leader副本所在的Broker宕机后，随时准备应聘Leader副本。

加分点：

• 强调Follower副本也能对外提供读服务。自Kafka 2.4版本开始，社区通过引入新的Broker端参数，允许Follower副本有限度地提供读服务。
• 强调Leader和Follower的消息序列在实际场景中不一致。通常情况下，很多因素可能造成Leader和Follower之间的不同步，比如程序问题，网络问题，broker问题等，短暂的不同步我们可以关注（秒级别），但长时间的不同步可能就需要深入排查了，因为一旦Leader所在节点异常，可能直接影响可用性。

注意：之前确保一致性的主要手段是高水位机制（HW），但高水位值无法保证Leader连续变更场景下的数据一致性，因此，社区引入了Leader Epoch机制，来修复高水位值的弊端。

分区的Leader副本选举对用户是完全透明的，它是由Controller独立完成的。你需要回答的是，在哪些场景下，需要执行分区Leader选举。每一种场景对应于一种选举策略。

• OfflinePartition Leader选举：每当有分区上线时，就需要执行Leader选举。所谓的分区上线，可能是创建了新分区，也可能是之前的下线分区重新上线。这是最常见的分区Leader选举场景。
• ReassignPartition Leader选举：当你手动运行kafka-reassign-partitions命令，或者是调用Admin的alterPartitionReassignments方法执行分区副本重分配时，可能触发此类选举。假设原来的AR是[1，2，3]，Leader是1，当执行副本重分配后，副本集合AR被设置成[4，5，6]，显然，Leader必须要变更，此时会发生Reassign Partition Leader选举。
• PreferredReplicaPartition Leader选举：当你手动运行kafka-preferred-replica-election命令，或自动触发了Preferred Leader选举时，该类策略被激活。所谓的Preferred Leader，指的是AR中的第一个副本。比如AR是[3，2，1]，那么，Preferred Leader就是3。
• ControlledShutdownPartition Leader选举：当Broker正常关闭时，该Broker上的所有Leader副本都会下线，因此，需要为受影响的分区执行相应的Leader选举。

这4类选举策略的大致思想是类似的，即从AR中挑选首个在ISR中的副本，作为新Leader。

在Kafka中，体现Zero Copy使用场景的地方有两处：基于mmap的索引和日志文件读写所用的TransportLayer。

先说第一个。索引都是基于MappedByteBuffer的，也就是让用户态和内核态共享内核态的数据缓冲区，此时，数据不需要复制到用户态空间。不过，mmap虽然避免了不必要的拷贝，但不一定就能保证很高的性能。在不同的操作系统下，mmap的创建和销毁成本可能是不一样的。很高的创建和销毁开销会抵消Zero Copy带来的性能优势。由于这种不确定性，在Kafka中，只有索引应用了mmap，最核心的日志并未使用mmap机制。

再说第二个。TransportLayer是Kafka传输层的接口。它的某个实现类使用了FileChannel的transferTo方法。该方法底层使用sendfile实现了Zero Copy。对Kafka而言，如果I/O通道使用普通的PLAINTEXT，那么，Kafka就可以利用Zero Copy特性，直接将页缓存中的数据发送到网卡的Buffer中，避免中间的多次拷贝。相反，如果I/O通道启用了SSL，那么，Kafka便无法利用Zero Copy特性了。

在 Kafka 中，生产者写入消息、消费者读取消息的操作都是与 leader 副本进行交互的，从 而实现的是一种主写主读的生产消费模型。

Kafka 并不支持主写从读，因为主写从读有 2 个很明 显的缺点:

• 数据一致性问题。数据从主节点转到从节点必然会有一个延时的时间窗口，这个时间 窗口会导致主从节点之间的数据不一致。某一时刻，在主节点和从节点中 A 数据的值都为 X， 之后将主节点中 A 的值修改为 Y，那么在这个变更通知到从节点之前，应用读取从节点中的 A 数据的值并不为最新的 Y，由此便产生了数据不一致的问题。
• 延时问题。类似 Redis 这种组件，数据从写入主节点到同步至从节点中的过程需要经历网络→主节点内存→网络→从节点内存这几个阶段，整个过程会耗费一定的时间。而在 Kafka 中，主从同步会比 Redis 更加耗时，它需要经历网络→主节点内存→主节点磁盘→网络→从节点内存→从节点磁盘这几个阶段。对延时敏感的应用而言，主写从读的功能并不太适用。