关键要点
- 分布式消息传递系统支持两种类型的语义：流式传输和排队。每个都最适合某些类型的用例。
- Apache Pulsar 的独特之处在于它同时支持流式处理和排队用例。
- Pulsar 的多层架构允许用户比其他消息传递系统更方便地扩展主题的数量和大小。
- Pulsar 提供了可扩展性、可靠性和功能的适当平衡，以取代 Iterable 的 RabbitMQ，并最终取代其他消息传递系统，如 Kafka 和 Amazon SQS。
在 Iterable，我们每天代表客户发送大量营销信息。这些包括电子邮件、推送、SMS 和应用内消息。Iterable 还每天处理更多的用户更新、事件和自定义工作流状态，其中许多可以触发系统中的其他操作。这导致系统不仅对我们的客户非常有用，而且非常复杂。随着我们客户群的增长，管理这种复杂性变得更加重要。
Iterable 管理复杂性的一种方法是在其架构的几个部分中使用分布式消息传递系统。分布式消息系统的主要目的是存储需要消费者处理的消息，并在处理这些消息时跟踪这些消费者的状态。这样，消费者可以专注于处理每条消息的任务。
Iterable 使用工作队列方法来执行客户指定的营销工作流、webhook 和其他类型的作业调度和处理。其他组件，例如用户和事件摄取，使用流模型来处理有序的消息流。
通常，分布式消息传递系统支持两种类型的语义：流式传输和排队。每个都最适合某些类型的用例。
流式传输和排队
在流式消息系统中，生产者将数据附加到一组仅附加的消息流中。在每个流中，消息必须按特定顺序处理，并且消费者在流中标记它们的位置。可以使用某种策略（例如散列用户 ID）对消息进行分区以允许更大的并行性，并且每个分区都充当单独的数据流。因为每个流中的数据都是不可变的，并且只存储了偏移量条目，所以消息可能不会被跳过。流式处理在消息顺序很重要的情况下效果很好，例如数据摄取。Kafka和Amazon Kinesis是使用流语义来消费消息的消息系统示例。
在队列消息系统中，生产者将消息发送到可能由多个消费者共享的队列。消费者在收到消息时对其进行处理，并在处理每条消息时向排队系统发送确认。因为多个消费者可能共享一个队列并且消息序列并不重要，所以通常更容易扩展基于队列的系统的消费者端。排队系统非常适合不需要按特定顺序执行任务的工作队列——例如，向多个收件人发送一封电子邮件。RabbitMQ和Amazon SQS是流行的基于队列的消息系统的示例。
排队系统通常包括简化处理消息级错误任务的功能。例如，发生错误后，RabbitMQ 可以轻松地将消息传输到特殊队列，在该队列中保留指定的时间，然后返回到原始队列进行重试。它还可以否定地确认一条消息，以便在失败后重新传递它。由于大多数消息队列在确认消息后通常不会将消息存储在积压中，因此调试和灾难恢复更加困难，因为没有要检查的消息。
像 Kafka 这样的基于流的系统可用于排队用例，但有一些注意事项。事实上，许多用户选择此选项是因为这些系统通常提供卓越的性能。然而，这种解决方案可能是一个挑战，因为它给开发人员带来了过度的负担，无法处理严格的流排序所带来的限制。如果消费者消费消息的速度很慢或需要在暂时失败后重试处理，则同一流上其他消息的处理可能会延迟。一个常见的解决方案是通过将消息重新发布到另一个主题来重试处理，但这会引入复杂性，因为应用程序逻辑必须管理额外的状态。
为什么 Iterable 需要一个新的消息传递平台
我们一直在大量使用 RabbitMQ，并依靠它的特性来处理内部消息传递。我们大量使用生存时间 (TTL) 值，不仅用于固定长度的重试，还用于在消息处理中实现显式延迟。例如，我们可能会延迟发送营销电子邮件，以便营销信息可以在每个收件人最有可能打开的时间发送给他们。我们还依靠否定确认来重试排队的消息。
这是我们架构的简化版本：
当我们开始评估 Pulsar 时，上面提到的所有队列都在 RabbitMQ 上，除了使用 Kafka 的摄取。Kafka 非常适合摄取，因为它提供了必要的性能和排序保证。Kafka 不适合其他用例，因为它缺乏必要的工作队列语义。我们使用了许多 RabbitMQ 特有的特性，比如延迟，这一事实也使得寻找替代方案变得更具挑战性。
当我们扩展我们的系统时，RabbitMQ 开始出现以下限制：
- 在高负载下，RabbitMQ 经常遇到流量控制问题。流控制是一种在消息代理无法跟上时减慢发布者的机制，通常是因为内存和其他资源限制。这阻碍了生产者发布的能力，从而导致其他领域的服务延迟和请求失败。具体来说，我们注意到当大量消息的 TTL 同时过期时，流控制更频繁地发生。在这些情况下，RabbitMQ 试图一次性将过期消息传递到它们的目标队列。这使 RabbitMQ 实例的内存容量不堪重负，从而触发了正常生产者的流控机制，阻止了他们的发布尝试。
- 调试变得更加困难，因为 RabbitMQ 的代理在消息被确认后不存储消息。换言之，无法为消息设置保留时间。
- 复制很难实现，因为 RabbitMQ 中的复制组件对于我们的用例来说不够健壮，导致 RabbitMQ 成为我们消息状态的单点故障。
- RabbitMQ 难以处理大量队列。由于我们有许多需要专用队列的用例，我们通常一次需要超过 10,000 个队列。在这个级别，RabbitMQ 遇到了性能问题，通常首先出现在管理界面和 API 中。
评估 Apache Pulsar
总的来说，Apache Pulsar似乎提供了我们需要的所有功能。虽然我们在 Pulsar 周围看到的很多宣传都将其与 Kafka 进行流式工作负载进行比较，但我们也发现 Pulsar 非常适合我们的排队需求。Pulsar 的共享订阅功能允许将主题用作队列，可能为同一主题中的不同订阅者提供多个虚拟队列。Pulsar 还原生支持延迟和预定消息，尽管这些功能在我们开始考虑 Pulsar 时是非常新的。
除了提供丰富的功能集之外，Pulsar 的多层架构让我们能够比其他消息传递系统更方便地扩展主题的数量和大小。
Pulsar 的顶层由代理组成，它们接受来自生产者的消息并将它们发送给消费者，但不存储数据。单个代理处理每个主题分区，但代理可以轻松交换主题所有权，因为它们不存储主题状态。这使得添加代理以增加吞吐量并立即利用新代理变得容易。这也使 Pulsar 能够处理代理故障。
Pulsar 的底层BookKeeper将主题数据存储在分段中，这些分段分布在整个集群中。如果需要额外的存储空间，我们可以轻松地将 BookKeeper 节点（bookies）添加到集群中，并使用它们来存储新的主题段。经纪人与博彩公司协调，以在每个主题发生变化时更新它的状态。Pulsar 将 BookKeeper 用于主题数据也有助于它支持大量主题，这对于 Iterable 当前的许多用例至关重要。
在评估了几个消息传递系统之后，我们认为 Pulsar 提供了可扩展性、可靠性和功能之间的适当平衡，以取代 Iterable 的 RabbitMQ，并最终取代其他消息传递系统，如 Kafka 和 Amazon SQS。
第一个 Pulsar 用例：消息发送
Iterable 平台最重要的功能之一是代表 Iterable 的客户安排和发送营销电子邮件。为此，我们将消息发布到客户特定的队列，然后让另一个服务处理消息的最终呈现和发送。这些队列是我们决定从 RabbitMQ 迁移到 Pulsar 的第一件事。
我们选择营销信息发送作为我们的第一个 Pulsar 用例有两个原因。首先，因为发送包含了我们一些更复杂的 RabbitMQ 用例。其次，因为它代表了我们 RabbitMQ 使用的很大一部分。这不是风险最低的用例；然而，经过广泛的性能和可扩展性测试，我们认为这是 Pulsar 可以增加最大价值的地方。
以下是在 Iterable 平台上创建的三种常见活动类型：
- 同时向所有收件人发送营销信息的爆炸式营销活动。假设客户想要向过去一个月活跃的用户发送电子邮件通讯。在这种情况下，我们可以在计划活动时查询ElasticSearch的用户列表，并将它们发布到该客户的 Pulsar 主题。
- 为每个收件人指定自定义发送时间的 Blast 营销活动。发送时间可以是固定的（例如，“收件人当地时区的上午 9 点”），也可以由我们的发送时间优化功能计算得出。在每种情况下，我们都希望将排队消息的处理延迟到指定时间。
- 用户触发的广告系列。这些可以由自定义工作流程或用户发起的交易（例如在线购买）触发。用户触发的营销发送是根据需要单独完成的。
在上述每种情况下，在任何给定时间执行的发送数量可能会有很大差异，因此我们还需要能够向上和向下扩展消费者以适应不断变化的负载。
迁移到 Apache Pulsar
尽管 Pulsar 在负载测试中表现良好，但我们不确定它是否能够在生产中维持高负载水平。这是一个特别关注的问题，因为我们计划利用 Pulsar 的几个新功能，包括否定确认和预定消息传递。
为了建立我们的信心，我们实现了一个并行管道，在该管道中我们向 RabbitMQ 和 Pulsar 发布消息；在这种情况下，我们在这些主题上设置消费者来确认排队的消息而不实际处理它们。我们还模拟了消费延迟。这有助于我们了解 Pulsar 在特定生产环境中的行为。我们对测试主题和实际生产主题都使用了客户级别的功能标志，因此我们可以逐一迁移客户以进行测试，并最终用于生产用途。
在测试过程中，我们发现了 Pulsar 中的一些错误。例如，我们发现了与延迟消息相关的竞争条件，Pulsar 开发人员帮助识别和修复了这种情况。这是我们发现的最严重的问题，因为它导致消费者卡住，造成未消费消息的积压。
我们还注意到一些与 Pulsar 消息批处理相关的有趣问题，这些问题在 Pulsar 生产者中默认启用。例如，我们注意到 Pulsar 的积压指标报告的是批次数，而不是实际的消息数量，这使得为消息积压设置警报阈值更具挑战性。后来我们在否定确认和批处理之间的交互中发现了一个更严重的错误，最近已经修复了。最终，我们认为批处理不值得麻烦。幸运的是，在 Pulsar 生产者中禁用批处理很容易，并且没有批处理的性能足以满足我们的需求。这些问题也可能在即将发布的版本中得到修复。
延迟和否定确认在当时是相对较新的功能，因此我们预计我们可能会发现一些问题。这就是为什么我们选择在几个月内缓慢迁移到 Pulsar，最初发布仅用于测试主题，然后逐渐迁移真实发送。这种方法使我们能够在问题成为我们客户的问题之前发现问题。尽管花了大约六个月的时间才完全相信 Pulsar 正在按预期工作，但结果值得花时间。
在大约六个月的时间里，我们将整个营销发送操作迁移到了 Pulsar。迁移完成后，我们发现 Pulsar 将我们的运营成本降低了近一半，并且在我们增加新客户时还有增长空间。成本降低显着，部分原因是我们的 RabbitMQ 实例被过度配置以弥补性能问题。迄今为止，我们的 Pulsar 集群已经顺利运行了六个多月，没有出现任何问题。
实施和工具
Iterable 主要在后端使用Scala，因此为 Pulsar 提供良好的 Scala 工具对我们来说很重要。我们使用了优秀的pulsar4s库，并做出了许多支持新功能的贡献，例如延迟消息。我们还提供了一个基于 Akka Streams 的连接器，用于将消息用作源，并提供单独的确认支持。
例如，我们可以像这样使用命名空间中的所有主题：
// Create a consumer on all topics in this namespace
val createConsumer = () => client.consumer(ConsumerConfig(
topicPattern = "persistent://email/project-123/.*".r,
subscription = Subscription("email-service")
))
// Create an Akka streams `Source` stage for this consumer
val pulsarSource = committableSource(createConsumer, Some(MessageId.earliest))
// Materialize the source and get back a `control` to shut it down later.
val control = pulsarSource.mapAsync(parallelism)(handleMessage).to(Sink.ignore).run()- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
我们喜欢为消费者使用正则表达式订阅。它们使在创建新主题时自动订阅它们变得容易，并使消费者不必了解特定的主题分区策略。同时，我们也在利用 Pulsar 支持大量主题的能力。由于 Pulsar 会在发布时自动创建新主题，因此为新消息类型甚至单个活动创建新主题很简单。这也使得为不同的客户和消息类型实施速率限制变得更加容易。
我们学到了什么
由于 Pulsar 是一个快速发展的开源项目，我们遇到了一些挑战——主要是在加快速度和学习它的怪癖方面——我们可能不会在其他更成熟的技术中看到这些挑战。文档并不总是完整的，我们经常需要向社区寻求帮助。也就是说，社区非常热情和乐于助人，我们很高兴更多地参与 Pulsar 的开发并参与有关新功能的讨论。
Pulsar 的独特之处在于它同时支持流式传输和排队用例，同时还支持广泛的功能集，使其成为我们架构中当前使用的许多其他分布式消息传递技术的可行替代方案。Pulsar 涵盖了我们所有的 Kafka、RabbitMQ 和 SQS 用例。这使我们能够专注于围绕单个统一系统构建专业知识和工具。
自 2019 年初开始与 Pulsar 合作以来，我们一直对 Pulsar 的开发进展感到鼓舞，尤其是在初学者的准入门槛方面。工具有了很大的改进：例如，Pulsar Manager现在提供了一个非常方便的 GUI 来管理集群。我们还看到许多公司提供托管和托管的 Pulsar 服务，这使初创公司和小型团队更容易开始使用 Pulsar。
总体而言，Iterable 向 Pulsar 的过渡很有趣，有时也很有挑战性，但到目前为止相当成功。在许多方面，我们的用例代表了一条尚未被广泛采用的新路径。我们预计会遇到一些问题，但我们的测试过程有助于最大限度地减少它们对客户的影响。我们现在对使用 Pulsar 充满信心，并将继续扩展我们对 Pulsar 的使用，用于 Iterable 平台中的其他现有和新组件。