RabbitMQ VS Apache Kafka （八）—— Kafka消息路由原语与路由保证

冷不防 2022-05-12 14:52 227阅读 0赞

Kafka的路由保证主要基于以下实现：

*  消息持久化：一旦消息存储到主题中，则确认不会出现丢失消息的问题
 *  消息确认：Kafka（或者是Zookeeper）与发布者、订阅者之间的消息通信

### 消息批量 ###

Kafka与RabbitMQ一个最大的不同之处在于Kafka支持在消息发送和处理时批量操作。当然，RabbitMQ也可以实现类似批量的操作：

*  每多少条消息暂停一次直至收到所有的消息确认。
 *  消费者设置预取阈值，通过**multiple**标识打包多个发送消息确认。

但消息并不是以批量的形式进行发送，而是通过消息流的持续发送并配合**multiple**标识打包多个发送消息确认，这其实是TCP所做的。

而Kafka则明确支持消息批量，使用批量的原因在于性能但这之间也是有一个权衡。RabbitMQ也会面对同样的权衡，比如，在途未确认消息越多，当失败发生时，也就意味着越多的消息重复或者处理重复。

在消费端，Kafka可以更有效的执行批量处理，因为Kafka是通过分区实现并发而不是通过竞争消费者的形式进行的。每一个分区对应一个消费者，所以大批量的使用并不影响工作的分布。但对于RabbitMQ来说，如果想实现类似效果，我们需要通过已经废弃的“拉”API来读取，但导致的结果很就是各竞争消费者之间的负载不均衡，RabbitMQ的设计初衷并适用于消息的批量处理。

### 持久化原语 ###

#### 日志复制 ####

Kakfa提供了日志分区级别的主从（领导者-追随者）架构以实现容错。每一个分区领导者都可以拥有多个追随者，当领导者所在的服务节点宕机，Kafka会从剩余的追求者选择一个提升为领导者。当服务短暂中断后，Kafka允许生产者和管理者来同通过一定的配置来确保已存储的消息不会丢失。当然，消息持久化时间跨度越久，Kafka的写延迟就会越长。

此外Kafka还提供了同步复制的概念（ISR），每一个副本都可以与领导者保持异步或同步状态，保持同步的含义是数据与领导者保持一致，误差时间不超过一定的时间周期（默认10秒）。当副本落后于领导者，那么副本就处于异步状态，导致异步状态的主要原因可能因为网络延迟，或者因为副本所在宿主主机发生故障。只有当领导者出现宕机掉线，且没有其他同步副本在线的场景才会导致信息丢失。关于这一点，我们后续章节中讨论。

#### 消息确认与偏移量跟踪 ####

##### 生产者消息确认 #####

当生产者发送消息给Kafka，它会通过Acks属性告诉Kafka代理期待哪种类型的消息确认。

*  无需确认，Acks=0
 *  领导者需持久化当前消息，Acks=1
 *  领导者和所有副本都需持久化当前消息，Acks=All

和RabbitMQ一样，采用消息确认可能会出现消息重复的问题。比如，Kafka代理收到生产者发送的消息，并对其进行了持久化操作，当Kafka准备发送确认消息给生产者，发送过程中出现了代理宕机或者网络中断，导致生产者无法收到消息确认，进而导致生产者必须再次重新发送消息。

但Kafka采用了一种的消息去重方式来避免了消息重复的问题，尽管这一方式有一定的局限性：

*  *enable.idempotence*设置为true
 *  *max.in.flight.requests.per.connection*限定为5或者更少
 *  *retries*标识设置为1或者更大
 *  *acks*设置为all

因此，如果批量消息个数超过5，或者acks=0或者1，那么你是无法使用Kafka的消息去重特性的。

#### 消费者偏移量跟踪 ####

消费者需要保存各自在Kafka日志中的偏移量，这样，即便消费者出现掉线或者错误时，新的消费者可以从失败的地方继续执行。偏移量通常被存储在Zookeeper中或者其他Kafka主题中。

一旦消费者需要从一个分区中批量读取消息，它可以采取如下方式来保存偏移量：

*  定时自动提交：客户端随着消息的处理定时提交偏移量，记录已处理消息的位置。从开发者角度来看，这种处理方式简单适用，并且性能良好。但当消息者出现错误或者失败时，增加重复路由的风险，原因在于消费者可能已经完成了批量消息的处理，但对应的消息偏移量却因消费者出现宕机或者失败导致提交失败。
 *  消息收到时主动提交：这种模式与之前我们讨论的*at-most-once*路由语义是一样，也就是说无论消息者是否会出现失败的情况，消息都将不会被处理两次，尽管有可能出现消息不被处理的情况。假设，有10条消息需要处理，但消费者在处理第5条消息时出现失败，那么实际情况就是只有4条消息被处理，剩余的其他消息将会被跳过执行，下一个消费者将会从本次批量之后的位置开始处理。
 *  结束之后主动提交：在所有的消息都被处理完毕，再执行主动提交，这种场景对应之前我们讨论的*at-least-once*路由语义，不管消费者是否会出现失败的场景，在这种模式下可以确保没有消息遗漏，尽管一条消息可能会被处理多次。比如，还是上面的例子，有10条消息需要处理，在第5条消息处理时，消费者失败，那么这10条消息会被重新读取并再次处理，那么之前已经被处理的4条记录会被处理两次。
 *  一次一提交：这种避免了消息重复，但严重影响吞吐效率。

仅一次路由保证仅限于[Kafka Streams][]，具体机制由Java库实现，如果使用Java语言开发，建议可以研究一下。仅一次的主要问题在于单个消息处理的输出与偏移量的保存需要在一个事务内完成。比如，如果消息处理的输出是发送邮件，那么我们就不能使用仅一次的消息处理业务场景。如果我们发送了邮件，但消费者执行偏移量保存的时候出现了错误，那么接下来新的消费者就会再次发送同样的邮件。

如果当前消息的输出作为另一个消息的输入，Kafka Streams仅一次的消息处理模式就有了实际的用武之地。在这样的业务场景中，我们可以通过Kafka的事务功能来完成消息的写入和偏移量的保存，即便消费者失败，我们也可以确保全部成功或者全部失败，不会导致重复出现消息输出。

#### 事务与隔离级别 ####

上面已经讨论，Kafka的事务功能主要用于“读-处理-写”的场景，事务可以跨多个主题和分区执行，生产者负责打开事务，执行批量消息写入，然后提交事务。

当消费者使用默认的*read uncommited*隔离级别时，无论事务是否提交或者中止，消费者都可以看到所有消息。当消费者使用*read commited*隔离级别时，消费者是无法读取到未提交或者中止的事务消息，只能处理已提交的事务消息。

或许，这里你会产生疑问，使用*read committed*隔离级别会不会影响路由顺序保证？答案是不会。消费者仍旧会以正确的顺序来读取消息，并且如果目标消息未提交，消费者会停止处理等待消息提交。因此，事务的使用会阻塞*read committed*消费者。LSO（Last Stable Offset）是第一打开事务偏移量之前的偏移量，实际运用中，由LSO负责标识*read committed*消费者可以读取的消息边界。

#### 总结 ####

RabbitMQ和Kafka均提供了可靠的、可持久化的消息路由，两者均提供了相对可靠的路由保证，然而，就我个人而言，考虑其幂等性，Kafka仍具一定的优势，即便实际运用中可能出现偏移量混乱，但这也不意味着消息永久丢失，事实上，消息仍旧保留着在分区中。

*  两者均提供了最多一次和最少一次的路由保证
 *  两者均提供消息复制功能
 *  在系统吞吐效率及消息重复方面，两者均实现了一种均衡。
 *  针对未确认在途消息数量，两者均有一定的控制
 *  两者均提供了消息路由顺序保证
 *  Kafka提供事务支持
 *  Kafka支持消息回溯，而RabbitMQ则会导致消息丢失
 *  Kafka提供了消息批处理，而RabbitMQ使用了竞争消费的推送模型，不适用于消息批量处理。

##### *原文链接* #####

[https://jack-vanlightly.com/blog/2017/12/15/rabbitmq-vs-kafka-part-4-message-delivery-semantics-and-guarantees][https_jack-vanlightly.com_blog_2017_12_15_rabbitmq-vs-kafka-part-4-message-delivery-semantics-and-guarantees]

[Kafka Streams]: https://kafka.apache.org/documentation/streams/
[https_jack-vanlightly.com_blog_2017_12_15_rabbitmq-vs-kafka-part-4-message-delivery-semantics-and-guarantees]: https://jack-vanlightly.com/blog/2017/12/15/rabbitmq-vs-kafka-part-4-message-delivery-semantics-and-guarantees