流筛选内部机制

一篇之前的文章介绍了流筛选，这是 RabbitMQ 3.13 中一项激动人心的新功能。在这篇文章中，我们将探讨流筛选的内部机制。了解其设计和实现将帮助您以最优的方式配置和使用流筛选以满足您的用例。

概念

流筛选的理念是在代理端提供第一层高效的过滤，而无需代理解释消息。这样，只需要流子集的消费者就不需要获取所有数据并自行处理所有过滤。这可以大幅减少传输到消费者的数据量。

通过过滤，可以将一个过滤器值与每条消息关联。它可以是地理信息，例如每条消息来自的世界区域，如下图所示

每条消息都可以关联一个过滤器值，例如其来源的世界区域。

因此，我们的流有 1 条AMER（绿色）消息、1 条APAC（深蓝色）消息、2 条EMEA（紫色）消息，然后是 2 条AMER消息。

消息发布

发布者可以将每个出站消息与其过滤器值关联起来

发布者为每个出站消息提供过滤器值。

在上图中，发布者发布了 1 条AMER（绿色）消息和 2 条EMEA（紫色）消息，这些消息将被添加到流中。

消息消费

当消费者订阅时，它可以指定一个或多个过滤器值，并且预期代理只分发具有此或这些过滤器值的消息。我们很快就会看到这在实践中有点不同，但这足以理解这些概念。

在下图中，顶部的消费者指定它只希望AMER（绿色）消息，代理只分发这些消息。中间的消费者使用EMEA消息，底部的消费者使用APAC消息也是如此。

消费者可以指定它只希望具有给定过滤器值的消息，代理将尽其所能相应地分发消息。

关于概念就介绍到这里，现在让我们来了解一下实现细节。

流的结构

我们需要了解流的结构才能理解流筛选的内部机制。流是一个目录，其中包含段文件。每个段文件都有一个关联的索引文件（用于知道在段文件的给定偏移量处将消费者附加到哪里，等等）。拥有多个“小”段文件比拥有整个流的一个大型单片文件更好：例如，删除“旧”段文件以截断流比删除大型文件的开头更有效和更安全。

段文件由包含消息的块组成。块中的消息数量取决于输入速率（高输入速率意味着块中有很多消息，低输入速率意味着块中消息较少）。块中的消息数量从几个（甚至 1 个）到几千个不等。

块有什么作用？块是流中的工作单元：它们用于复制，更重要的是，对于我们的主题，用于消费者传递。代理一次向消费者发送一个块，使用sendfile系统调用（将文件系统中的整个块发送到网络套接字，无需将数据复制到用户空间）。

下图说明了流的结构

流由段文件组成。段文件包含块，块包含消息。代理将块分发给消费者，而不是单个消息。

有了这些，让我们看看代理如何知道是否要分发块。

代理端过滤

假设我们有一个消费者只想要AMER（绿色）消息。当代理即将分发块时，它需要知道该块是否包含AMER消息。如果包含，它可以将块发送给消费者，如果不包含，代理可以跳过该块，继续下一个块，并重复此过程。

每个块都有一个标题，其中可以包含一个布隆过滤器，它告诉代理该块是否包含具有给定过滤器值的消息。一个布隆过滤器是一种节省空间的概率数据结构，用于测试元素是否为集合的成员。在我们的示例中，集合包含AMER、EMEA和APAC，元素是AMER。

下图说明了我们 3 个块的代理端过滤过程

代理端过滤。代理使用每个块中包含的布隆过滤器来判断它是否包含具有给定过滤器值的消息。布隆过滤器效率很高，但它可能会返回误报，因此代理可能会返回不包含请求的过滤器值的消息的块。

如上图所示，过滤器可能会返回误报，即不包含预期过滤器值的消息的块。这是正常的，因为布隆过滤器是概率性的。但它们不会返回误判：如果过滤器说没有AMER（绿色）消息，我们可以确定这是真的。我们必须忍受这种不确定性：我们有时可能会无缘无故地分发一些块，但这总比分发所有块要好。

可以肯定的是，消费者可能会收到它不需要的消息：看看我们左边的第一个块，它包含消费者请求的AMER（绿色）消息，但也包含EMEA（紫色）和APAC（深蓝色）消息。这就是为什么客户端也必须进行过滤的原因。

客户端过滤

代理在传递消息时处理第一级过滤，但由于传递的单元是块，因此消费者仍然可能收到它不需要的消息。因此，客户端也必须进行一些过滤，这显然必须与订阅时设置的过滤器值一致。

下图说明了一个只想要AMER（绿色）消息并且必须进行最后一步过滤的消费者

由于块可能包含不需要的消息，因此客户端也必须过滤掉消息。

让我们看看如何在应用程序代码中实现这一点。

API 示例

过滤不会造成侵入，可以作为横切关注点进行处理，最大程度地减少对应用程序代码的影响。以下是如何在使用流 Java 客户端（filterValue(Function<Message,String>)方法）声明生产者时设置提取消息过滤器值的逻辑。

Producer producer = environment.producerBuilder()
  .stream("invoices")
  .filterValue(msg -> msg.getApplicationProperties().get("region").toString())  
  .build();

在消费端，流 Java 客户端提供filter().values(String... filterValues)方法来设置过滤器值，以及filter().postFilter(Predicate<Message> filter)方法来设置客户端过滤逻辑。在声明消费者时必须调用这两种方法。

Consumer consumer = environment.consumerBuilder()
  .stream("invoices")
  .filter()
    .values("AMER")  
    .postFilter(msg -> "AMER".equals(msg.getApplicationProperties().get("region")))  
  .builder()
  .messageHandler((ctx, msg) -> {
    // message processing code
  })
  .build();

如您所见，过滤不会更改发布和消费代码，只会更改生产者和消费者的声明。

现在让我们看看如何以最适合用例的方式配置流筛选。

流筛选配置

关于流筛选的第一篇文章提供了一些数据（与不进行筛选相比，筛选后带宽节省约 80%）。流筛选的好处在很大程度上取决于用例：输入速率、过滤器值的基数和分布，以及过滤器大小。过滤器越大越好（错误率越低）。可以为块中使用的过滤器的尺寸设置 16 到 255 字节之间的值，默认为 16 字节。

流 Java 客户端提供filterSize(int)方法，用于在创建流时设置过滤器大小（它在内部设置stream-filter-size-bytes参数）。

environment.streamCreator()
  .stream("invoices")
  .filterSize(32)
  .create()

如何估算过滤器的尺寸？网上有很多布隆过滤器计算器可用。参数是散列函数的数量（RabbitMQ 流筛选为 2）、预期元素的数量、错误率和尺寸。您通常对元素的数量有所了解，因此您需要在错误率和过滤器尺寸之间找到平衡。

以下是一些示例

• 10 个值，16 字节 => 2% 错误率
• 30 个值，16 字节 => 14% 错误率
• 200 个值，128 字节 => 10% 错误率

那么，过滤器越大越好？不完全是：尽管布隆过滤器在存储方面非常有效，因为它不存储元素，只存储元素是否在集合中，但过滤器尺寸是预先分配的。如果您将过滤器尺寸设置为 255，并且每个块至少包含一条具有过滤器值的消息，则每个块标题中将分配 255 字节。如果块包含许多大型消息，这很好，因为与块尺寸相比，过滤器尺寸可以忽略不计。但是对于像单消息块（包含 10 字节的消息和 10 字节的过滤器值）这样的退化情况，您最终得到的过滤器尺寸将大于实际数据。

您需要根据自己的用例进行实验，以估算过滤器大小对流大小的影响。在关于流过滤的第一篇文章中提供了一个技巧，可以使用 Stream PerfTest 估算流的大小（读取整个流而不进行过滤，并查看rabbitmq_stream_read_bytes_total指标）。

额外内容：AMQP 上的流过滤

尽管访问流的首选方式是流协议，但其他协议（如 AMQP）也受支持。流过滤也支持任何 AMQP 客户端库。

• 声明：在声明流时，将x-queue-type参数设置为stream，并使用x-stream-filter-size-bytes设置过滤器大小。
• 发布：使用x-stream-filter-value头设置出站消息的过滤器值。
• 消费：使用x-stream-filter消费者参数设置预期的过滤器值（字符串或字符串数组），并可选地使用x-stream-match-unfiltered消费者参数接收不带任何过滤器值的邮件（默认值为false）。客户端过滤仍然是必要的！

总结

这篇博文深入描述了 RabbitMQ 3.13 中的流过滤。它补充了第一篇文章，该文章介绍了流过滤的使用和演示。

流过滤易于使用和受益，但了解内部机制对于优化其使用（尤其是在复杂用例中）很有帮助。请记住，客户端过滤是必要的，并且必须与配置的过滤器值一致。这通常很容易实现。还可以根据特定用例以最合适的方式设置过滤器大小。