消息队列kafka，RabbitMQ，ActiveMQ，RocketMQ对比

文章目录

• • 综述
  • kafka
  • • 架构
    • 优势：
    • 应用场景：
    • 特点：
    • • 1 磁盘顺序读写
      • 2 零拷贝
      • 3 分区分段+索引
      • 4 批量压缩，批量读写
      • 5 直接操作page chahe
      • 6 pull，push分析
      • 7 异步
    • 消息丢失分析
    • • 1 Producer端-ack机制-ack=-1
      • 2 broker端-异步批量刷盘-减小刷盘间隔
      • 3 Consumer 端-提交 Offset 记录-手动提交
    • rablance机制
  • 1 如何保证分布式事务的一致性
  • 2如何保证幂等性
  • RocketMQ
  • • 架构
    • 集群
    • 刷盘机制
    • 事务消息
    • 顺序消息

综述

ActiveMQ：
1 很是成熟，功能强大，在业内大量的公司以及项目中都有应用，
2 偶尔会有较低几率丢失消息，
3 社区以及国内应用都愈来愈少，官方社区如今对ActiveMQ 5.x维护愈来愈少，几个月才发布一个版本，并且确实主要是基于解耦和异步来用的，较少在大规模吞吐的场景中使用日志

RabbitMQ：
1 erlang语言开发，性能极其好，延时很低；
2 吞吐量到万级，MQ功能比较完备，并且开源提供的管理界面很是棒，用起来很好用，
3 社区相对比较活跃，几乎每月都发布几个版本分在国内一些互联网公司近几年用rabbitmq也比较多一些，可是问题也是显而易见的，
4 RabbitMQ确实吞吐量会低一些，这是由于他作的实现机制比较重。并且erlang开发，国内有几个公司有实力作erlang源码级别的研究和定制？若是说你没这个实力的话，确实偶尔会有一些问题，你很难去看懂源码，你公司对这个东西的掌控很弱，基本职能依赖于开源社区的快速维护和修复bug。并且rabbitmq集群动态扩展会很麻烦，不过这个我以为还好。其实主要是erlang语言自己带来的问题。很难读源码，很难定制和掌控。

RocketMQ：
1 接口简单易用，并且毕竟在阿里大规模应用过，有阿里品牌保障，
2 日处理消息上百亿之多，能够作到大规模吞吐，性能也很是好，分布式扩展也很方便，社区维护还能够，可靠性和可用性都是ok的，还能够支撑大规模的topic数量，支持复杂MQ业务场景，
3 阿里出品都是java系的，咱们能够本身阅读源码，定制本身公司的MQ，能够掌控，社区活跃度相对较为通常，不过也还能够，文档相对来讲简单一些，而后接口这块不是按照标准JMS规范走的有些系统要迁移须要修改大量代码，还有就是阿里出台的技术，你得作好这个技术万一被抛弃，社区黄掉的风险，那若是大家公司有技术实力我以为用RocketMQ挺好的

Kafka：
1 提供超高的吞吐量，ms级的延迟，
2 极高的可用性以及可靠性，并且分布式能够任意扩展，同时kafka最好是支撑较少的topic数量便可，保证其超高吞吐量
3 kafka惟一的一点劣势是有可能消息重复消费，那么对数据准确性会形成极其轻微的影响，在大数据领域中以及日志采集中，这点轻微影响能够忽略这个特性自然适合大数据实时计算以及日志收集。

kafka 架构

Producer：Producer 即生产者，消息的产生者，是消息的入口

Broker：Broker 是 kafka 一个实例，每个服务器上有一个或多个 kafka 的实例，简单的理解就是一台 kafka 服务器，kafka cluster表示集群的意思

Topic：消息的主题，可以理解为消息队列，kafka的数据就保存在topic。在每个 broker 上都可以创建多个 topic 。

Partition：Topic的分区，每个 topic 可以有多个分区，分区的作用是做负载，提高 kafka 的吞吐量。同一个 topic 在不同的分区的数据是不重复的，partition 的表现形式就是一个一个的文件夹！

Replication：每一个分区都有多个副本，副本的作用是做备胎，主分区（Leader）会将数据同步到从分区（Follower）。当主分区（Leader）故障的时候会选择一个备胎（Follower）上位，成为 Leader。在kafka中默认副本的最大数量是10个，且副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本

Message：每一条发送的消息主体。

Consumer：消费者，即消息的消费方，是消息的出口。

Consumer Group：我们可以将多个消费组组成一个消费者组，在 kafka 的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量！

Zookeeper：kafka 集群依赖 zookeeper 来保存集群的的元信息，来保证系统的可用性。

简而言之，kafka 本质就是一个消息系统，与大多数的消息系统一样，主要的特点如下：

使用推拉模型将生产者和消费者分离
为消息传递系统中的消息数据提供持久性，以允许多个消费者
提供高可用集群服务，主从模式，同时支持横向水平扩展
与 ActiveMQ、RabbitMQ、RocketMQ 不同的地方在于，它有一个分区Partition的概念。

这个分区的意思就是说，如果你创建的topic有5个分区，当你一次性向 kafka 中推 1000 条数据时，这 1000 条数据默认会分配到 5 个分区中，其中每个分区存储 200 条数据。

这样做的目的，就是方便消费者从不同的分区拉取数据，假如你启动 5 个线程同时拉取数据，每个线程拉取一个分区，消费速度会非常非常快！

Zookeeper 在 kafka 中的具体作用

1 Broker注册
Broker是分布式部署并且相互之间相互独立，但是需要有一个注册中心对整个集群的Broker进行管理，此时就使用了Zookeeper。在Zookeeper上会有一个专门用来记录Broker服务器列表的节点：/brokers/ids

每个Broker在启动时，都会在Zookeeper上进行注册，即到**/brokers/ids下创建属于自己的节点，如/brokers/ids/[0…N]。**

Kafka使用了全局唯一的数字来指代每个Broker服务器，不同的Broker必须使用不同的Broker ID进行注册，创建完节点后，每个Broker就会将自己的IP地址和端口信息记录到该节点中去

2 Topic注册
在kafka中，用户可以自定义多个topic，每个topic又被划分为多个分区，每个分区存储在一个独立的broker上。这些分区信息及与Broker的对应关系都是由Zookeeper进行维护

3 生产者负载均衡
同一个Topic消息会被分区并将其分布在多个Broker上。由于每个Broker启动时，都会在Zookeeper上进行注册，生产者会通过该节点的变化来动态地感知到Broker服务器列表的变更，这样就可以实现动态的负载均衡

4 消费者负载均衡
与生产者类似，Kafka中的消费者同样需要进行负载均衡来实现多个消费者合理地从对应的Broker服务器上接收消息，每个消费者分组包含若干消费者，每条消息都只会发送给分组中的一个消费者，不同的消费者分组消费自己特定的Topic下面的消息，互不干扰。
每个消费者都需要关注所属消费者分组中其他消费者服务器的变化情况，即对/consumers/[group_id]/ids节点注册子节点变化的Watcher监听，一旦发现消费者新增或减少，就触发消费者的负载均衡。还对Broker服务器变化注册监听。消费者需要对/broker/ids/[0-N]中的节点进行监听，如果发现Broker服务器列表发生变化，那么就根据具体情况来决定是否需要进行消费者负载均衡。

5 分区与消费者的关系
消费者组 Consumer group 下有多个 Consumer（消费者）。
对于每个消费者组 (Consumer Group)，Kafka都会为其分配一个全局唯一的Group ID，Group 内部的所有消费者共享该 ID。订阅的topic下的每个分区只能分配给某个 group 下的一个consumer(当然该分区还可以被分配给其他group)。同时，Kafka为每个消费者分配一个Consumer ID。
在Kafka中，规定了每个消息分区 只能被同组的一个消费者进行消费，因此，需要在 Zookeeper 上记录 消息分区 与 Consumer 之间的关系，每个消费者一旦确定了对一个消息分区的消费权力，需要将其Consumer ID 写入到 Zookeeper 对应消息分区的临时节点上，例如：
/consumers/[group_id]/owners/[topic]/[broker_id-partition_id]
其中，**[broker_id-partition_id]**就是一个 消息分区 的标识，节点内容就是该消息分区上消费者的Consumer ID。
、
6 记录消息消费的进度Offset
在消费者对指定消息分区进行消息消费的过程中，需要定时地将分区消息的消费进度Offset记录到Zookeeper上，以便在该消费者进行重启或者其他消费者重新接管该消息分区的消息消费后，能够从之前的进度开始继续进行消息消费。Offset在Zookeeper中由一个专门节点进行记录，其节点路径为:
/consumers/[group_id]/offsets/[topic]/[broker_id-partition_id]
节点内容就是Offset的值。

7 Consumer注册
注册新的消费者分组
当新的消费者组注册到zookeeper中时，zookeeper会创建专用的节点来保存相关信息，其节点路径为 /consumers/{group_id}，其节点下有三个子节点，分别为**[ids, owners, offsets]**。

ids节点：记录该消费组中当前正在消费的消费者；

owners节点：记录该消费组消费的topic信息；

offsets节点：记录每个topic的每个分区的offset；

优势：

高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒

可扩展性：kafka集群支持热扩展

持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

容错性：允许集群中节点故障（若副本数量为n,则允许n-1个节点故障）

高并发：支持数千个客户端同时读写

应用场景：

日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer

消息系统：解耦生产者和消费者、缓存消息等

用户活动跟踪：kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后消费者通过订阅这些topic来做实时的监控分析，亦可保存到数据库

运营指标：kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告

流式处理：比如spark streaming和storm；

特点： 1 磁盘顺序读写

传统的随机读写，需要磁头不断的寻址，读到数据再返回给内存

顺序写-基于硬盘存储，数据堆积量大，kafka的写入是append操作，partion是有序的，而且是批量操作
顺序读-连续的按顺序读，每次读多页，放在内存中，所以实际看起来就像从内盘读取一样快速

2 零拷贝

用户态进程想要访问内核态的磁盘或网卡数据，需要切换到内核态，内核从内核空间读取数据后copy到用户空间，
kafka从磁盘读到数据，直接把数据给内核读缓冲区，缓冲区将数据发到网卡，

3 分区分段+索引

Kafka将一个分区的文件是按照片段来存储的，一个片段的默认大小为1GB，可以在server.properties配置文件中修改片段大小，并且同时维护了index索引文件。
Partition：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列
Segment：partition物理上由多个segment组成，每个Segment存着message信息

这种partition-segment的分区分段，在并发的时候可以分段加锁，提高的效率

offset：表示的是相对于该分区的记录偏移量，指的是第几条记录，比如0代表第一条记录。
position：表示该记录相对于当前片段文件的偏移量。
CreateTime:记录创建的时间。
isvalid：记录是否有效。
keysize：表示key的长度。
valuesize：表示value的长度
magic：表示本次发布kafka服务程序协议版本号。
compresscodec：压缩工具。
producerId：生产者ID（用于幂等机制）。
sequence：消息的序列号（用于幂等机制）。
payload：表示具体的消息

index文件：

4 批量压缩，批量读写

批量压缩，降低传输消耗

5 直接操作page chahe

java写入磁盘，需要先创建对象，通过JVM序列化的写入磁盘

kafka避开了java的JVM，直接写入磁盘的page，避免了对象的创建，gc的耗时，读写速度高，进程重启也不会丢缓存

Kafka Broker 集群接收到数据后会将数据进行持久化存储到磁盘，为了提高吞吐量和性能，采用的是「异步批量刷盘的策略」，也就是说按照一定的消息量和间隔时间进行刷盘。首先会将数据存储到 「PageCache」 中，至于什么时候将 Cache 中的数据刷盘是由「操作系统」根据自己的策略决定或者调用 fsync 命令进行强制刷盘，

6 pull，push分析

7 异步

kafak生产一条消息后，没有直接发给broker，而是先缓存起来，返回给业务ok，待消息达到一定数量，再批量发送给broker，
好处：减少网络Io，提高吞吐率
坏处：如果生产者宕机，可能造成消息丢失，牺牲可靠性

消息丢失分析

首先需要保证本地事务的原子性

比较容易出现笑死丢失的四个环节：

2.1生产者发送消息不丢失

1 Producer端-ack机制-ack=-1

1、主线程KafkaProducer创建消息，然后通过生产者拦截器
2、生产者拦截器对消息的key ，value做一定的处理，交给序列化器，
3、序列化器对消息key和vlue做序列化处理，然后给分区器
4、分区器给消息分配分区、并发送给消息收集器
5、一条ProducerRecord，添加到RecordAccumulator，首先会根据分区确定对应分区所在的双端队列，在双端队列获取尾部的一个ProducerBatch对象，查看该ProducerBatch是否可以写入该ProducerRecord消息，如果可以则写入，不能的话，会在双端队列末尾在创建一个ProducerBatch对象，创建时会评估这条消息是否是否超过batch.size参数的大小。如果不超过，就以batch.size参数的大小来创建ProducerBatch对象。通过sender线程发送消息
6、sender线程获取RecordAccumulator中的消息，需要将原本的<分区，Deque>形式再次封装成的形式，这样就可以将请求发往各个Node了。这里的Request是指kafka的各种请求协议。
7、sender线程发往Kafka之前还会保存到InFlightRequest中，InFlightRequest保存对象的具体形式为Map,他的主要作用是缓存了已经发出去但是还没有收到相应的请求。
8、sender将Request交给Selector准备发送。
9、Selector将Request发送到对应的kafka节点（Broker）。
10、Selector相应结果反馈给InFlightRequest。
11、主线程清理RecordAccumulator已经发送完毕的消息。

kafka的ack机制默认设置为ack=1，这就可能造成消息丢失
ack机制：
acks = 0：由于发送后就自认为发送成功，这时如果发生网络抖动， Producer 端并不会校验 ACK 自然也就丢了，且无法重试。
acks = 1：消息发送 Leader Parition 接收成功就表示发送成功，这时只要 Leader Partition 不 Crash 掉，就可以保证 Leader Partition 不丢数据，但是如果 Leader Partition 异常 Crash 掉了， Follower Partition 还未同步完数据且没有 ACK，这时就会丢数据。

acks = -1 或者 all： 消息发送需要等待 ISR 中 Leader Partition 和 所有的 Follower Partition 都确认收到消息才算发送成功, 可靠性最高, 但也不能保证不丢数据,比如当 ISR 中只剩下 Leader Partition 了, 这样就变成 acks = 1 的情况了。

选举机制：
AR-分区中的所有副本统称为 AR (Assigned Replicas)
ISR-所有与leader副本保持一定程度同步的副本（包括leader副本在内）组成 ISR (In Sync Replicas)。
OSR-于leader副本同步滞后过多的副本（不包括leader副本）将组成 OSR （Out-of-Sync Replied）

解决：
1 ack=-1/all
2 Leader Partition选举， 从ISR选，不从OSR选
3 重试次数 retries：>0, 这样Producer 端就会一直进行重试直到 Broker 端返回 ACK 标识
同步副本数min.insync.replices>1

2 broker端-异步批量刷盘-减小刷盘间隔

Kafka Broker 集群接收到数据后会将数据进行持久化存储到磁盘，为了提高吞吐量和性能，采用的是「异步批量刷盘的策略」，也就是说按照一定的消息量和间隔时间进行刷盘。首先会将数据存储到 「PageCache」 中，至于什么时候将 Cache 中的数据刷盘是由「操作系统」根据自己的策略决定或者调用 fsync 命令进行强制刷盘，如果此时 Broker 宕机 Crash 掉，且选举了一个落后 Leader Partition 很多的 Follower Partition 成为新的 Leader Partition，那么落后的消息数据就会丢失。
既然 Broker 端消息存储是通过异步批量刷盘的，那么这里就可能会丢数据的!!!
由于 Kafka 中并没有提供「同步刷盘」的方式，所以说从单个 Broker 来看还是很有可能丢失数据的。
kafka 通过「多 Partition （分区）多 Replica（副本）机制」已经可以最大限度的保证数据不丢失，如果数据已经写入 PageCache 中但是还没来得及刷写到磁盘，此时如果所在 Broker 突然宕机挂掉或者停电，极端情况还是会造成数据丢失。

解决：保证消息落到磁盘，受动调用fsync，减小刷盘间隔

3 Consumer 端-提交 Offset 记录-手动提交

Consumer的工作可大致理解为两部分：
1 获取元数据并从 Kafka Broker 集群拉取数据。
2 处理消息，并标记消息已经被消费，提交 Offset 记录。

解决：
1因此正确的做法是：拉取数据、业务逻辑处理、提交消费 Offset 位移信息。

2 我们还需要设置参数 enable.auto.commit = false, 采用手动提交位移的方式

3 另外对于消费消息重复的情况，业务自己保证幂等性, 保证只成功消费一次即可–（sequence序列号？）

rablance机制

1 如何保证分布式事务的一致性

分布式事务：业务相关的多个操作，保证他们同时成功或者同时失败

最终一致性：与之对应的就是强一致性

MQ中要保证事务的最终一致性，就需要做到两点
1.生产者要保证100%的消息投递。事务消息机制
2.消费者这一端需要保证幂等消费。唯一ID+业务自己实现的幂等

分布式MQ的三种语义：
at least once
at most once
exactly once:

exactly-once：即使producer重试发送消息，消息也会保证最多一次地传递给最终consumer。该语义是最理想的，但也难以实现，这是因为它需要消息系统本身与生产和消费消息的应用程序进行协作。例如如果在消费消息成功后，将Kafka consumer的偏移量rollback，我们将会再次从该偏移量开始接收消息。这表明消息传递系统和客户端应用程序必须配合调整才能实现excactly-once

at-least-once：如果producer收到来自Kafka broker的确认（ack）或者acks = all，则表示该消息已经写入到Kafka。但如果producer ack超时或收到错误，则可能会重试发送消息，客户端会认为该消息未写入Kafka。如果broker在发送Ack之前失败，但在消息成功写入Kafka之后，此重试将导致该消息被写入两次，因此消息会被不止一次地传递给最终consumer，这种策略可能导致重复的工作和不正确的结

at least once 最少一次，消息不会丢失，但是可能会重复
exactly once 恰好一次。消息有且仅会被传输一次

RocketMQ并不能保证exactly once。商业版本当中提供了exactly once的实现机制。
Kafka:在最新版本的源码当中，提供了exactly once的demo。
RabbitMQ:erlang天生就成为了一种屏障

2如何保证幂等性

RocketMQ 架构

Producer：就是消息生产者，可以集群部署。它会先和 NameServer 集群中的随机一台建立长连接，得知当前要发送的 Topic 存在哪台 Broker Master上，然后再与其建立长连接，支持多种负载平衡模式发送消息。

Consumer：消息消费者，也可以集群部署。它也会先和 NameServer 集群中的随机一台建立长连接，得知当前要消息的 Topic 存在哪台 Broker Master、Slave上，然后它们建立长连接，支持集群消费和广播消费消息。

Broker：主要负责消息的存储、查询消费，支持主从部署，一个 Master 可以对应多个 Slave，Master 支持读写，Slave 只支持读。Broker 会向集群中的每一台 NameServer 注册自己的路由信息。

NameServer：是一个很简单的 Topic 路由注册中心，支持 Broker 的动态注册和发现，保存 Topic 和 Borker 之间的关系。通常也是集群部署，但是各 NameServer 之间不会互相通信， 各 NameServer 都有完整的路由信息，即无状态。

我再用一段话来概括它们之间的交互：

先启动 NameServer 集群，各 NameServer 之间无任何数据交互，Broker 启动之后会向所有 NameServer 定期（每 30s）发送心跳包，包括：IP、Port、TopicInfo，NameServer 会定期扫描 Broker 存活列表，如果超过 120s 没有心跳则移除此 Broker 相关信息，代表下线。

这样每个 NameServer 就知道集群所有 Broker 的相关信息，此时 Producer 上线从 NameServer 就可以得知它要发送的某 Topic 消息在哪个 Broker 上，和对应的 Broker （Master 角色的）建立长连接，发送消息。

Consumer 上线也可以从 NameServer 得知它所要接收的 Topic 是哪个 Broker ，和对应的 Master、Slave 建立连接，接收消息。

集群

1 NameServer 集群
提供轻量级的服务发现及路由，每个 NameServer 记录完整的路由信息，提供相应的读写服务，支持快速存储扩展。

NameServer是一个功能齐全的服务器，主要包含两个功能：

Broker 管理，接收来自 Broker 集群的注册请求，提供心跳机制检测 Broker 是否存活
路由管理，每个 NameServer 持有全部有关 Broker 集群和客户端请求队列的路由信息
2 Broker 集群
通过提供轻量级的 Topic 和Queue 机制处理消息存储。同时支持推（Push）和拉（Pull）两种模型，包含容错机制。提供强大的峰值填充和以原始时间顺序累积数千亿条消息的能力。此外还提供灾难恢复，丰富的指标统计数据和警报机制，这些都是传统的消息系统缺乏的。

Broker 有几个重要的子模块：

远程处理模块，Broker 入口，处理来自客户端的请求
客户端管理，管理客户端（包括消息生产者和消费者），维护消费者的主题订阅
存储服务，提供在物理硬盘上存储和查询消息的简单 API
HA 服务，提供主从 Broker 间数据同步
索引服务，通过指定键为消息建立索引并提供快速消息查询
3 Producer 集群
消息生产者支持分布式部署，分布式生产者通过多种负载均衡模式向 Broker 集群发送消息。

.4 Consumer 集群
消息消费者也支持 Push 和 Pull 模型的分布式部署，还支持集群消费和消息广播。提供了实时的消息订阅机制，可以满足大多数消费者的需求。

刷盘机制

事务消息

（1）正常情况：在事务主动方服务正常，没有发生故障的情况下，发消息流程如下：

步骤①：MQ 发送方向 MQ Server 发送 half 消息，MQ Server 标记消息状态为 prepared，此时该消息 MQ 订阅方是无法消费到的

步骤②：MQ Server 将消息持久化成功之后，向发送方 ACK 确认消息已经成功接收

步骤③：发送方开始执行本地事务逻辑

步骤④：发送方根据本地事务执行结果向 MQ Server 提交二次确认，commit 或 rollback

最终步骤：MQ Server 如果收到的是 commit 操作，则将半消息标记为可投递，MQ订阅方最终将收到该消息；若收到的是 rollback 操作则删除 half 半消息，订阅方将不会接受该消息；如果本地事务执行结果没响应或者超时，则 MQ Server 回查事务状态，具体见步骤（2）的异常情况说明。

（2）异常情况：在断网或者应用重启等异常情况下，图中的步骤④提交的二次确认超时未到达 MQ Server，此时的处理逻辑如下：

步骤⑤：MQ Server 对该消息进行消息回查

步骤⑥：发送方收到消息回查后，检查该消息的本地事务执行结果

步骤⑦：发送方根据检查得到的本地事务的最终状态再次提交二次确认。

最终步骤：MQ Server基于 commit/rollback 对消息进行投递或者删除

顺序消息