RabbitMQ 和 Kafka 的消息可靠性对比

Posted 2021-12-15 royfans

RabbitMQ和Kafka都提供持久的消息保证。两者都提供至少一次和至多一次的保证，另外，Kafka在某些限定情况下可以提供精确的一次（exactly-once）保证。

让我们首先理解一下上述术语的含义：

至多一次投递：消息绝对不会被重复投递，但是消息可能丢失

至少一次投递：消息绝对不会被丢失，但是有可能重复被消费

精确的一次投递：消息系统的圣杯。所有的消息精确的被投递一次。

“投递”貌似不是准确的语言描述，“处理”才是。无论怎么描述，我们关心的是，消费者能否处理消息，以及处理的次数。但是使用“处理”会使问题变得复杂。比如说，消息必须投递两次才能被处理一次。再比如，如果消费者在处理的过程中宕机，消息必须被第二次投递（给另一个消费者）。

其次，使用“处理”来表达会使得部分失败（partial failure）变得头疼。处理消息一般包括多个步骤。处理的开始到结束包括应用的逻辑以及应用与消息系统的通信。应用逻辑的部分失败由应用来处理。如果应用处理的逻辑是事务的，结果是all or nothing, 那么应用逻辑可以避免部分失败。但是实际上，多个步骤往往涉及不同的系统，使得事务性变得不可能。如果我们考虑到通信，应用，缓存，数据库，我们无法达到精确的一次处理（exactly-once processing）.

所以，精确地一次只出现在如下情况中：消息的处理只包括消息系统本身，并且消息系统本身的处理是事务的。在该限定场景下，我们可以处理消息，写消息，发送消息被处理的ACK， 一切都在事务中。而这正是Kafka流能提供的。

但是，如果消息处理是幂等（idempotent）的，我们就可以绕过基于事务的精确一次保证。如果消息处理是幂等的，我们可以安全的处理重复的消息。当然，并不是所有的消息处理都是幂等的。

 

责任链

本质上讲，生产者不能知道消息是否被消费。他们能知道的是，消息系统是否接收了消息，是否把消息安全的存储起来以便投递。这里存在一条责任链，开始于生产者，移动到消息系统，最后到达消费者。每个环节都要正确执行，环节间的交接也要正确执行。这意味着，作为一个应用开发者，你要正确的写程序，防止丢失消息，或者滥用消息。

消息顺序

这篇文章主要关注RabbitMQ和Kafka如何提供至少一次和至多一次的投递。但是，也包括消息的顺序。简单来讲，两者都支持FIFO顺序。RabbitMQ在队列这个层次，Kafka在话题的分区层次。

RabbitMQ

投递保证依赖于：

消息的持久性——一旦存储下来，就不会丢失

消息的ACK——RabbitMQ与生产者、消费者之间的信号

队列镜像

队列可以在节点间被镜像（复制）。对于每个队列，存在一个主队列，在单独一个节点上。假设我们有3个节点，10 个队列，每个队列2个镜像。那么10个主队列和20个镜像将分布在3个节点间。主队列如何分布是可以被配置的。当一个节点宕机后，

在宕机的节点上的每一个主队列，在另一个节点上的镜像队列会被提升为主队列

在其他节点上的镜像队列会被创建出来，以代替宕机的节点上的镜像队列，从而维护复制因子（replication factor）

持久的队列

RabbitMQ有两种队列：持久的和非持久的。持久的队列会被存储在磁盘上，节点重启后会重新构建出来。

持久的消息

持久的队列不能保证消息可以在宕机时被保留下来。只有被设定为持久的消息才会在宕机重启后恢复。

对于RabbitMQ，越多的消息是持久的，队列的吞吐率就越差。因此如果你有实时流，而且轻微的丢数据不会有大问题，那么你不该考虑队列镜像，并且消息应该设定为非持久的。然而，如果你必须不能在节点宕机时丢失数据，那么应该使用队列表镜像，持久的队列和持久的消息。

消息的ACK

消息发布

消息发布时，可能会被丢失或重复。这取决于生产者的行为。

Fire and Forget 发布者可以选择不使用生产者ACK，简单的发动消息弃之不顾。消息不会被复制，但是可能被丢失（至多一次投递）

发布确认：当发布者与中间人（broker）建立频道后，可以 设置该频道使用确认消息。则中间人会回复发布者的消息如下：

basic.ack:正ACK.消息已经收到，现在消息在RabbitMQ这边了。

basic.nack:负ACK.发生错误，消息未被处理。责任还在发布者。发布者可能需要重发。

除了以上两种，还有一种回复basic.return。有时发布者不仅需要知道中间人收到了消息，而且需要知道消息已经在若干队列中持久化了。比如，有时发布者发布了一条消息给交换机，但是交换机上没有绑定任何匹配的队列，那么中间人会简单的丢弃消息。大多数情况下，这没有问题，但是有时，发布者需要知道消息是被丢弃了还是被处理了。可以对每个消息设定mandatory标记，如此一来，如果消息没有被处理而是被丢弃，那么会返回一个basic.return

发布者可以选择发送每一条消息都等待ACK，但是会严重影响吞吐率。所以，发布者一般发布消息流，但是会限制未ACK的消息的数目。一旦达到了message in flight 的数目限制，发布者会暂停，等待ACK的到来。

现在，我们有了多条在途中的消息（在发布者与RabbitMQ之间），为了提高吞吐率，RabbitMQ使用multiple标记位来将ACK组成一组。如此一来，所有的消息会被分配一个单调递增的序列号（Sequence Number）。消息的ACK中会包含对应的序列号。当组合使用Multiple标记位时，发布者需要维护发送出去消息的序列号，以便它知道哪些消息被ACK。

所以，利用ACK，我们可以通过以下方法避免消息丢失：

当收到nack,重新发布消息。

当收到nack或者basic.return,将消息持久化到某地。

 

事务:在RabbitMQ中，并不常用事务。因为

不明确的保证：如果消息被路由到多个队列，或者起用了mandatory标记，那么事务的原子性是不可靠的。

性能比较差。

坦率的讲，我从未使用过事务，它增加了额外的保证，提高了不确定性。

连接/频道异常：除了消息的ACK外，发布者还需要考虑连接断开或者中间人出错，两者都会导致频道丢失。频道丢失会导致无法接收消息的ACK.在这个问题上，发布者可以考虑妥协，一种是冒消息丢失的风险一种是冒消息重复的风险。

如果中间人宕机，可能此时消息还在OS的buffer中，或者正在被解析，因此被丢失。又或者，这条消息已经持久化，正当中间人发送ACK时，宕机了，在这种情况下，其实消息已经成功投递了。

连接断开同样如此。我们无法得知宕机的具体时机，所以只能选择：

不重新发布，冒消息丢失的风险

重新发布，冒消息重复的风险

如果发布者有很多在途的消息，问题会恶化。一种方式是发布者提供提示，告诉消费者消息是重发的，让消费者尝试去重。

 

 

消费者

对于ACK，消费者有两种选择

无ACK模式

手动ACK模式。

无ACK模式：或者称为自动ACK模式，是危险的。首先，只要消息投递给应用层，就会被从队列中删除。这会导致消息丢失：

消息还在内部buffer中，但是应用层宕机

消息处理失败

其次，我们无法控制消息传递的速度。使用手动ACK，我们可以设定预取（QoS）值，来限制应用获得的未ACK的消息的数目。如果没有这个功能，RabbitMQ会很快的传递消息，超出消费者可以处理的讷讷管理，导致内部buffer溢出或内存问题。

手动ACK模式：消费者必须手动给出消息的ACK.消费者可以设定预取值大于一，便可以并行的处理多条数据。消费者可以选择单条消息的发送ACK，也可以设定multiple标记位，一次ACK多条消息。批处理会提高性能。

当消费者打开一个频道，被投递的消息会收到一个单调上升的整数值Delivery Tag。这个信息会包括在ACK当中作为消息的标识。

ACK有如下几种：

basic.ack.RabbitMQ会从队列中删除该条消息。可以使用multiple标记。

basic.nack。消费者需要告诉RabbitMQ是否需要重新将消息压入队列。重入队列意味着消息会被放在队列头，再次投递给消费者。也支持multiple标志位。

basic.reject.与basic.nack类似，但是不支持multiple标记位。

所以从语义上级讲，basic.ack与(basic.nack&requeue==false)是等价的。都会导致消息从队列中删除。

下一个问题是，什么时候发送ACK?如果消息处理很快，可以选择消息处理完再发送ACK.但是，如果消息处理需要几分钟，那么处理完再发送ACK是有问题的。如果频道宕机，所有未ACK的消息会重入队列，导致消息重复。

通信/频道 故障

如果通信故障，或者中间人故障导致频道宕机，那么所有的未ACK的消息都会重新入队列再次投递，这不会导致消息丢失，但是会导致消息重复。

消费者保持未ACK的消息越久，消息被重新投递的风险越高。当消息是被重投递时，消息会设置redelivered标志位。所以最坏情况下，至少消费者是可以知道消息是一条重发的消息。

幂等性

如果你需要幂等并且保证消息不会丢失，那么意味着你需要实现消息去重或其他幂等模式。如果消息去重非常耗时，那么你可以让发布者对重发的消息添加头数据，让消费者检查头数据和redelivered 标志位。

结论
RabbitMQ提供提供强大的，可靠地，持久的消息保证，但是，你有很多办法把它弄糟。

以下是一些注意事项

如果想要保证至少一次投递，使用队列镜像，持久的队列，持久的消息，发布者ACK，mandatory标志位，手动消费者ACK；

使用最少一次投递，你或许需要增加去重逻辑或者使用幂等范式

如果你不关心消息丢失，而更关注低延时和高度可扩展，那么你不需要使用队列镜像，持久的消息和发布者ACK.当然，我自己会保留使用手动消费者ACK,通过设定预取2值来控制消息投递的速度，当然，你需要设定multiple标志位并批量ACK.

 

Kafka

Kafka的投递通过如下保证：

消息持久性：一旦存入话题，消息不会丢失

消息ACK:kafka（或者包括Zookeeper）与生产者、消费者信号

关于批处理

Kaka和RabbitMQ有在消息批量发送、消费方面不同。RabbitMQ可以实现如下：

每发送x条消息就暂停，直到所有消息的ACK被收到。RabbitMQ通常将多条ACK组成一组，使用multiple标志位

消费者设定一个预取值，将消息的ACK组成一组

但是消息本身不是批量发送的，它更多的是指允许一组消息在途，使用multiple 标志位。这一点跟TCP很像。

而Kafka则有明确的消息批量处理。批处理可以提高性能，同时也需要权衡，正如RabbitMQ权衡在途的未ACK消息一样。越多的在途消息，会导致越严重的消息重复（当故障发生时）。

Kafka可以更高效的在消费者端进行批处理，因为kafka有分区的概念。每个分区对应一个消费者，所以及时一个很大的批处理也不会营子昂负载的分布。然而，对于RabbitMQ而言，如果使用已经被废弃的拉取API拉取批量的消息，会导致非常严重的负载不均衡。以及很长的处理延时。RabbitMQ在设计时就不适合批处理。

持久性

日志复制

为了容错，Kafka在分区层面有一个主从架构，主分区成为master,复制分区成为slave或者follower.每个master可以有很多follower.当主分区的服务器宕机后，follower中会有一份被提升为主分区，所以只会导致短暂的服务停止，但是不会导致数据丢失。

Kafka有一个概念，叫做In Sync Replicas(同步的复制)。每一个复制都可以是同步的，或是非同步的。同步意味着跟主分区相比，拥有相同的消息。复制可能会变成非同步的，如果它落后了。这可能是因为网络延迟，宿主机故障等。消息丢失只会发生在如下情况：主分区服务器宕机，所有的复制都是非同步的。

消息ACK与偏移追踪

取决于Kafka如何存储消息以及消费者如何消费消息，Kafka依赖于消息ACK来进行偏移追踪。

生产者的消息ACK

当生产者发送消息时，会告诉中间人何种期待ACK：

不需要ACK:fire and forget, 对应于acks = 0

主分区已经将消息持久化。 对应于acks=1

主分区以及所有同步的复制都将消息持久化， 对应于acks=ALL

消息可以在发布时被复制，正如RabbitMQ一样。如果中间人宕机或者网络故障，发布者会把没有收到ACK的消息重发。当然，大多数情况下，消息应该是被主分区持久化并复制了。

然而，Kafka有一个很好的去重的特性，但是必须如下设置：

enable.idempotence 设置成true

max.in.flight.requests.per.connection 低于5

retries设置1或更高

acks设置成ALL

在这种配置下，如果你为了吞吐率，批处理的单位设置成6或者acks设置成0/1,那么你就没办法获得去重。

消费者偏移追踪

消费者需要存储他们的偏移以备宕机，让另一个消费者接替。偏移存储在zookeeper上或者kafka的话题中。

一旦消费者从分区中读取一批量的消息，它有多种选择去更新偏移:

立即更新：在开始处理消息前。这对应于最多一次投递。无论消费者是否宕机，消息都不会被重复。比如10条正在被处理，此时消费者在第五条消息处理时宕机，那么只有前4条消息被处理，其余被跳过，接替的消费者从下一个批次开始。

最后更新。当所有消息都被处理后。这对应于至少一次投递。无论消费者是否宕机，没有消息会被丢失，尽管消息会被处理两次。比如10条消息正在被处理，当消费者在消费第五条消息时宕机，则整个10条消息会被接替的消费者再次处理。

精确地一次语义只有在使用Java Library Kafka Stream时被保证。如果你使用Java,我强烈推荐使用。精确一次语义的只要问题在于消息的处理和偏移的更新需要哎事务中完成。例如，如果消息处理是发送一条邮件的话，那么我们就无法完成精确的一次。例如我们发送玩邮件后，消费者宕机，我们可以更新偏移，但是会导致邮件再次被发送。

Kafka Stream 的Java 应用，将消息处理后生成新的消息不同的话题，那么这个应用将是满足精确一次语义的。因为我们可以使用Kafka的事务功能与写消息并更新偏移。

关于事务和隔离层次

Kafka中事务的应用主要是读-处理-写模式。事务可以跨越多个话题和分区。一个生产者打开一个事务，写一个批量的消息，然后提交事务。

当消费者使用默认的read uncommited 隔离级别时，消费者可以看到所有的消息，无论是提交的，未提交的，还是终止的。当消费者使用read committed隔离级别时，消费者不会看到未提交的或者终止的消息。

你可能比较疑惑，隔离级别如何影响消息顺序。答案是，不影响。消费者依旧按序读取消息。Last Stable Offset（LSO）之前的消息都会被读取。

总结

RabbitMQ和Kafka都提供可靠的，持久的消息系统，所以如果可靠性对你来说很重要，那么你大可放心，两者都是可靠的。当时，Kafka更胜一筹，因为提供幂等的发布，并且，及时错误的操作偏移，消息也不会丢失。

显然，没有十全十美的产品，但是只要应用正确的使用ACK,管理员正确的配置复制，并且你的数据中心没有轰然倒塌，你就可以放心，消息不会丢失。至于容错和可用性，也需要另外讨论。

下面是一些简单结论：

两者都提供至多一次和至少一次语义

两者都提供复制

两者对消息重复和吞吐率有相同的取舍。尽管kafka提供幂等的发布，但是仅限于一定的体量。

两者都可以控制在途的未ACK消息数量

两者都保证顺序

Kafka提供真正的事务操作，主要用于读-处理-写。尽管你需要注意吞吐率。

使用Kafka,及时消费者错误处理，但是可以使用偏移进行回退。RabbitMQ则不行。

Kafka基于分区的概念，可以使用批处理提高性能。而RabbitMQ不适合批处理，因为它基于推送模型，并且使用竞争的消费者。