流处理平台中的消息传递语义

1. 概述

在本教程中，我们将讨论流处理平台中的消息传递语义。

首先，我们将快速了解事件在流处理平台主要组件中的传输过程。接下来，我们将讨论此类平台中数据丢失和重复的常见原因。最后，我们将重点介绍三种主要的传递语义。

我们将讨论如何在流处理平台中实现这些语义，以及如何处理数据丢失和重复问题。

在每种传递语义中，我们将非常简要地介绍在Apache Kafka中获取传递保证的方法。

2. 流处理平台基础知识

简而言之，像Apache Kafka和Apache ActiveMQ这样的流平台以实时或近实时的方式处理来自一个或多个源(也称为生产者)的事件，并将它们传递到一个或多个目的地(也称为消费者)进行进一步处理、转换、分析或存储。

生产者和消费者通过代理分离，从而实现可扩展性。

流处理应用程序的一些用例可能是电子商务网站中的大量用户活动跟踪、实时金融交易和欺诈检测、需要实时处理的自主移动设备等。

消息传递平台有两个重要的考虑因素：

• 准确性
• 延迟

通常，在分布式实时系统中，我们需要根据系统更重要的因素在延迟和准确性之间进行权衡。

在这里，我们需要了解流处理平台提供的开箱即用的传递保证，或者使用消息元数据和平台配置实现所需的传递保证。

接下来，让我们简要地看一下流处理平台中的数据丢失和重复问题，然后我们将讨论如何管理这些问题的传递语义。

3. 可能的数据丢失和重复情况

为了理解流处理平台中的数据丢失和/或重复，让我们快速回顾一下流处理平台中的高级事件流：

在这里，我们可以看到从生产者到消费者的流程中可能存在多个故障点。

这常常会导致数据丢失、延迟和消息重复等问题。

让我们关注上图的每个组件，看看可能出现的问题及其对流处理系统可能造成的后果。

3.1 生产者故障

生产失败可能会导致一些问题：

• 生产者生成消息后，在通过网络发送之前可能会失败，这可能会导致数据丢失。
• 生产者在等待代理确认时可能会失败，当生产者恢复后，它会假设代理确认丢失，并尝试重新发送消息，这可能会导致代理上出现数据重复。

3.2 生产者和代理之间的网络问题

生产者和代理之间可能存在网络故障：

• 由于网络问题，生产者发送的消息可能永远无法到达代理。
• 还可能存在这样的情况：代理接收消息并发送确认，但由于网络问题，生产者从未收到确认。

在这两种情况下，生产者都会重新发送消息，从而导致代理处的数据重复。

3.3 代理故障

类似地，代理故障也可能导致数据重复：

• 代理可能会在将消息提交到持久化存储之后、向生产者发送确认之前发生故障，这可能会导致生产者重新发送数据，从而造成数据重复。
• 代理可能正在跟踪消费者迄今为止已读取的消息，代理可能在提交此信息之前发生故障，这可能导致消费者多次读取同一条消息，从而导致数据重复。

3.4 消息持久化问题

从内存状态将数据写入磁盘时可能会发生故障，从而导致数据丢失。

3.5 消费者和代理之间的网络问题

代理和消费者之间可能存在网络故障：

• 尽管代理发送了该消息并记录了它发送了该消息，但消费者可能永远不会收到该消息。
• 类似地，消费者可能会在收到消息后发送确认，但该确认可能永远不会到达代理。

在这两种情况下，代理可能会重新发送消息，导致数据重复。

3.6 消费者故障

• 消费者可能在处理消息之前失败。
• 消费者可能会在持久存储中记录其处理该消息之前失败。
• 消费者在记录其处理消息之后但在发送确认之前也可能会失败。

这可能导致消费者再次向代理请求相同的消息，从而造成数据重复。

接下来，让我们看看流处理平台中可用的传递语义，以处理这些问题并满足个别系统要求。

4. 传递语义

传递语义定义了流处理平台如何保证流处理应用程序中事件从源到目的地的传递。

有三种不同的传递语义可用：

• 最多一次
• 至少一次
• 恰好一次

4.1 最多一次传递

在这种方法中，消费者首先保存最后接收到的事件的位置，然后处理它。

简单来说，如果事件处理中途失败，消费者重启后就无法回去读取旧事件。

因此，无法保证所有接收到的事件都能成功处理。

对于某些数据丢失不是问题且准确性不是必需的情况，最多语义是理想的。

以Apache Kafka为例，它使用消息偏移量，最多一次保证的序列将是：

• 持久化偏移量
• 持久化结果

为了在Kafka中启用最多一次语义，我们需要在消费者处将“enable.auto.commit”设置为“true”。

如果发生故障并且消费者重新启动，它将查看最后一个持久化的偏移量。由于偏移量是在实际事件处理之前持久化的，因此我们无法确定消费者收到的每个事件是否都已成功处理，在这种情况下，消费者最终可能会丢失一些事件。

让我们形象化一下这个语义：

4.2 至少一次传递

在这种方法中，消费者处理接收到的事件，将结果保存在某处，然后最终保存最后接收到的事件的位置。

与最多一次不同，如果发生故障，消费者可以读取并重新处理旧事件。

在某些情况下，这可能会导致数据重复。让我们考虑这样一个示例：消费者在处理并保存事件后，但在保存最后一个已知事件位置(也称为偏移量)之前失败了。

消费者将重新启动并从偏移量处读取，此时，消费者会多次重新处理事件，因为尽管在故障之前消息已成功处理，但最后收到的事件的位置并未成功保存：

对于任何需要更新股票行情指示器或仪表以显示当前值的应用程序来说，这种方法都是理想的选择。然而，对于需要聚合准确性的用例(例如求和和计数器)，至少一次处理并不理想，主要是因为重复事件会导致错误的结果。

因此，在这种传递语义中，不会丢失任何数据，但可能会出现重新处理同一事件的情况。

为了避免多次处理同一个事件，我们可以使用幂等消费者。

本质上，幂等消费者可以多次消费一条消息，但只能处理一次。

以下方法的组合可实现幂等消费者至少一次传递：

• 生产者给每条消息分配一个唯一的messageId。
• 消费者在数据库中维护所有已处理消息的记录。
• 当新消息到达时，消费者会根据持久存储表中现有的messageId进行检查。
• 如果匹配，消费者将更新偏移量而不重新消费，发回确认，并有效地将消息标记为已消费。
• 如果事件尚未存在，则会启动数据库事务，并插入新的messageId。接下来，将根据所需的业务逻辑处理这条新消息，消息处理完成后，事务最终提交。

在Kafka中，为了确保至少一次语义，生产者必须等待来自代理的确认。

如果生产者没有收到来自代理的任何确认，则生产者会重新发送消息。

此外，由于生产者批量向代理写入消息，如果写入失败并且生产者重试，则批次中的消息可能会在Kafka中写入多次。

然而，为了避免重复，Kafka引入了幂等生产者的特性。

本质上，为了在Kafka中启用至少一次语义，我们需要：

• 在生产者端将属性“ack”设置为值“1”
• 在消费者端将“enable.auto.commit”属性设置为值“false”
• 将“enable.idempotence”属性设置为值“true”
• 将序列号和生产者ID附加到生产者发出的每条消息

Kafka代理可以使用序列号和生产者ID来识别主题上的消息重复。

4.3 恰好一次传递

这种传递保证类似于“至少一次”语义，首先，处理收到的事件，然后将结果存储在某个地方。如果发生故障或重启，消费者可以重新读取并重新处理旧事件。然而，与“至少一次”处理不同的是，任何重复的事件都会被丢弃而不被处理，从而实现“恰好一次”处理。

这对于任何重视准确性的应用程序来说都是理想的，例如涉及聚合的应用程序(如精确计数器)或任何其他需要仅处理一次事件且不会丢失的应用程序。

该序列如下：

• 持久化结果
• 持久化偏移量

让我们看看当消费者在处理事件后失败但没有保存偏移量时会发生什么情况，如下图所示：

我们可以通过以下方式消除一次性语义中的重复：

• 幂等更新：我们将结果保存在生成的唯一键或ID上，如果出现重复，生成的键或ID已经存在于结果集中(例如，数据库)，因此消费者可以移除重复项而无需更新结果。
• 事务更新：我们将批量保存结果，每个批次都需要一个事务开始和一个事务提交，因此一旦提交，事件就能成功处理。这里我们只是删除重复的事件，而不会进行任何结果更新。

让我们看看我们需要做什么才能在Kafka中启用恰好一次语义：

• 通过为每个生产者设置唯一的“transaction.id”值，启用幂等生产者和生产者的事务功能
• 通过将属性“isolation.level”设置为值“read_committed”来在消费者处启用事务功能

5. 总结

在本文中，我们讨论了流处理平台中使用的三种传递语义之间的差异。

在简要概述了流平台中的事件流之后，我们研究了数据丢失和重复问题。然后，我们了解了如何使用各种传递语义来缓解这些问题。接下来，我们研究了至少一次传递、最多一次传递以及最终的精确一次传递语义。