连接到 Streams

RabbitMQ Streams 概述 介绍了 Streams，这是 RabbitMQ 3.9 中的一项新功能。本文介绍客户端应用程序应如何连接到 RabbitMQ 节点，以在 Streams 协议 使用时，从 Streams 中获得最大的好处。

Streams 针对高吞吐量场景进行了优化，因此数据局部性等技术细节对于充分发挥 RabbitMQ 集群的性能至关重要。客户端库可以处理大部分细节，但当设置涉及容器和负载均衡器等额外层时，对底层工作原理的基本理解至关重要。如果您想了解有关 Streams 的更多信息，并避免在部署第一个 Streams 应用程序时遇到一些麻烦，请继续阅读！

流的拓扑结构

流是复制的且持久化的，它由一个leader Erlang 进程和replica Erlang 进程组成。这些进程分布在 RabbitMQ 集群的节点上，如下图所示

流是复制的且持久化的数据结构。它有一个 leader 进程负责接收写入操作，以及可以向应用程序分发消息的 replica 进程。

只有 leader 进程处理写入操作，如入站消息。流的任何成员——leader 和 replica——都可以将消息分发给客户端应用程序。

使用 stream protocol 向流发布消息的应用程序可以连接到集群中的任何节点：消息会自动从连接节点传输到承载 leader 进程的节点。如果连接和流 leader 不在同一节点上，这个操作不是最优的，这是显而易见的。向流发布消息的应用程序应连接到承载流 leader 的节点，以避免额外的网络跳数。

对于消费应用程序来说，情况略有不同。使用 stream protocol，消息使用 sendfile 进行分发：包含消息的文件块直接从节点文件系统发送到网络套接字，无需经过用户空间。此优化要求消费应用程序连接的节点承载流的一个成员。该成员是 leader 还是 replica 并不重要，只要数据在文件系统上，就可以通过 sendfile 进行网络套接字传输。

对于消费应用程序来说，这个限制似乎并不太苛刻：在上图所示的示例中，每个节点都有一个流成员，因此应用程序可以连接到任何节点进行消费。但是，设想一个拥有 5 个节点的集群，其中流的复制因子为 2：每个流将只有 3 个节点中的 5 个节点上拥有成员。在这种情况下，消费应用程序必须适当地选择节点。

有了这些知识，让我们来尝试提出一些最佳实践。

行为良好的客户端

因此，发布应用程序可以连接到集群中的任何节点，它总是能到达 leader 进程。消费应用程序必须连接到承载目标流成员的节点，该成员可以是 leader 或 replica。客户端应用程序可以遵循这些规则，但以下最佳实践应尽可能强制执行：

• 发布应用程序应始终连接到承载目标流 leader 进程的节点.
• 消费应用程序应始终连接到承载目标流 replica 的节点.

下图说明了这些最佳实践

发布到流的客户端应用程序应连接到承载流 leader 的节点，从流消费的客户端应用程序应连接到承载该流 replica 的节点。

如前所述，直接连接到流 leader 所在的节点可以避免网络跳数，因为消息无论如何都要经过 leader 进程。那么，为什么在消费时总是使用 replica 呢？嗯，这只是为了分载 leader 进程，它已经忙于处理所有写入操作。

这些最佳实践最好由客户端库强制执行，所有这些技术细节都不应该泄露到应用程序代码中。接下来我们将了解 stream protocol 如何允许客户端应用程序发现给定流的拓扑结构。

集群中的流

在继续讨论 stream protocol 的 metadata 命令之前，让我们先弄清楚流（复数）如何在 RabbitMQ 集群的节点之间分布。我们提到一个流有一个位于某个节点上的 Erlang leader 进程，以及位于其他节点上的 replica Erlang 进程。对于多个流，所有这些 Erlang 进程（可以理解为“非常轻量级的线程”）分布在集群节点上，并且任何时候都不会期望给定的 RabbitMQ 节点承载所有流的 leader。

将流想象成 RabbitMQ 集群中的一个小集群，如下图中所示的多个流

流 leader 分布在集群的节点上。这意味着在任何时候，给定的节点不必包含所有 leader。

leader 在集群中的分布方式取决于创建流时的“leader 定位策略”。默认策略是“最少 leader”，这意味着为新流的 leader 选择具有最少流 leader 数量的节点。还有其他策略，但在此文章中不予介绍。

明确了这一点后，让我们继续讨论发现流拓扑结构的方法。

metadata 命令

stream protocol 提供了一个 metadata 命令 来查询一个或多个流的拓扑结构。对于每个查询的流，响应包含承载 leader 和 replica 的节点的 hostname 和 port。

下图说明了已连接到其中一个节点的客户端应用程序如何发现给定流的拓扑结构

客户端可以使用 metadata 命令发现流的拓扑结构。

因此，一种常见的模式是为客户端库提供一个或多个入口点，连接后使用 metadata 命令来了解目标流的拓扑结构，然后根据操作（发布或消费）连接到相应的节点。:

一旦客户端应用程序知道了流的拓扑结构，它就可以连接到相应的节点来处理它：连接到承载流 leader 的节点进行发布，连接到承载流 replica 的节点进行消费。

metadata 命令是客户端库强制执行上述最佳实践的关键。

不幸的是，默认返回的元数据并不总是准确的，或者至少对客户端应用程序连接来说不够准确。我们来看一个例子。

默认元数据的局限性

stream 插件返回每个节点的 hostname 作为 host 元数据（更准确地说，这是 Erlang 节点名称的主机部分，但在这里差别不大）。这没问题……只要客户端能够解析 hostname 来连接节点！

在当今容器化的时代，hostname 的概念可能变得模糊，一旦你脱离了容器环境，它就没有多大意义了。下图展示了一个 3 节点的 RabbitMQ 集群，这些节点是在不同虚拟机上运行的容器。客户端应用程序可以通过正确映射端口来连接到节点，但无法通过容器的 hostname 来连接。

在节点运行在容器中时，在元数据中使用 hostname 是行不通的，因为客户端和节点之间直接通信的可能性很小。

stream 插件尽力而为，但它也无能为力。幸运的是，可以配置节点在被问及 metadata 命令的“坐标”时返回什么值。

广告主机和端口

stream 插件的 advertised_host 和 advertised_port 配置项 可以被设置，以告知节点在被询问如何联系时应返回什么。这里没有什么窍门：插件只会信任进行了配置的操作员，并会盲目地返回这些值。这意味着客户端应用程序必须能够使用这些信息连接到节点。

如果默认值不合适，可以配置广告主机和端口。

如果您使用容器化节点部署 RabbitMQ 集群并使用流，advertised_host 和 advertised_port 设置应该可以解决客户端应用程序因不正确的流元数据而无法连接节点的所有问题。请始终牢记这一点。

在这种发现机制可能出现问题的一种常见用例仍然存在：当负载均衡器位于客户端应用程序和集群节点之间时。

使用负载均衡器

在 RabbitMQ 集群前面放置负载均衡器是一种常见场景。由于 streams 的工作方式，这可能会导致一些问题，但总有解决方案。如果我们回到 metadata 命令，但加上负载均衡器，事情就会出错：客户端会收到节点信息，并使用这些信息直接连接到节点，绕过负载均衡器。下图说明了这种不幸的情况

当负载均衡器位于客户端和节点之间时，元数据提示的用处不大。客户端应用程序将绕过负载均衡器并尝试直接连接到节点。这可能是不可行的，或者是一个安全隐患。

这可能不是你想要的。

那么，是否可以将 advertised_host 和 advertised_port 配置项设置为使用负载均衡器信息呢？这样客户端应用程序将始终连接到负载均衡器！这并不是一个好主意，因为我们将无法强制执行最佳实践（发布到 leader，消费到 replica），并且在流不在所有节点上的部署中，如果应用程序最终连接到一个没有流成员的节点，消费将会失败。

好吧，这有点令人沮丧，但打起精神来，因为客户端库可以实现一种变通方法来解决这个问题。

负载均衡器下的客户端变通方法

客户端应用程序仍然可以始终连接到负载均衡器，并最终连接到合适的节点。它如何做到这一点？两个步骤：

• 使用 metadata 命令，但忽略信息，始终连接到负载均衡器
• 重试连接，直到成功连接到合适的节点

你可能会想，一旦建立连接，如何找到节点呢？节点（hostname 和 port，或者配置的 advertised_host 和 advertised_port）的“坐标”在 stream protocol 连接中是可用的。因此，客户端应用程序可以知道它连接到了哪个节点。

这意味着在使用负载均衡器时，不应配置 advertised_host 和 advertised_port。 在这种情况下，metadata 命令返回的节点的“坐标”不会被用来连接，因为客户端总是连接到负载均衡器。它们用于关联负载均衡器提供的连接与客户端期望的节点，而 hostname 在此方面非常有用。

我们来看一个例子

• 发布应用程序通过 metadata 请求的响应，了解到其目标流的 leader 位于 node-1
• 它使用负载均衡器地址创建一个新连接
• 负载均衡器选择连接到 node-3
• 连接成功建立，但客户端应用程序发现它连接到了 node-3，它立即关闭连接，并重试
• 这次负载均衡器选择了 node-1
• 应用程序对它连接到的节点感到满意，它继续使用此连接进行发布

下图说明了这个过程

客户端可以选择忽略元数据提示，始终使用负载均衡器。由于流连接会传递它们源自的节点 hostname，因此客户端可以知道它是否连接到正确的节点，并保持连接或关闭连接并重试。

由于流连接是长期的，而且流应用程序通常不会有大量的连接变化，因此重试连接在这里不是问题。

此解决方案还假定负载均衡器不会总是连接到同一后端服务器。在此情况下，轮询是一种合适的平衡策略。

现在也应该清楚，在使用此技术时，设置 advertised_host 和 advertised_port 不是必需的，并且将它们设置为所有节点的负载均衡器坐标是一个坏主意。让每个节点返回其 hostname 在这里是可以的，因为 hostname 在网络中应该是唯一的。

因此，这取决于客户端库。现在让我们看看 stream Java 客户端如何实现这一点。

使用 Stream Java 客户端与负载均衡器

该 stream Java client 提供了一个 AddressResolver 扩展点。它在创建新连接时使用：从传入的 Address（基于 metadata 查询要连接的节点）开始，地址解析器可以提供一些逻辑来计算实际要使用的地址。默认实现只是返回给定的地址。如果您想在负载均衡器已部署的情况下实现上述变通方法，请始终返回负载均衡器的地址，如下面的代码片段所示：

Address entryPoint = new Address("my-load-balancer", 5552);
Environment environment = Environment.builder()
    .host(entryPoint.host())
    .port(entryPoint.port())
    .addressResolver(address -> entryPoint)
    .build();

该 stream PerfTest 工具 在启用 --load-balancer 选项时也支持此模式。以下是告诉该工具始终为发布者和消费者连接使用同一入口点的方法：

# with the Java binary
java -jar stream-perf-test.jar --uris rabbitmq-stream://my-load-balancer:5552 --load-balancer

# with Docker
docker run -it --rm pivotalrabbitmq/stream-perf-test --uris rabbitmq-stream://my-load-balancer:5552 --load-balancer

总结

本文介绍了在使用 stream protocol 时客户端应用程序应如何连接。要点如下：

• 发布应用程序应连接到承载目标流 leader 的节点
• 消费应用程序应连接到承载目标流 replica 的节点
• 客户端应用程序必须使用 metadata stream protocol 命令 来了解它们想要交互的流的拓扑结构
• stream Java 和 Go 客户端强制执行这些最佳实践
• metadata 命令默认返回节点的 hostname 和监听端口，这可能会有问题，例如在使用容器时
• advertised_host 和 advertised_port 配置项 允许指定节点在 metadata 命令中应返回的值
• 负载均衡器可能会误导客户端库，使其尝试绕过负载均衡器直接连接到节点
• 客户端库可以提供一种变通方法来与负载均衡器正常工作
• stream Java 和 Go 客户端实现了这种变通方法