在过去的一年里，我一直是负责Wix的事件驱动消息基础设施（基于Kafka之上）的数据流团队的一员。该基础设施被 1400 多个微服务使用。
在此期间，我已经实现或目睹了事件驱动消息传递设计的几个关键模式的实现，这些模式有助于创建一个健壮的分布式系统，可以轻松处理不断增长的流量和存储需求。
一、消费与投射
…那些非常受欢迎的服务会成为瓶颈
当您遇到存储大型领域对象的“流行”数据的瓶颈时，此模式可以提供帮助。
在 Wix，我们的MetaSite服务就是这种情况，它为 Wix 用户创建的每个站点保存了大量元数据，例如站点版本、站点所有者以及站点上安装了哪些应用程序-已安装的应用程序上下文。
此信息对于 Wix 的许多其他微服务（团队）很有价值，例如Wix Stores、Wix Bookings、Wix Restaurants等等。这个单一的服务被超过 100 万 RPM 的请求轰炸，以获取网站元数据的各个部分。
通过查看服务的各种 API 可以明显看出，它正在处理其客户端服务的太多不同的问题。
MetaSite 服务处理约 1M RPM 的各种请求
我们想要回答的问题是，我们如何以最终一致的方式从该服务转移读取请求？
使用 Kafka 创建“物化视图”负责这项服务的团队决定创建一项附加服务，该服务仅处理 MetaSite 的一个问题——来自其客户端服务的“已安装应用程序上下文”请求。
- 首先，他们将所有数据库的站点元数据对象流式传输到 Kafka 主题，包括新站点创建和站点更新。一致性可以通过在 Kafka Consumer 中进行数据库插入来实现，或者通过使用像Debezium这样的[url=https://en.wikipedia.org/wiki/Change_data_capture]CDC[/url]产品来实现。
- 其次，他们使用自己的数据库创建了一个“只写”服务（反向查找写入器），该服务使用站点元数据对象，但仅获取已安装应用程序上下文并将其写入数据库。即，它将站点元数据的某个“视图”（已安装的应用程序）投射到数据库中。
使用和项目安装的应用程序上下文
- 第三，他们创建了一个“只读”服务，只接受与已安装应用程序上下文相关的请求，他们可以通过查询存储计划的“已安装应用程序”视图的数据库来完成这些请求。
拆分读写
结果：
- 通过将数据流式传输到 Kafka，MetaSite 服务与数据的消费者完全分离，从而显着降低了服务和数据库的负载。
- 通过使用来自 Kafka 的数据并为特定上下文创建“物化视图”，反向查找编写器服务能够创建最终一致的数据投影，该投影针对其客户服务的查询需求进行了高度优化。
- 将读取服务与写入服务分开，可以轻松扩展只读数据库复制和服务实例的数量，以处理来自全球多个数据中心的不断增长的查询负载。
2.端到端的事件驱动
…便于业务流程状态更新
请求-回复模型在浏览器-服务器交互中特别常见。通过将 Kafka 与websocket一起使用，我们可以驱动整个流事件，包括浏览器-服务器交互。
这使得交互更具容错性，因为消息保存在 Kafka 中，并且可以在服务重新启动时重新处理。这种架构也更具可扩展性和解耦性，因为状态管理完全从服务中移除，并且不需要数据聚合和查询维护。
考虑以下用例 - 将所有 Wix 用户的联系人导入 Wix 平台。
此过程涉及多项服务——Contacts Jobs 服务处理导入请求并创建导入批处理作业，Contacts Importer执行联系人的实际格式化和存储（有时在 3rd 方服务的帮助下）。
注册，然后会告诉你结果传统的请求-回复方式需要浏览器不断轮询导入状态，前端服务保持部分数据库表的状态更新，同时轮询用于状态更新的下游服务。
相反，通过使用 Kafka 和websockets 管理器服务，我们可以实现一个完全分布式的事件驱动流程，其中每个服务完全独立工作。
E2E事件驱动使用Kafka和Websockets
首先，浏览器根据请求开始导入，将订阅 web-sockets 服务。 它需要提供一个通道 ID，以便websockets 服务能够将通知正确地路由回正确的浏览器：
为通知打开 websocket “通道”
其次，浏览器需要向作业服务发送CSV 格式的联系人的 HTTP 请求，并附加通道 ID，因此作业服务（和下游服务）将能够向websockets 服务发送通知。请注意，HTTP 响应将立即返回，没有任何内容。
第三，jobs service处理完请求后，产生对kafka topic的job请求。
HTTP 导入请求 + 生成的导入作业消息
第四，Contacts 导入服务消费来自 Kafka 的作业请求并执行实际的导入任务。当它完成时，它可以通知websockets 服务工作已经完成，这反过来可以通知浏览器。
通知已使用、已处理和完成状态的作业
结果：
- 使用这种设计，在导入过程的各个阶段通知浏览器变得轻松，无需保持任何状态，也无需任何轮询。
- 使用 Kafka 使导入过程更具弹性和可扩展性，因为多个服务可以处理来自同一个原始导入 http 请求的作业。
- 使用 Kafka 复制，很容易将每个阶段都放在最合适的数据中心和地理位置。也许导入器服务需要在 google dc 上才能更快地导入 google 联系人。
- 对 web sockets 的传入通知请求也可以生成到 kafka 并复制到 websockets 服务实际驻留的数据中心。
3.内存KV存储
…用于 0 延迟数据访问
有时我们需要为我们的应用程序进行动态而持久的配置，但我们不想为它创建一个完整的关系数据库表。
一种选择是使用HBase / Cassandra / DynamoDB为所有应用程序创建一个大的Wide Column Store表，其中主键包含标识应用程序域的前缀（例如“stores_taxes_”）。
此解决方案运行良好，但通过网络获取值存在内置延迟。它比配置数据更适合更大的数据集。
另一种方法是拥有一个内存中的键/值缓存，它也具有持久性——Redis AOF提供了这种能力。
Kafka 以压缩主题的形式为键/值存储提供了类似的解决方案（其中保留模型确保不会删除键的最新值）。
在 Wix，我们将这些压缩主题用于内存中的 kv 存储，我们在应用程序启动时加载（使用）来自主题的数据。一个很好的好处（Redis 没有提供）是该主题仍然可以被其他想要获取更新的消费者使用。
订阅和查询考虑以下用例——两个微服务使用压缩主题来维护他们维护的数据：Wix Business Manager（帮助 Wix 网站所有者管理他们的业务）使用压缩主题来支持国家列表，以及Wix Bookings（允许安排约会和课程）维护一个“时区”压缩主题。从这些内存中的 kv 存储中检索值的延迟为 0。
每个 In-memory KV Store 及其各自的压缩 Kafka 主题
Wix Bookings侦听“支持的国家/地区”主题的更新：
Bookings 消耗来自 Country 压缩主题的更新
当Wix Business Manager将另一个国家/地区添加到“国家/地区”主题时，Wix Bookings会使用此更新并自动为“时区”主题添加新的时区。现在内存中的“时区” kv-store 也更新为新时区：
压缩主题中添加了南苏丹的新时区
我们不需要停在这里。Wix Events（允许 Wix 用户管理活动门票和 RSVP）还可以使用Bookings的时区主题，并在一个国家/地区更改其时区以实现夏令时自动获取其内存中 kv 存储的更新。
从同一个压缩主题消费的两个内存中 KV 存储
4. 安排并忘记
…当您需要确保计划的事件最终得到处理时
在很多情况下，Wix 微服务需要根据某个时间表执行作业。
一个例子是管理基于订阅的支付（例如订阅瑜伽课程）的Wix 支付订阅服务。 对于每个每月或每年订阅的用户，必须与支付提供商进行续订过程。
为此，Wix 自定义Job Scheduler服务调用由Payments Subscription服务预先配置的 REST 端点。
订阅续订过程发生在幕后，无需（人类）用户参与。这就是为什么即使出现临时错误（例如，第三个支付提供商不可用），续订最终也会成功很重要。
确保此过程完全有弹性的一种方法是，作业调度程序向Payment Subscriptions服务发出频繁的重复请求，其中当前的续订状态保存在 DB 中，并针对尚未到期的续订的每个请求进行轮询扩展。这将需要对数据库进行悲观/乐观锁定，因为同一用户可能同时有多个订阅扩展请求（来自两个单独的正在进行的请求）。
更好的方法是首先向 Kafka 发出请求。为什么？处理请求将由 Kafka 消费者按顺序（针对特定用户）完成，因此不需要用于同步并行工作的机制。
此外，一旦将消息生成到 Kafka，我们可以通过引入消费者重试来确保它最终会被成功处理。由于这些重试，请求的计划也可能不那么频繁。
在这种情况下，我们要确保保持处理顺序，因此重试逻辑可以简单地在具有指数退避间隔的尝试之间休眠。
Wix 开发人员使用我们定制的Greyhound消费者，因此他们只需指定一个 BlockingPolicy 和适当的重试间隔来满足他们的需求。
在某些情况下，消费者和生产者之间可能会出现延迟，以防错误长时间持续存在。在这些情况下，有一个特殊的仪表板用于解锁和跳过我们的开发人员可以使用的消息。
如果消息处理顺序不是强制性的，那么 Greyhound 中也存在利用“重试主题”的非阻塞重试策略。
配置重试策略后，Greyhound Consumer 将创建与用户定义的重试间隔一样多的重试主题。内置的重试生产者将在出错时生成消息到下一个重试主题，并带有一个自定义标头，指定在下一次处理程序代码调用之前应该发生多少延迟。
对于所有重试尝试都已用尽的情况，还有一个死信队列。在这种情况下，消息被放入死信队列，供开发人员手动查看。
这种重试机制的灵感来自这篇uber 文章。
Wix 最近开源了Greyhound，很快就会对 beta 用户开放。要了解更多信息，您可以阅读 github自述文件。
概括：
- Kafka 允许按某个键顺序处理请求（例如 userId 进行订阅续订），从而简化工作逻辑
- 由于 Kafka 重试策略的实施大大提高了容错能力，更新请求的作业计划频率可以大大降低。
5. 交易事务中的事件
…当幂等性难以实现时
考虑以下经典电子商务流程：
我们的支付服务向 Kafka生成订单 购买完成事件。现在Checkout服务将使用此消息并生成自己的Order Checkout Completed消息以及所有购物车项目。
然后所有下游服务（交付、库存和发票）将需要使用此消息并继续处理（分别准备交付、更新库存和创建发票）。
如果下游服务可以依赖Order Checkout Completed事件仅由 Checkout 服务生成一次，则此事件驱动流的实现将容易得多。
为什么？因为多次处理相同的 Checkout Completed 事件可能会导致多次交付或不正确的库存。为了防止下游服务发生这种情况，他们需要存储重复数据删除状态，例如，轮询一些存储以确保他们之前没有处理过这个 Order Id。
这通常使用常见的数据库一致性策略来实现，例如悲观锁定和乐观锁定。
幸运的是，Kafka 为这种流水线事件流提供了一个解决方案，其中每个事件只处理一次，即使服务有一个消费者-生产者对（例如 Checkout），它既消费一条消息又产生一条新消息。
简而言之，当Checkout服务处理传入的Payment Completed事件时，它需要将 Checkout Completed 事件的发送包装在生产者事务中，它还需要发送消息偏移量（以允许 Kafka 代理跟踪重复消息） .
此事务期间产生的任何消息仅在事务完成后对下游消费者（库存服务）可见。
此外，基于 Kafka 的流程开始时的支付服务生产者必须变成一个幂等生产者——这意味着代理将丢弃它产生的任何重复消息。
有关更多信息，您可以观看我关于Kafka 中的 Exactly once 语义的简短介绍性演讲
6. 事件聚合
…当你想知道一整批事件已经被消费了
在将联系人导入 Wix CRM 平台的业务流程。后端包括两个服务。提供 CSV 文件并向 Kafka 生成作业事件的作业服务。以及使用和执行导入作业的联系人导入器服务。
让我们假设有时 CSV 文件非常大，将工作负载拆分为较小的作业更有效，每个作业中要导入的联系人更少。这样，可以将工作并行化到 Contacts Importer 服务的多个实例。但是，当导入工作被拆分为许多较小的工作时，您如何知道何时通知最终用户所有联系人都已导入？
显然，已完成作业的当前状态需要持久化，否则内存中已完成作业的记帐可能会丢失到随机的 Kubernetes pod 重启。
在不离开 Kafka 的情况下保持这种会计处理的一种方法是使用 Kafka Compacted Topics。这种话题可以认为是一个流式KV存储。
在我们的示例中，Contacts Importer服务（在多个实例中）将使用带有索引的作业。每次完成处理某个作业时，它都需要使用 Job Completed事件更新 KV 存储。这些更新可以同时发生，因此可能会发生潜在的竞争条件并使作业完成计数器无效。
Atomic KV Store 为了避免竞争条件，Contacts Importer服务会将完成事件写入Atomic KVStore类型的Jobs-Completed-Store 。
原子存储确保所有作业完成事件将按顺序处理。它通过创建一个“commands”主题和一个压缩的“store”主题来实现这一点。
顺序处理
在下图中，您可以看到原子存储如何以 [Import Request Id]+[total job count] 作为键生成每个新的导入作业完成的“更新”消息。通过使用key，我们可以依靠 Kafka 始终将特定 requestId 的“更新”放在特定分区中。
接下来，作为 atomic store 一部分的消费者-生产者对将首先监听每个新更新，然后执行 atomicStore 用户请求的“命令”——在这种情况下，将已完成作业的数量从以前的值。
端到端更新流程示例 让我们回到 Contacts Importer 服务流程。一旦这个服务实例完成了一些作业的处理，它会更新 Job-Completed KVAtomicStore（例如，Import Job 3 of request Id YYY 已经完成）：
Atomic Store 将向 job-completed-commands 主题生成一条新消息，其中 key = YYY-6 和 Value — Job 3 Completed。
接下来，Atomic Store 的消费者-生产者对将使用此消息并增加 KV Store 主题的 key = YYY-6 的已完成作业计数。
Exactly Once Processing 请注意，处理“命令”请求必须恰好发生一次，否则完成计数器可能不正确（错误增量）。为消费者-生产者对创建一个 Kafka 事务（如上面的模式 4 中所述）对于确保会计保持准确至关重要。
AtomicKVStore 值更新回调 最后，一旦已完成作业计数的最新 KV 生成值与总数匹配（例如 YYY 导入请求的 6 个已完成作业），就可以通知用户（通过 web 套接字 — 参见第一部分的模式 3文章）关于导入完成。通知可以作为 KV 存储主题产生操作的副作用发生 - 即调用其用户提供给 KV 原子存储的回调。
重要笔记：
- 完成通知逻辑不必驻留在Contacts Importer服务中，它可以在任何微服务中，因为此逻辑与此流程的其他部分完全解耦，仅依赖于 Kafka 主题。
- 不需要进行预定的轮询。整个过程是事件驱动的，即以管道方式处理事件。
- 通过使用基于键的排序和恰好一次 Kafka 事务，作业完成通知或重复更新之间不可能存在竞争条件。
- Kafka Streams API 非常适合这样的聚合需求，其 API 功能包括groupBy（按导入请求 ID 分组）、reduce或count（计数已完成的作业）和filter（计数等于总作业数），然后是 webhook 通知副作用. 对于 Wix，使用现有的生产者/消费者基础设施更有意义，并且对我们的微服务拓扑的干扰更少。