Message Queuing Service

Base

什么是消息队列

消息队列（message queue, MQ）也叫消息中间件（Message-oriented middleware, MOM）是在消息的传输过程中保存消息的容器，用于接收消息并以文件的方式存储，一个消息队列可以被一个也可以被多个消费者消费，包含以下 3 元素：

  • Producer：消息生产者，负责产生和发送消息到 Broker；
  • Broker：消息处理中心，负责消息存储、确认、重试等，一般其中会包含多个 Queue；
  • Consumer：消息消费者，负责从 Broker 中获取消息，并进行相应处理。

消息队列模式

  • 点对点模式：多个生产者可以向同一个消息队列发送消息，一个具体的消息只能由一个消费者消费。
  • 发布/订阅模式：单个消息可以被多个订阅者并发的获取和处理。

消息队列应用场景

  • 应用解耦：消息队列减少了服务之间的耦合性，不同的服务可以通过消息队列进行通信，而不用关心彼此的实现细节。
  • 异步处理：消息队列本身是异步的，它允许接收者在消息发送很长时间后再取回消息。
  • 流量削锋：当上下游系统处理能力存在差距的时候，利用消息队列做一个通用的”载体”，在下游有能力处理的时候，再进行分发与处理。
  • 日志处理：日志处理是指将消息队列用在日志处理中，比如 Kafka 的应用，解决大量日志传输的问题。
  • 消息通讯：消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯，比如实现点对点消息队列，或者聊天室等。
  • 消息广播：如果没有消息队列，每当一个新的业务方接入，我们都要接入一次新接口。有了消息队列，我们只需要关心消息是否送达了队列，至于谁希望订阅，是下游的事情，无疑极大地减少了开发和联调的工作量。

常用消息队列

由于官方社区现在对 ActiveMQ 5.x 维护越来越少，较少在大规模吞吐的场景中使用，文本主要讲解 Kafka、RabbitMQ 和 RocketMQ。

Kafka

Apache Kafka 最初由 LinkedIn 公司基于独特的设计实现为一个分布式的提交日志系统，之后成为 Apache 项目的一部分，号称大数据的杀手锏，在数据采集、传输、存储的过程中发挥着举足轻重的作用。

它是一个分布式的，支持多分区、多副本，基于 Zookeeper 的分布式消息流平台，它同时也是一款开源的基于发布订阅模式的消息引擎系统。

概念

  • 主题（Topic） ：消息的种类称为主题，可以说一个主题代表了一类消息，相当于是对消息进行分类，主题就像是数据库中的表。
  • 分区（partition） ：主题可以被分为若干个分区，同一个主题中的分区可以不在一个机器上，有可能会部署在多个机器上，由此来实现 kafka 的伸缩性。
  • 批次：为了提高效率， 消息会分批次写入 Kafka，批次就代指的是一组消息。
  • 消费者群组（Consumer Group） ：消费者群组指的就是由一个或多个消费者组成的群体。
  • Broker: 一个独立的 Kafka 服务器就被称为 broker，broker 接收来自生产者的消息，为消息设置偏移量，并提交消息到磁盘保存。
  • Broker 集群：broker 集群由一个或多个 broker 组成。
  • 重平衡（Rebalance） ：消费者组内某个消费者实例挂掉后，其他消费者实例自动重新分配订阅主题分区的过程。

Kafka 架构

一个典型的 Kafka 集群中包含 Producer、broker、Consumer Group、Zookeeper 集群。

生产者将数据生产出来，交给 broker 进行存储，消费者需要消费数据了，就从 broker 中去拿出数据来，然后完成一系列对数据的处理操作。多个 broker 协同合作，producer 和 consumer 部署在各个业务逻辑中被频繁的调用，三者通过 zookeeper 管理协调请求和转发，这样一个高性能的分布式消息发布订阅系统就完成了。

Zookeeper 管理集群配置，选举 leader，以及在 Consumer Group 发生变化时进行 rebalance。Producer 使用 push 模式将消息发布到 broker，Consumer 使用 pull 模式从 broker 订阅并消费消息。

Kafka 工作原理

消息经过序列化后，通过不同的分区策略，找到对应的分区。

相同主题和分区的消息，会被存放在同一个批次里，然后由一个独立的线程负责把它们发到 Kafka Broker 上。

分区是 Kafka 读写数据的最小粒度，比如主题 A 有 15 条消息，有 5 个分区，如果采用顺序轮询的方式，15 条消息会顺序分配给这 5 个分区，后续消费的时候，也是按照分区粒度消费。由于分区可以部署在多个不同的机器上，所以可以通过分区实现 Kafka 的伸缩性，比如主题 A 的 5 个分区，分别部署在 5 台机器上，如果下线一台，分区就变为 4。

Kafka 消费是通过消费群组完成，同一个消费者群组，一个消费者可以消费多个分区，但是一个分区，只能被一个消费者消费。

如果消费者增加，会触发 Rebalance，也就是分区和消费者需要重新配对。

不同的消费群组互不干涉，比如下图的 2 个消费群组，可以分别消费这 4 个分区的消息，互不影响。

分区的策略包括顺序轮询、随机轮询和 key hash 这 3 种方式

RabbitMQ

RabbitMQ 2007 年发布，是使用 Erlang 语言开发的开源消息队列系统，基于 AMQP 协议来实现。

AMQP 的主要特征是面向消息、队列、路由、可靠性、安全。AMQP 协议更多用在企业系统内，对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

概念

  • 信道（Channel） ：消息读写等操作在信道中进行，客户端可以建立多个信道，每个信道代表一个会话任务。
  • 交换器（Exchange） ：接收消息，按照路由规则将消息路由到一个或者多个队列；如果路由不到，或者返回给生产者，或者直接丢弃。
  • 路由键（RoutingKey） ：生产者将消息发送给交换器的时候，会发送一个 RoutingKey，用来指定路由规则，这样交换器就知道把消息发送到哪个队列。
  • 绑定（Binding） ：交换器和消息队列之间的虚拟连接，绑定中可以包含一个或者多个 RoutingKey。

RabbitMQ 工作原理

总结一下中间件服务器是什么:它是一个接受消息的数据服务器，并主要做两件事情，依据条件将消息路由给不同的消费者,当消费者消费速度不够快时，它会把消息缓存在内存或磁盘上．
在AMQP之前的服务器中，它们会通过实现了特定类型路由和缓存的庞大引擎来完成. AMQ模块使用较小的模块结合更多样和稳健的方案来实现. 它把这些任务分成了两个不同角色:

  • 交换器, 它接受来自生产者的消息并将它们路由到消息队列．
  • 消息队列, 它存储消息消息并把它们转发给消费者应用程序．

在交换器和消息队列之间有一个明显的界面，称为绑定(binding),随后会进行讲解.

AMQP提供了运行时程序语义，主要有两方面：

  • 运行时通过该协议可创建任意的交换器和消息队列类型的能力（有些是在标准中定义的，但可以添加其他作为服务器扩展）。
  • 运行时通过协议包装交换器和消息队列来创建任何需要的消息处理系统的能力．

RabbitMQ 是在各企業中最為廣泛使用的 message queue，而在 RabbitMQ 的世界裡面有三個角色，分別是 Producer、Queue、Consumer

RabbitMQ 常見的五種模式

在 RabbitMQ 中，常見的五種模式如下，詳情也可參照官網

  1. Direct 模式
  2. Worker 模式
  3. Publish/Subscribe 模式
  4. Routing 模式
  5. Topics 模式

Direct 模式

最簡單的模式，只會有一個 Producer 負責發送 message 到 Queue 裡、而也只有一個 Consumer 去 Queue 裡消費 message

Work Queues

为了避免等待一些占用大量资源、时间的操作。当我们把任务（Task）当作消息发送到队列中，一个运行在后台的工作者（worker）进程就会取出任务然后处理。当你运行多个工作者（workers），任务就会在它们之间共享。这个概念在网络应用中是非常有用的，它可以在短暂的HTTP请求中处理一些复杂的任务。

跟 Direct 模式很像，但是差別是 Worker 模式中會 同時 有多個 Consumer 會去消費 Queue 裡的 message，增加 message 消化的速率

Worker 模式是滿常用的一種模式，通常是用在大量 asynchronous job 的處理上，可以想像成 n 個 Consumer 就是一堆 Worker or Thread，一起幫忙消化 message

在現在 microservice 很流行的情形下，如果有大量的 Tjob 要執行，通常就不會使用 Java 裡的 ThreadPool，而是會改使用這種 Worker 模式，讓許多台 node 可以一起帮忙处理

默认来说，RabbitMQ会按顺序得把消息发送给每个消费者（consumer）。平均每个消费者都会收到同等数量得消息。这种发送消息得方式叫做——轮询（round-robin）。试着添加三个或更多得工作者（workers）。

消息确认

当处理一个比较耗时得任务的时候，你也许想知道消费者（consumers）是否运行到一半就挂掉。当前的代码中，当消息被RabbitMQ发送给消费者（consumers）之后，马上就会在内存中移除。这种情况，你只要把一个工作者（worker）停止，正在处理的消息就会丢失。同时，所有发送到这个工作者的还没有处理的消息都会丢失。

我们不想丢失任何任务消息。如果一个工作者（worker）挂掉了，我们希望任务会重新发送给其他的工作者（worker）。

为了防止消息丢失，RabbitMQ提供了消息响应（acknowledgments）。消费者会通过一个ack（响应），告诉RabbitMQ已经收到并处理了某条消息，然后RabbitMQ就会释放并删除这条消息。

如果消费者（consumer）挂掉了，没有发送响应，RabbitMQ就会认为消息没有被完全处理，然后重新发送给其他消费者（consumer）。这样，即使工作者（workers）偶尔的挂掉，也不会丢失消息。

消息是没有超时这个概念的；当工作者与它断开连的时候，RabbitMQ会重新发送消息。这样在处理一个耗时非常长的消息任务的时候就不会出问题了。

在本教程中，我们将使用手动消息确认，通过为auto-ack参数传递false，一旦有任务完成，使用d.Ack（false）向RabbitMQ服务器发送消费完成的确认（这个确认消息是单次传递的）。

忘记确认

忘记ack是一个常见的错误。这是一个简单的错误，但后果是严重的。当客户端退出时，消息将被重新传递（这可能看起来像随机重新传递），但是RabbitMQ将会占用越来越多的内存，因为它无法释放任何未经消息的消息 为了排除这种错误，你可以使用rabbitmqctl命令，输出messages_unacknowledged字段：

1
sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

Windows上执行：

1
rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged

消息持久化

如果你没有特意告诉RabbitMQ，那么在它退出或者崩溃的时候，将会丢失所有队列和消息。为了确保信息不会丢失，有两个事情是需要注意的：我们必须把“队列”和“消息”设为持久化。

首先，为了不让队列消失，需要把队列声明为持久化（durable）：

1
2
3
4
5
6
7
8
9
q, err := ch.QueueDeclare(
  "hello",      // name
  true,         // durable
  false,        // delete when unused
  false,        // exclusive
  false,        // no-wait
  nil,          // arguments
)
failOnError(err, "Failed to declare a queue")

尽管这行代码本身是正确的，但是达不到我们预期的结果。因为我们已经定义过一个叫hello的非持久化队列。RabbitMq不允许你使用不同的参数重新定义一个队列，它会返回一个错误。但我们现在使用一个快捷的解决方法——用不同的名字，例如task_queue。

1
2
3
4
5
6
7
8
9
q, err := ch.QueueDeclare(
  "task_queue", // name
  true,         // durable
  false,        // delete when unused
  false,        // exclusive
  false,        // no-wait
  nil,          // arguments
)
failOnError(err, "Failed to declare a queue")

这个durable必须在生产者（producer）和消费者（consumer）对应的代码中修改。

此时，已经确保即使RabbitMQ重新启动，task_queue队列也不会丢失。现在我们需要将消息标记为持久性 - 通过设置amqp.Publishing的amqp.Persistent属性完成。

1
2
3
4
5
6
7
8
9
10
err = ch.Publish(
  "",           // exchange
  q.Name,       // routing key
  false,        // mandatory
  false,
  amqp.Publishing {
    DeliveryMode: amqp.Persistent,
    ContentType:  "text/plain",
    Body:         []byte(body),
})

将消息设为持久化并不能完全保证不会丢失。以上代码只是告诉了RabbitMq要把消息存到硬盘，但从RabbitMq收到消息到保存之间还是有一个很小的间隔时间。因为RabbitMq并不是所有的消息都使用fsync(2)——它有可能只是保存到缓存中，并不一定会写到硬盘中。并不能保证真正的持久化，但已经足够应付我们的简单工作队列。如果需要更强的保证，可以使用publisher confirms.。

公平调度

你应该已经发现，它仍旧没有按照我们期望的那样进行分发。比如有两个工作者（workers），处理奇数消息的比较繁忙，处理偶数消息的比较轻松。然而RabbitMQ并不知道这些，它仍然一如既往的派发消息。

这时因为RabbitMQ只管分发进入队列的消息，不会关心有多少消费者（consumer）没有作出响应。它盲目的把第n-th条消息发给第n-th个消费者。

我们可以设置预取计数值为1。告诉RabbitMQ一次只向一个worker发送一条消息。换句话说，在处理并确认前一个消息之前，不要向工作人员发送新消息。

1
2
3
4
5
6
err = ch.Qos(
  1,     // prefetch count
  0,     // prefetch size
  false, // global
)
failOnError(err, "Failed to set QoS")

关于队列大小:

如果所有的工作者都处理繁忙状态，你的队列就会被填满。你需要留意这个问题，要么添加更多的工作者（workers），要么使用其他策略。

Publish/Subscribe 模式

從這個模式之後，在 Producer、Queue、Consumer 之間就多了一個叫做 Exchange 的東西出現，所以以後 Producer 不再是直接把 message 丟到 Queue 裡，而是讓 Producer 把 message 丟給 Exchange，再交由 Exchange 去決定要把這個 message 丟給哪個 Queue

RabbitMQ消息模型的核心理念是：发布者（producer）不会直接发送任何消息给队列。事实上，发布者（producer）甚至不知道消息是否已经被投递到队列。

发布者（producer）只需要把消息发送给一个交换机（exchange）。交换机非常简单，它一边从发布者方接收消息，一边把消息推送到队列。交换机必须知道如何处理它接收到的消息，是应该推送到指定的队列还是是多个队列，或者是直接忽略消息。这些规则是通过交换机类型（exchange type）来定义的。

Exchange 自己就有4種 type，用來表示要用什麼樣的規則把 message 丟到想要丟的 Queue 裡面

  • direct
  • fanout
  • topic
  • headers

在 Publish/Subscribe 模式中，使用的是 Exchange 的 fanout type，當 Producer 把 message 丟給 Exchange 時，Exchange 會把這個 message 丟到他綁定的所有 Queue 上

临时队列

给一个队列命名是很重要的——我们需要把工作者（workers）指向正确的队列。如果你打算在发布者（producers）和消费者（consumers）之间共享同队列的话，给队列命名是十分重要的。

如果我们的功能关心的是最新的消息而不是旧的。为了解决这个问题，需要做两件事情。

首先，当连接上RabbitMQ的时候，需要一个全新的、空的队列。可以手动创建一个随机的队列名，或者让服务器选择一个随机的队列名（推荐）。

其次，当与消费者（consumer）断开连接的时候，这个队列应当被立即删除。 在amqp客户端中，当我们将队列名称作为空字符串提供时，我们创建一个具有生成名称的非持久队列：

1
2
3
4
5
6
7
8
q, err := ch.QueueDeclare(
  "",    // name
  false, // durable
  false, // delete when usused
  true,  // exclusive
  false, // no-wait
  nil,   // arguments
)

该方法返回时，队列实例包含RabbitMQ生成的随机队列名称。例如，它可能看起来像amq.gen-JzTY20BRgKO-HjmUJj0wLg。 当声明它的连接关闭时，队列将被删除，因为它被声明为exclusive。可以在guide on queues了解有关exclusive和其他队列属性的更多信息

绑定（Bindings）

需要告诉交换机如何发送消息给队列。交换器和队列之间的联系称之为绑定（binding）。

可以使用rabbitmqctl list_bindings 列出所有现存的绑定。

Routing 模式

Routing 模式也是一個用到了 Exchange 的模式，這個模式使用的是 Exchange 的 direct type

當 Producer 把 message 丟給 Exchange 時，同時要在這個 message 上面帶上一個 routing key，而 Exchange 就會根據這個 routing key，將 message 丟到指定的 Queue 上

在这个场景中，我们可以看到direct交换机 X和两个队列进行了绑定。第一个队列使用orange作为binding key，第二个队列有两个绑定，一个使用black作为binding key，另外一个使用green。

这样一来，当消息发布到routing key为orange的交换机时，就会被路由到队列Q1。routing key为black或者green的消息就会路由到Q2。其他的所有消息都将会被丢弃。

乍看之下 Routing 模式和 Direct 模式很像，而且甚至會覺得 Routing 模式有點多餘，因為既然是透過 routing key 來決定要將這個 message 丟到哪個 Queue 上，那我就在 code 裡自己寫 if/else 來決定要丟到哪個 Queue 就好了啊，根本不用經過 Exchange，所以就會像下面這樣

1
2
3
4
5
6
String routingKey = "...";
if (routingKey.equals("orange")) {
    sendMessageToQ1();
} else if (routingKey.equals("black") || routingKey.equals("green")) {
    sendMessageToQ2();
}

但是 Routing 模式之所以有它存在的意義，重點就在於他可以 多重綁定，也就是說同一個 routing key 可以綁到 n 個 Queue 上，而不是一對一的綁定

多个队列使用相同的binding key是合法的。这个例子中，可以添加一个X和Q1之间的绑定，使用black为binding key。这样一来，direct交换机就和fanout交换机的行为一样，会将消息广播到所有匹配的队列。带有routing key为black的消息会同时发送到Q1和Q2。

所以我們就可以用一個 Exchange，將 info, error, warning routing key 綁到一般的 log queue上，然後可以再將 error routing key 綁到另一個 error log queue 上，這樣就可以做出一個帶有全部 log 的 Queue、以及一個只包含 error log 的 Queue

不過這種模式要慎用，因為他的重要邏輯都在 Exchange 上，所以如果是對 RabbitMQ 不熟的新手，從 code 中會看不出來原來還有這種用法，只會知道 Producer 把 message 丟給 Exchange，但卻不知道為什麼 log Queue 和 error log Queue 中卻都會有 message 出現

Topics 模式

用到了 Exchange 的 topic type，用法基本上跟 Routing 模式一樣，只是 routing key 進化成可以使用模糊綁定而已

binding key也必须拥有同样的格式。topic交换机背后的逻辑跟direct交换机很相似 —— 一个携带着特定routing_key的消息会被topic交换机投递给绑定键与之想匹配的队列。但是它的binding key和routing_key有两个特殊应用方式：

  • * (星号) 用来表示一个单词.
  • # (井号) 用来表示任意数量（零个或多个）单词。

RocketMQ

RocketMQ 是阿里开源的消息中间件，它是纯 Java 开发，具有高性能、高可靠、高实时、适合大规模分布式系统应用的特点。

RocketMQ 思路起源于 Kafka，但并不是 Kafka 的一个 Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog 分发等场景。

重要概念

  • Name 服务器（NameServer） ：充当注册中心，类似 Kafka 中的 Zookeeper。
  • Broker: 一个独立的 RocketMQ 服务器就被称为 broker，broker 接收来自生产者的消息，为消息设置偏移量。
  • 主题（Topic） ：消息的第一级类型，一条消息必须有一个 Topic。
  • 子主题（Tag） ：消息的第二级类型，同一业务模块不同目的的消息就可以用相同 Topic 和不同的 Tag 来标识。
  • 分组（Group） ：一个组可以订阅多个 Topic，包括生产者组（Producer Group）和消费者组（Consumer Group）。
  • 队列（Queue） ：可以类比 Kafka 的分区 Partition。

RocketMQ 工作原理

RockerMQ 中的消息模型就是按照主题模型所实现的，包括 Producer Group、Topic、Consumer Group 三个角色。

为了提高并发能力，一个 Topic 包含多个 Queue，生产者组根据主题将消息放入对应的 Topic，下图是采用轮询的方式找到里面的 Queue。

RockerMQ 中的消费群组和 Queue，可以类比 Kafka 中的消费群组和 Partition：不同的消费者组互不干扰，一个 Queue 只能被一个消费者消费，一个消费者可以消费多个 Queue。

消费 Queue 的过程中，通过偏移量记录消费的位置。

RocketMQ 架构

RocketMQ 技术架构中有四大角色 NameServer、Broker、Producer 和 Consumer，下面主要介绍 Broker。

Broker 用于存放 Queue，一个 Broker 可以配置多个 Topic，一个 Topic 中存在多个 Queue。

如果某个 Topic 消息量很大，应该给它多配置几个 Queue，并且尽量多分布在不同 broker 上，以减轻某个 broker 的压力。Topic 消息量都比较均匀的情况下，如果某个 broker 上的队列越多，则该 broker 压力越大。

Broker 通过集群部署，并且提供了 master/slave 的结构，salve 定时从 master 同步数据（同步刷盘或者异步刷盘），如果 master 宕机，则 slave 提供消费服务，但是不能写入消息。

看到这里，大家应该可以发现，RocketMQ 的设计和 Kafka 真的很像！

消息队列对比

Kafka

优点：

  • 高吞吐、低延迟：Kafka 最大的特点就是收发消息非常快，Kafka 每秒可以处理几十万条消息，它的最低延迟只有几毫秒；
  • 高伸缩性：每个主题（topic）包含多个分区（partition），主题中的分区可以分布在不同的主机（broker）中；
  • 高稳定性：Kafka 是分布式的，一个数据多个副本，某个节点宕机，Kafka 集群能够正常工作；
  • 持久性、可靠性、可回溯： Kafka 能够允许数据的持久化存储，消息被持久化到磁盘，并支持数据备份防止数据丢失，支持消息回溯；
  • 消息有序：通过控制能够保证所有消息被消费且仅被消费一次；
  • 有优秀的第三方 Kafka Web 管理界面 Kafka-Manager，在日志领域比较成熟，被多家公司和多个开源项目使用。

缺点：

  • Kafka 单机超过 64 个队列/分区，Load 会发生明显的飙高现象，队列越多，load 越高，发送消息响应时间变长；
  • 不支持消息路由，不支持延迟发送，不支持消息重试；
  • 社区更新较慢。

RocketMQ

优点：

  • 高吞吐：借鉴 Kafka 的设计，单一队列百万消息的堆积能力；
  • 高伸缩性：灵活的分布式横向扩展部署架构，整体架构其实和 kafka 很像；
  • 高容错性：通过ACK机制，保证消息一定能正常消费；
  • 持久化、可回溯：消息可以持久化到磁盘中，支持消息回溯；
  • 消息有序：在一个队列中可靠的先进先出（FIFO）和严格的顺序传递；
  • 支持发布/订阅和点对点消息模型，支持拉、推两种消息模式；
  • 提供 docker 镜像用于隔离测试和云集群部署，提供配置、指标和监控等功能丰富的 Dashboard。

缺点：

  • 不支持消息路由，支持的客户端语言不多，目前是 java 及 c++，其中 c++ 不成熟；
  • 部分支持消息有序：需要将同一类的消息 hash 到同一个队列 Queue 中，才能支持消息的顺序，如果同一类消息散落到不同的 Queue中，就不能支持消息的顺序。
  • 社区活跃度一般。

RabbitMQ

优点：

  • 支持几乎所有最受欢迎的编程语言：Java，C，C ++，C＃，Ruby，Perl，Python，PHP等等；
  • 支持消息路由：RabbitMQ 可以通过不同的交换器支持不同种类的消息路由；
  • 消息时序：通过延时队列，可以指定消息的延时时间，过期时间TTL等；
  • 支持容错处理：通过交付重试和死信交换器（DLX）来处理消息处理故障；
  • 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker；
  • 社区活跃度高。

缺点：

  • Erlang 开发，源码比较晦涩，不利于做二次开发和维护，基本职能依赖于开源社区的快速维护和修复 bug；
  • RabbitMQ 吞吐量会低一些，这是因为他做的实现机制比较重；
  • 不支持消息有序、持久化不好、不支持消息回溯、伸缩性一般。

消息队列选型

  • Kafka：追求高吞吐量，一开始的目的就是用于日志收集和传输，适合产生大量数据的互联网服务的数据收集业务，大型公司建议可以选用，如果有日志采集功能，肯定是首选 kafka。
  • RocketMQ：天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况。RoketMQ 在稳定性上可能更值得信赖，这些业务场景在阿里双 11 已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择 RocketMQ。
  • RabbitMQ：结合 erlang 语言本身的并发优势，性能较好，社区活跃度也比较高，但是不利于做二次开发和维护，不过 RabbitMQ 的社区十分活跃，可以解决开发过程中遇到的 bug。如果你的数据量没有那么大，小公司优先选择功能比较完备的 RabbitMQ。
  • ActiveMQ：官方社区现在对 ActiveMQ 5.x 维护越来越少，较少在大规模吞吐的场景中使用。

reference

https://zhuanlan.zhihu.com/p/486076939