Dropbox 推出实时异步处理框架：Cape

每天，Dropbox 需要存储数十亿个文件，为了响应这些操作并实现 Dropbox 所提供的各类功能，我们需要运行大量异步作业。例如需要通过异步作业为文件创建索引，借此提供内容搜索功能，为文件生成预览并显示在 Dropbox 展示这些文件的网页上，以及在文件有改动时为使用 Dropbox 开发者 API 的第三方应用发出通知。这些地方都可以用到 Cape。Cape 是一种实时异步处理框架，每天可处理数十亿条事件，Dropbox 的很多功能都用到了这项技术。

我们最初于 2016 年底发布 Cape，并用它取代了早在 2014 年就在使用的 Livefill 。本文将概括介绍 Cape 的主要功能，高层架构，以及 Dropbox 内部的不同用例。

需求

按照设计，Cape 是一种通用框架，可用于处理不同来源的事件流，开发者可执行自定义业务逻辑对这些事件做出响应。因此这个框架需要满足很多前提需求：

低延迟

为了驱动用户通过移动设备，实时并且不断进行的内容协作和共享，Cape 必须能在不到一秒的延迟内完成处理任务。例如，我们希望用户能够在上传了文件之后，立刻能够搜索或查看所上传的文件，并能立刻分享给他人。

支持多种事件类型

与原本使用的 Livefill 类似，Cape 需要足够通用，能够处理不同类型的事件，而不仅仅是 Dropbox 文件事件本身。这样 Dropbox 产品才能实时响应与文件元数据改动无关的事件，例如文件分享、共享文件夹权限改动，或针对文件添加的备注。随着 Dropbox 逐渐从一个单纯的文件同步解决方案转变成一种协作平台，这样的功能会变得越来越重要。

规模

Cape 的吞吐率必须足够高，以支持每秒上万个与文件有关和无关的事件。另外，同一个事件可能会触发很多不同类型的工作负载，这也进一步放大了我们所要支持的规模。

多变的工作负载

从数毫秒到数十秒，甚至某些情况下数分钟，取决于工作负载的类型以及所处理的事件，不同时段的工作负载可能有所差异。

隔离

Cape 必须能在不同 Cape 用户的事件处理代码中实现合理程度的隔离，这样一个用户处理过程中出现的问题才不会对其他所有使用该框架的用户产生太大的不利影响。

“至少一次”的保证

Cape 必须能够确保每个事件至少被处理一次，对很多用例来说，这是确保整个体验准确一致的关键。

数据模型

Cape 中的每个事件流可称为一个域 (Domain)，其中会包含某一特定类型的事件。每个 Cape 事件都有一个自己的主题 (Subject)和一个序列号 (Sequence ID)。主题是指发生该事件的主体，对应的主题 ID 则可用于区分不同主题。序列号是一种单调递增 (Monotonically increasing) 的编号，可为某一主题范围内的多个事件提供严格的顺序保证。

第一版 Cape 支持两个事件源，这两个事件源承担了 Dropbox 很多非常重要的用户事件。继续介绍前，我们会首先简要介绍一下这些事件源。

第一个事件源是 SFJ （Server File Journal，服务器文件日志），这是 Dropbox 中文件的元数据数据库。对用户的 Dropbox 中存储的每个文件所做的每个改动都会被关联一个名称空间ID (NSID) 和一个日志ID (JID)，这两个ID 结合在一起即可用于标识SFJ 中的每个事件。可支持的第二个事件源是 Edgestore ，这是一种面向非文件元数据的元数据存储，Dropbox 的很多服务和产品都用到了该事件源。针对特定 Edgestore实体(Entity) 或关联(Association) 类型进行的所有改动都可通过配合使用 GID （全局 ID）和 Revision ID（修订 ID）的方式进行唯一区分。

下表介绍了 SFJ 和 Edgestore 事件融入 Cape 事件抽象的具体方法：

源 域 主题 ID 序列号 SFJ SFJ（所有 SFJ 事件） 名称空间 ID 日志 ID Edgestore 实体或关联类型 GID 修订 ID

以后，Cape 除了支持 SFJ 和 Edgestore 事件，还将支持处理自定义事件流。例如流入 Kafka 集群的事件可以按照下列方式融入 Cape 的事件流抽象：

源 域 主题 ID 序列号 Kafka 一个 Kafka 集群 {Topic, Partition} 偏移量

架构

下图概括展示了 Cape 的架构：

Cape 系统架构

SFJ 和 Edgestore 服务在事件发生时，通过 RPC 向Cape 前端发送包含相关事件元数据的 Ping。这些 Ping 并不位于关键的 SFJ 和 Edgestore 路径上，因此可以异步发送。这样的设计可将对重要 Dropbox 服务整体可用性的影响降至最低（当 Cape 或所依赖的服务遇到问题时），同时依然可以在常规情况下实现实时事件处理。Cape 前端会将这些 Ping 发布至 Kafka 队列，并在队列中持久保存，直到处理完成。

上文提到的 Kafka 队列的Cape 调度器 (Dispatcher)订户会收到事件 Ping，并开始进行必要的处理。调度器中包含了 Cape 的所有智能业务逻辑，会根据用户的 Cape 配置，酌情调度事件至相应的 Lambda 工作进程 (Worker)。此外调度器还会为 Cape 的其他保证提供支撑，例如不同事件的顺序，以及 Lambda 之间的依赖性。

Lambda 工作进程通过 Redis 收到来自调度器的事件，执行用户的业务逻辑，并使用处理状态作为对 Cape 调度器的响应：如果成功处理完成，该事件的处理过程就算告一段落。

正如上文所述，来自 SFJ 和 Edgestore 的 Ping 会以异步的方式发送，并会独立于 Cape 前端的关键路径之外，当然这就意味着无法确保每个事件都能发送。你可能发现，这似乎会导致 Cape 无法保证每个事件至少被处理一次。例如可能一个文件已经同步至 Dropbox，但我们可能错过了随后需要进行的所有异步处理。因此我们设计了Cape Refresh，这些工作进程会持续扫描 SFJ 和 Edgestore 元数据中可能错过的最新事件，并将必要的 Ping 发送至 Cape 前端，借此确保所有事件都被处理。此外这也为我们提供了一种机制，使得我们可从 Cape 所处理的数十亿个主题中检测到用户应用程序代码中可能存在的永久性失败。

Cape 的使用

为了实现与 Cape 的集成，开发者需要执行下列操作：

1. Lambda：编写执行后可对事件做出响应的代码，借此实现一个妥善定义的接口（目前支持使用 Python 或 Go），这也叫做 Lambda。
2. 部署：使用 Dropbox 的标准部署和监视工具部署 Cape 工作进程，借此接受事件并运行新创建的 Lambda。
3. 配置：通过指定域，将需要处理的事件集包含在 Cape 的配置中，随后即可运行 Lambda 对其做出响应。

上述操作完成后，新部署的 Cape 工作进程即可收到与相关 Lambda 有关的事件，并酌情处理这些事件，随后使用处理状态响应 Cape 调度器。用户还可以访问一个自动生成的监视仪表盘，其中显示了特定 Lambda 所执行事件处理工作的相关基本指标。

总结

Cape 目前每天已经可以处理数十亿条事件，对于 SFJ 和 Edgestore，95% 的情况下延迟分别低于 100 毫秒和 400 毫秒。该延迟是通过衡量从 Cape 前端收到事件，到 Lambda 工作进程开始处理该事件这一过程之间的时间确定的。对于 SFJ，端到端延迟（例如从变更提交至 SFJ 再到 Lambda 工作进程开始处理）更高一些，分别为大约 500 毫秒和 1 秒，这是由于 SFJ 服务自身的批处理造成的。

目前 Cape 已经在 Dropbox 内部获得了广泛应用，下文将列举一二：

1. 审计日志：Cape 可用于对 Dropbox 与审计日志有关的事件进行实时索引，这样 Dropbox Business 的管理员就可以随时搜索这些日志。
2. 搜索：在 Dropbox 中存储的文件有改动时，Cape 可进行实时索引，方便用户搜索文件的内容。
3. 开发者 API：Cape 可为使用 Dropbox 开发者 API 的第三方应用提供有关文件改动的实时通知。
4. 共享权限：Cape 可用于以异步方式执行开销巨大的操作，例如针对包含大量内容或具备较深结构的共享文件夹应用更改后的权限。

鉴于该框架的通用性极强，并且 Dropbox 的大部分重要事件都可使用，我们预计未来 Cape 还将用来巩固 Dropbox 的其他各种功能。

本文对 Cape 的架构进行了概括的介绍，但并未涉及太多细节和设计方面的决策。稍后我们还将发布更多文章，深入探讨 Cape 的技术细节和与 Cape 的架构、设计，以及实现有关的内容。

感谢为 Cape 的诞生做出奉献的人们：Bashar Al-Rawi、Daisy Zhou、Iulia Tamas、Jacob Reiff、Peng Kang、Rajiv Desai，以及 Sarah Tappon。

作者：Neeraj Kumar，阅读英文原文： Introducing Cape

感谢郭蕾对本文的审校。

给InfoQ 中文站投稿或者参与内容翻译工作，请邮件至 editors@cn.infoq.com 。也欢迎大家通过新浪微博（ @InfoQ ， @丁晓昀），微信（微信号： InfoQChina ）关注我们。