使用 Rails 和 Kafka 构建面向服务的架构

这篇博文改编自 Stella Cotton 在 RailsConf 2018 上的演讲，题为“所以你已经拥有了一个 Kafka ”。

近年来，将软件设计为一系列服务（而非单一的、庞大的代码库）已成为构建应用程序的流行方式。在本文中，我们将学习 Kafka 的基础知识，以及如何使用其事件驱动的流程来支持您的 Rails 服务。我们还将讨论事件驱动的 Rails 服务在监控和扩展方面可能面临的实际考虑和运营挑战。

什么是 Kafka？

假设您想了解更多关于用户在平台上的互动情况：他们访问的页面、点击的按钮等等。一个足够受欢迎的应用程序可能会产生数十亿个事件，而将如此大量的数据发送到分析服务至少可以说是一项挑战。

Kafka是Web 应用程序不可或缺的组件，它需要实时数据流。Kafka 在生产者（生成事件）和消费者（读取事件）之间提供容错通信。任何单个应用程序中都可以有多个生产者和消费者。在 Kafka 中，每个事件都会在配置的时间内持久保存，因此多个消费者可以反复读取同一个事件。Kafka 集群由多个代理（broker）组成，代理（broker）只是运行 Kafka 实例的别称。

Kafka 的一个关键性能特征是它能够处理极高的事件吞吐量。传统的企业队列系统（例如 AMQP）会使用事件基础设施本身来跟踪每个消费者已处理的事件。随着消费者数量的增加，该基础设施将承受更大的负载，因为它需要跟踪越来越多的状态。甚至与消费者达成协议也并非易事。代理是否应该将"done"通过网络发送的消息标记为“已发送”？如果消费者宕机并需要代理重新发送事件，会发生什么？

另一方面，Kafka Broker 不跟踪任何消费者。消费者服务本身负责告诉 Kafka 它在事件处理流中的位置，以及它需要从 Kafka 获得什么。消费者可以从中间开始，向 Kafka 提供要读取的特定事件的偏移量，也可以从最开始甚至最末尾开始。消费者读取事件数据的时间复杂度为 O(1)；随着更多事件的到来，从流中查找信息的时间量不会改变。

Kafka 还具有良好的可扩展性和容错性。它以集群形式运行在一台或多台服务器上，可以通过添加更多服务器实现水平扩展。数据本身会被写入磁盘，然后复制到多个代理服务器中。至于可扩展性的具体数字，Netflix、LinkedIn 和 Microsoft 等公司每天通过 Kafka 集群发送的消息量都超过一万亿条！

在 Rails 中设置 Kafka

Heroku 提供了一个 Kafka 集群插件，可在任何环境中使用。对于 Ruby 应用，我们建议在实际用例中使用ruby-kafka gem。最低限度的实现只需要您提供代理的主机名：

# config/initializers/kafka_producer.rb
require "kafka"

# Configure the Kafka client with the broker hosts and the Rails
# logger.
$kafka = Kafka.new(["kafka1:9092", "kafka2:9092"], logger: Rails.logger)

# Set up an asynchronous producer that delivers its buffered messages
# every ten seconds:
$kafka_producer = $kafka.async_producer(
  delivery_interval: 10,
)

# Make sure to shut down the producer when exiting.
at_exit { $kafka_producer.shutdown }

设置好 Rails 初始化程序后，就可以开始使用 gem 发送事件负载了。由于发送事件是异步的，我们可以在 Web 执行线程之外编写事件，如下所示：

class OrdersController < ApplicationController
  def create
    @comment = Order.create!(params)

    $kafka_producer.produce(order.to_json, topic: "user_event", partition_key: user.id)
  end
end

我们将在下文中详细讨论 Kafka 的序列化格式，但在本例中，我们使用的是经典的 JSON 格式。topic关键字参数指的是 Kafka 将要写入事件的日志。主题本身被划分为多个分区，这使得您可以将特定主题中的数据“拆分”到多个代理中，以实现可扩展性和可靠性。每个主题最好有两个或更多分区，这样即使一个分区发生故障，您的事件仍然可以被写入和消费。Kafka 保证事件在分区内按顺序传递，但不保证在整个主题内按顺序传递。如果事件的顺序很重要，则传入一个分区键 (partition_key) 将确保特定类型的所有事件都进入同一个分区。

Kafka 为您的服务

Kafka 的一些特性使其在事件管道系统中具有很高的价值，也使其成为服务间 RPC 的一个非常有趣的容错替代方案。让我们以一个电商应用程序为例来说明 Kafka 在实践中的意义：

def create_order
  create_order_record
  charge_credit_card # call to Payments Service
  send_confirmation_email # call to Email Service
end

假设用户下单时，create_order会执行此方法。它会创建订单记录，从用户的信用卡扣款，并发送确认邮件。最后两个步骤已提取到服务中。

这种设置的一个挑战是，上游服务需要负责监控下游的可用性。如果电子邮件系统出现问题，上游服务需要了解该电子邮件服务是否可用。如果不可用，它还需要负责重试任何失败的请求。Kafka 的事件流在这种情况下能起到什么作用呢？我们来看一下：

在这个面向事件的世界中，上游服务可以向 Kafka 写入一个事件，表明订单已创建。由于 Kafka 具有“至少一次”保证，该事件至少会被写入 Kafka 一次，并且可供下游消费者读取。如果电子邮件服务宕机，该事件仍然会持久化，供下游消费者使用。当下游电子邮件服务恢复在线时，它可以继续按顺序处理错过的事件。

面向 RPC 的架构面临的另一个挑战是，在日益复杂的系统中，集成新的下游服务也意味着更改上游服务。假设您想集成一个新服务，该服务在订单创建时启动履行流程。在 RPC 世界中，上游服务需要向您的新履行服务添加新的 API 调用。但在面向事件的世界中，您需要在履行服务中添加一个新的消费者，该消费者在 Kafka 内部创建事件后便会消费该订单。

将事件纳入面向服务的架构

在一篇题为“事件驱动”的博客文章中，Martin Fowler 探讨了围绕“事件驱动应用”的困惑。当开发人员讨论这些系统时，他们实际上可能谈论的是截然不同的应用程序。为了就事件驱动系统达成共识，他开始定义一些架构模式。

让我们快速看看这些模式是什么！如果你想了解更多，可以看看他在 2017 年芝加哥 GOTO 大会上的主题演讲，其中对这些模式进行了深入的讲解。

事件通知

Fowler 谈到的第一个模式叫做事件通知。在这个场景中，一个服务只需用最少的信息通知下游服务发生了一个事件：

{
  "event": "order_created",
  "published_at": "2016-03-15T16:35:04Z"
}

如果下游服务需要有关发生了什么的更多信息，它将需要向上游进行网络调用来检索它。

事件携带状态转移

第二种模式称为事件携带状态转移。在这种设计中，上游服务会使用附加信息来扩充事件，以便下游消费者可以保留该数据的本地副本，而不必进行网络调用来从上游服务获取它：

{
  "event": "order_created",
  "order": {
    "order_id": 98765,
    "size": "medium",
    "color": "blue"
  },
  "published_at": "2016-03-15T16:35:04Z"
}

事件源

Fowler 提出的第三个概念是事件源架构。这种实现意味着，服务之间的每一次通信不仅都由事件启动，而且通过存储事件的表示，即使删除所有数据库，仍然可以通过重放该事件流来完全重建应用程序的状态。换句话说，每个有效负载都封装了系统在任何时刻的精确状态。

这种方法面临的巨大挑战是，随着时间的推移，代码会发生变化。未来对下游服务的 API 调用可能会返回与之前不同的一组数据，这使得重新计算当时的状态变得非常困难。

命令查询责任分离

最后提到的模式是命令查询职责分离（CQRS）。其理念是，您可能需要对记录执行的操作（创建、读取、更新）被拆分到不同的域中。这意味着一个服务负责写入，另一个服务负责读取。在面向事件的架构中，您经常会在图中看到事件系统嵌套在实际写入命令的位置。

写入服务将从事件流中读取数据，处理命令，并将其存储到写入数据库中。任何查询都发生在只读数据库上。将读写逻辑分离到两个不同的服务会增加复杂性，但它确实允许您分别针对这两个系统进行性能优化。

实际考虑

让我们讨论一下将 Kafka 集成到面向服务的应用程序中时可能会遇到的一些实际考虑。

首先要考虑的是缓慢的消费者。在事件驱动系统中，您的服务需要能够像上游服务生成事件一样快速地处理事件。否则，它们会慢慢落后，没有任何迹象表明存在问题，因为不会出现任何超时或调用失败。您可以识别超时的一个地方是与 Kafka 代理的套接字连接。如果服务处理事件的速度不够快，该连接可能会超时，并且重新建立连接会产生额外的时间成本，因为创建这些套接字的成本很高。

如果消费者速度慢，该如何加速？对于 Kafka，可以增加消费者组中的消费者数量，以便并行处理更多事件。每个服务至少需要运行两个消费者进程，这样如果一个进程宕机，其他故障分区都可以重新分配。本质上，只要有主题分区，就可以在尽可能多的消费者之间并行处理工作。（与任何扩展问题一样，您不能无限地添加消费者；最终，您将达到共享资源（例如数据库）的扩展极限。）

拥有关于事件添加到队列后进度的指标和警报也非常有价值。ruby-kafka 内置 ActiveSupport 通知，但它也自动包含 StatsD 和 Datadog 报告器。您可以使用它们来报告事件添加时是否落后。ruby-kafka 甚至提供了一系列推荐的监控指标！

使用 Kafka 构建系统的另一个方面是设计消费者以应对故障。Kafka 保证至少发送一次事件；绝不会出现消息完全不发送的情况。但是，您需要设计消费者以应对重复事件。实现这一点的一种方法是始终依赖于UPSERT向数据库添加新记录。如果已经存在具有相同属性的记录，则该调用本质上是无操作的。或者，您可以为每个事件添加一个唯一标识符，这样就可以跳过对之前已经出现过的事件的操作。

有效载荷格式

Kafka 令人惊讶的一点是它对数据非常宽容。你可以以字节为单位发送任何数据，它都会直接将其发送回消费者，无需任何验证。这个特性使其使用起来非常灵活，因为你无需遵循特定的格式。但是，如果上游服务决定更改其生成的事件，会发生什么情况呢？如果你只是更改事件负载，那么很有可能你的下游消费者会崩溃。

在开始采用事件驱动架构之前，请选择一种数据格式，并评估它如何帮助您注册模式并随着时间的推移不断演进。在实际实施模式之前，考虑其验证和演进会更容易。

当然，可以使用的一种格式是 JSON，即 Web 格式。它易于人类阅读，并且基本上所有编程语言都支持它。但它也有一些缺点。首先，JSON 负载的实际大小可能非常大。负载需要您发送键值对，这些键值对虽然灵活，但通常会在每个事件中重复。负载内部没有内置文档，因此，即使给定一个值，您可能也不知道它的含义。模式演进也是一个挑战，因为如果您需要重命名字段，则没有内置支持将一个键别名化为另一个键。

Apache Kafka 的开发团队 Confluent 推荐使用Avro作为数据序列化系统。数据以二进制形式发送，因此人类无法读取。但其优点在于，它对模式的支持更加强大。完整的 Avro 对象包含其模式和数据。Avro 支持简单类型（例如整数）和复杂类型（例如日期）。它还将文档嵌入到模式中，方便您理解字段在系统中的作用。它提供内置工具，帮助您随着时间的推移以向后兼容的方式改进模式。

avro-builder是 Salsify 开发的一款 gem，它提供了一种非常 Ruby 风格的 DSL 来帮助你创建模式。如果你想了解更多关于 Avro 的知识，Salsify 的工程博客上有一篇关于 avro 和 avro-builder 的精彩文章！