几种常见消息队列(了解即可)


多维度对比5款主流分布式MQ消息队列

分布式之消息队列的特点、选型、及应用场景详解


开发语言:

1)Kafka:Scala

2)RabbitMQ:Erlang

3)ZeroMQ:C语言

4)RocketMQ:Java

5)ActiveMQ:Java


支持的协议

1)Kafka:自己定义的一套…(基于TCP)

2)RabbitMQ:AMQP

3)ZeroMQ:TCP、UDP

4)RocketMQ:自己定义的一套…

5)ActiveMQ:OpenWire、STOMP、REST、XMPP、AMQP


负载均衡


Kafka
支持负载均衡。

1)一个broker通常就是一台服务器节点。

对于同一个Topic的不同分区,Kafka会尽力将这些分区分布到不同的Broker服务器上,zookeeper保存了broker、主题和分区的元数据信息。

分区首领会处理来自客户端的生产请求,Kafka分区首领会被分配到不同的broker服务器上,让不同的broker服务器共同分担任务。

每一个broker都缓存了元数据信息,客户端可以从任意一个broker获取元数据信息并缓存起来,根据元数据信息知道要往哪里发送请求。

2)Kafka的消费者组订阅同一个topic,会尽可能地使得每一个消费者分配到相同数量的分区,分摊负载。

3)当消费者加入或者退出消费者组的时候,还会触发再均衡,为每一个消费者重新分配分区,分摊负载。

Kafka的负载均衡大部分是自动完成的,分区的创建也是Kafka完成的,隐藏了很多细节,避免了繁琐的配置和人为疏忽造成的负载问题。

4)发送端由topic和key来决定消息发往哪个分区,如果key为null,那么会使用轮询算法将消息均衡地发送到同一个topic的不同分区中。如果key不为null,那么会根据key的hashcode取模计算出要发往的分区。

RabbitMQ
对负载均衡的支持不好

ZeroMQ
去中心化,不支持负载均衡。

RocketMQ
支持负载均衡。

ActiveMQ
支持负载均衡。可以基于zookeeper实现负载均衡。


可用性


1)Kafka:非常高(分布式)

2)RabbitMQ:高(主从)

3)ZeroMQ:高

4)RocketMQ:非常高(分布式)

5)ActiveMQ:高(主从)


消息重复


1)Kafka:支持at least once、at most once

2)RabbitMQ:支持at least once、at most once

3)ZeroMQ:只有重传机制,但是没有持久化,消息丢了重传也没有用。既不是at least once、也不是at most once、更不是exactly only once

4)RocketMQ:支持at least once

5)ActiveMQ:支持at least once


吞吐量TPS


TPS: Transactions Per Second的缩写,即 事务数/秒.

1)Kafka:极大。Kafka按批次发送消息和消费消息。发送端将多个小消息合并,批量发向Broker,消费端每次取出一个批次的消息批量处理。

2)RabbitMQ:比较大

3)ZeroMQ:极大

4)RocketMQ:大。RocketMQ接收端可以批量消费消息,可以配置每次消费的消息数,但是发送端不是批量发送。

5)ActiveMQ:比较大


吞顺序消息


Kafka:支持。


设置生产者的max.in.flight.requests.per.connection为1,可以保证消息是按照发送顺序写入服务器的,即使发生了重试。

Kafka保证同一个分区里的消息是有序的,但是这种有序分两种情况:

①key为null,消息逐个被写入不同主机的分区中,但是对于每个分区依然是有序的

②key不为null , 消息被写入到同一个分区,这个分区的消息都是有序。


RabbitMQ:不支持

ZeroMQ:不支持

RocketMQ:支持

ActiveMQ:不支持


消息确认


Kafka
支持。

1)发送方确认机制:

ack=0,不管消息是否成功写入分区

ack=1,消息成功写入首领分区后,返回成功

ack=all,消息成功写入所有分区后,返回成功。

2)接收方确认机制:

自动或者手动提交分区偏移量,早期版本的Kafka偏移量是提交给Zookeeper的,这样使得zookeeper的压力比较大,更新版本的Kafka的偏移量是提交给Kafka服务器的,不再依赖于zookeeper群组,集群的性能更加稳定。


RabbitMQ
支持。

1)发送方确认机制,消息被投递到所有匹配的队列后,返回成功。如果消息和队列是可持久化的,那么在写入磁盘后,返回成功。支持批量确认和异步确认。

2)接收方确认机制,设置autoAck为false,需要显式确认,设置autoAck为true,自动确认。

当autoAck为false的时候,RabbitMQ队列会分成两部分,一部分是等待投递给consumer的消息,一部分是已经投递但是没收到确认的消息。

如果一直没有收到确认信号,并且consumer已经断开连接,RabbitMQ会安排这个消息重新进入队列,投递给原来的消费者或者下一个消费者。

未确认的消息不会有过期时间,如果一直没有确认,并且没有断开连接,RabbitMQ会一直等待,RabbitMQ允许一条消息处理的时间可以很久很久。


ZeroMQ
支持。

RocketMQ
支持。

ActiveMQ
支持。



消息回溯


1)Kafka:支持指定分区offset位置的回溯

2)RabbitMQ:不支持

3)ZeroMQ:不支持

4)RocketMQ:支持指定时间点的回溯

5)ActiveMQ:不支持


并发度


Kafka
并发度高。

一个线程一个消费者,Kafka限制消费者的个数要小于等于分区数,如果要提高并行度,可以在消费者中再开启多线程,或者增加consumer实例数量。

19.2 RabbitMQ
并发度极高。本身是用Erlang语言写的,并发性能高。

可在消费者中开启多线程,最常用的做法是一个channel对应一个消费者,每一个线程把持一个channel,多个线程复用connection的tcp连接,减少性能开销。

当RabbitMQ队列拥有多个消费者的时候,队列收到的消息将以轮询的分发方式发送给消费者。每条消息只会发送给订阅列表里的一个消费者,不会重复。

这种方式非常适合扩展,而且是专门为并发程序设计的。

如果某些消费者的任务比较繁重,那么可以设置basicQos限制信道上消费者能保持的最大未确认消息的数量,在达到上限时,RabbitMQ不再向这个消费者发送任何消息。


ZeroMQ
并发度高。


RocketMQ
并发度高。

1)RocketMQ限制消费者的个数少于等于队列数,但是可以在消费者中再开启多线程,这一点和Kafka是一致的,提高并行度的方法相同。

修改消费并行度方法:

同一个 ConsumerGroup 下,通过增加 Consumer 实例数量来提高并行度,超过订阅队列数的 Consumer实例无效。

提高单个 Consumer 的消费并行线程,通过修改参数consumeThreadMin、consumeThreadMax

2)同一个网络连接connection,客户端多个线程可以同时发送请求,连接会被复用,减少性能开销。


ActiveMQ
并发度高。

单个ActiveMQ的接收和消费消息的速度在1万笔/秒(持久化 一般为1-2万, 非持久化 2 万以上),在生产环境中部署10个ActiveMQ就能达到10万笔/秒以上的性能,部署越多的ActiveMQ broker 在MQ上latency也就越低,系统吞吐量也就越高。


Kafka相关


Terminology


  • Broker Kafka集群包含一个或多个服务器,这种服务器被称为 broker

  • Topic 每条发布到 Kafka 集群的消息都有一个类别,这个类别被称为 Topic。(物理上不同 Topic 的消息分开存储,逻辑上一个 Topic 的消息虽然保存于一个或多个 broker 上但用户只需指定消息的 Topic 即可生产或消费数据而不必关心数据存于何处)

  • Partition Parition是物理上的概念,每个 Topic 包含一个或多个 Partition.

  • Producer 负责发布消息到 Kafka broker

  • Consumer 消息消费者,向 Kafka broker 读取消息的客户端。

  • Consumer Group 每个 Consumer 属于一个特定的 Consumer Group(可为每个 Consumer 指定 group name,若不指定 group name 则属于默认的 group)。


Topic & Partition


Topic 在逻辑上可以被认为是一个 queue,每条消费都必须指定它的 Topic,可以简单理解为必须指明把这条消息放进哪个 queue 里。为了使得 Kafka 的吞吐率可以线性提高,物理上把 Topic 分成一个或多个 Partition,每个 Partition 在物理上对应一个文件夹,该文件夹下存储这个 Partition 的所有消息和索引文件。若创建 topic1 和 topic2 两个 topic,且分别有 13 个和 19 个分区,则整个集群上会相应会生成共 32 个文件夹


因为每条消息都被 append 到该 Partition 中,属于顺序写磁盘,因此效率非常高(经验证,顺序写磁盘效率比随机写内存还要高,这是 Kafka 高吞吐率的一个很重要的保证)


对于传统的 message queue 而言,一般会删除已经被消费的消息,而 Kafka 集群会保留所有的消息,无论其被消费与否。当然,因为磁盘限制,不可能永久保留所有数据(实际上也没必要),因此 Kafka 提供两种策略删除旧数据。一是基于时间,二是基于 Partition 文件大小。例如可以通过配置 $KAFKA_HOME/config/server.properties,让 Kafka 删除一周前的数据,也可在 Partition 文件超过 1GB 时删除旧数据


Kafka 读取特定消息的时间复杂度为 O(1),即与文件大小无关,所以这里删除过期文件与提高 Kafka 性能无关。选择怎样的删除策略只与磁盘以及具体的需求有关。另外,Kafka 会为每一个 Consumer Group 保留一些 metadata 信息——当前消费的消息的 position,也即 offset。这个 offset 由 Consumer 控制。正常情况下 Consumer 会在消费完一条消息后递增该 offset。当然,Consumer 也可将 offset 设成一个较小的值,重新消费一些消息。因为 offet 由 Consumer 控制,所以 Kafka broker 是无状态的,它不需要标记哪些消息被哪些消费过,也不需要通过 broker 去保证同一个 Consumer Group 只有一个 Consumer 能消费某一条消息,因此也就不需要锁机制,这也为 Kafka 的高吞吐率提供了有力保障。


Push vs. Pull


作为一个消息系统,Kafka 遵循了传统的方式,选择由 Producer 向 broker push 消息并由 Consumer 从 broker pull 消息。一些 logging-centric system,比如 Facebook 的 Scribe 和 Cloudera 的 Flume ,采用 push 模式。事实上,push 模式和 pull 模式各有优劣。

push 模式很难适应消费速率不同的消费者,因为消息发送速率是由 broker 决定的。push 模式的目标是尽可能以最快速度传递消息,但是这样很容易造成 Consumer 来不及处理消息,典型的表现就是拒绝服务以及网络拥塞。而 pull 模式则可以根据 Consumer 的消费能力以适当的速率消费消息。

对于 Kafka 而言,pull 模式更合适。pull 模式可简化 broker 的设计,Consumer 可自主控制消费消息的速率,同时 Consumer 可以自己控制消费方式——即可批量消费也可逐条消费,同时还能选择不同的提交方式从而实现不同的传输语义。


Kafka delivery guarantee


有这么几种可能的 delivery guarantee:

At most once 消息可能会丢,但绝不会重复传输

At least one 消息绝不会丢,但可能会重复传输

Exactly once 每条消息肯定会被传输一次且仅传输一次,很多时候这是用户所想要的。

当 Producer 向 broker 发送消息时,一旦这条消息被 commit,因数 replication 的存在,它就不会丢。但是如果 Producer 发送数据给 broker 后,遇到网络问题而造成通信中断,那 Producer 就无法判断该条消息是否已经 commit。虽然 Kafka 无法确定网络故障期间发生了什么,但是 Producer 可以生成一种类似于主键的东西,发生故障时幂等性的重试多次,这样就做到了 Exactly once。截止到目前 (Kafka 0.8.2 版本,2015-03-04),这一 Feature 还并未实现,有希望在 Kafka 未来的版本中实现。(所以目前默认情况下一条消息从 Producer 到 broker 是确保了 At least once,可通过设置 Producer 异步发送实现 At most once)。

接下来讨论的是消息从 broker 到 Consumer 的 delivery guarantee 语义。(仅针对 Kafka consumer high level API)。Consumer 在从 broker 读取消息后,可以选择 commit,该操作会在 Zookeeper 中保存该 Consumer 在该 Partition 中读取的消息的 offset。该 Consumer 下一次再读该 Partition 时会从下一条开始读取。如未 commit,下一次读取的开始位置会跟上一次 commit 之后的开始位置相同。当然可以将 Consumer 设置为 autocommit,即 Consumer 一旦读到数据立即自动 commit。如果只讨论这一读取消息的过程,那 Kafka 是确保了 Exactly once。但实际使用中应用程序并非在 Consumer 读取完数据就结束了,而是要进行进一步处理,而数据处理与 commit 的顺序在很大程度上决定了消息从 broker 和 consumer 的 delivery guarantee semantic。

读完消息先 commit 再处理消息。这种模式下,如果 Consumer 在 commit 后还没来得及处理消息就 crash 了,下次重新开始工作后就无法读到刚刚已提交而未处理的消息,这就对应于 At most once

读完消息先处理再 commit。这种模式下,如果在处理完消息之后 commit 之前 Consumer crash 了,下次重新开始工作时还会处理刚刚未 commit 的消息,实际上该消息已经被处理过了。这就对应于 At least once。在很多使用场景下,消息都有一个主键,所以消息的处理往往具有幂等性,即多次处理这一条消息跟只处理一次是等效的,那就可以认为是 Exactly once。(笔者认为这种说法比较牵强,毕竟它不是 Kafka 本身提供的机制,主键本身也并不能完全保证操作的幂等性。而且实际上我们说 delivery guarantee 语义是讨论被处理多少次,而非处理结果怎样,因为处理方式多种多样,我们不应该把处理过程的特性——如是否幂等性,当成 Kafka 本身的 Feature)

如果一定要做到 Exactly once,就需要协调 offset 和实际操作的输出。精典的做法是引入两阶段提交。如果能让 offset 和操作输入存在同一个地方,会更简洁和通用。这种方式可能更好,因为许多输出系统可能不支持两阶段提交。比如,Consumer 拿到数据后可能把数据放到 HDFS,如果把最新的 offset 和数据本身一起写到 HDFS,那就可以保证数据的输出和 offset 的更新要么都完成,要么都不完成,间接实现 Exactly once。(目前就 high level API 而言,offset 是存于 Zookeeper 中的,无法存于 HDFS,而 low level API 的 offset 是由自己去维护的,可以将之存于 HDFS 中)

总之,Kafka 默认保证 At least once,并且允许通过设置 Producer 异步提交来实现 At most once。而 Exactly once 要求与外部存储系统协作,幸运的是 Kafka 提供的 offset 可以非常直接非常容易得使用这种方式。


参考:

Kafka 背景及架构介绍–力荐

kafka消息分区机制

Kafka分区与消费者的关系

Apache kafka 工作原理介绍

Kafka 入门一篇文章就够了