Zulip 实时事件系统（Events System）深度解析：从事件生成、长轮询投递到初始数据原子同步

Zulip 实时事件系统（Events System）深度解析：从事件生成、长轮询投递到初始数据原子同步

【免费下载链接】zulipZulip server and web application. Open-source team chat that helps teams stay productive and focused.项目地址: https://gitcode.com/GitHub_Trending/zu/zulip

Zulip 的Events System（实时推送与事件系统）是其服务端到客户端的推送系统，承担了团队聊天产品中"一个客户端修改的数据如何实时同步到其他客户端"这一核心职责。本文以 docs/subsystems/events-system.md 为主线，结合仓库中 zerver/tornado/event_queue.py、zerver/lib/events.py、zerver/tornado/django_api.py 等核心源码，完整讲解事件生成（Generation）、投递（Delivery）、UI 更新三大部分的设计与实现，并深入剖析注册（register）接口的原子性初始数据获取算法、apply_events的自动化测试协议与事件 Schema 变更的向后兼容策略。读完本文，你将掌握 Zulip 实时同步的完整调用链，并具备在 Zulip 中新增一种事件类型并配套测试的能力。

为什么需要一套专门的事件系统

任何单页 Web 应用都需要回答"一个客户端做出的变更如何同步到其他客户端"这个问题。对于 Zulip 这样的聊天工具，状态时刻在变化，这个问题尤为关键。这里的"客户端"指需要接收 Zulip 数据更新的浏览器标签页、移动端 App 或 API Bot。最简单的例子是：一个客户端发送了一条新消息，其他客户端必须被通知才能显示这条消息。而一个完整的应用如 Zulip，有几十种需要同步的数据类型——新建频道、用户改名或换头像、设置变更等。在 Zulip 中，这些需要发送给其他客户端的更新统称为事件（events）。

设计这类系统时有一个重要原则：事件需要同步给每一个持有旧数据副本的客户端，否则客户端会向用户展示过期数据。因此，如果一个用户开两个浏览器窗口并发送消息，该用户控制的所有客户端以及消息的所有接收方（包括那两个浏览器窗口）都会收到事件。严格来说，不需要给触发变更的那个客户端发事件，但 Zulip 选择了"全量下发"——这样客户端无需为"自己触发的变更"和"别人触发的变更"各写一套 UI 更新代码，触发方只需复用与其他客户端完全相同的代码，最多再加一点"操作成功"的通知即可。

从架构上看，一个成功的实时同步系统需要三部分：

• Generation（生成）：数据发生变化时生成事件，并确定每个事件应发给哪些用户。
• Delivery（投递）：高效地把事件投递给感兴趣的客户端，理想情况下做到"恰好一次"（exactly-once）。
• UI updates（更新）：客户端收到事件后更新界面。

React、Vue 这类响应式 JavaScript 库可以帮助简化第三部分，但生成与投递没有成熟的标准系统可复用，Zulip 必须自己构建。本文接下来就聚焦这两部分如何在 Zulip 中以可扩展、正确且可预测的方式实现。

事件生成系统（Generation）

Zulip 的生成系统围绕一个 Python 函数send_event_on_commit(realm, event, users)构建，其真实实现位于 zerver/tornado/django_api.py。它接收三个参数：

• realm：用于分片（sharding），决定事件应投递到哪个 Tornado 端口/进程；
• event：事件数据结构，本质就是一个 Python 字典，type键始终存在，其余键随具体事件类型而定；
• users：应接收事件的用户 ID 列表。在消息投递等特殊场景下，users会是一组字典，把用户 ID 映射到用户相关数据，例如该用户是否在消息中被提及。

传入send_event_on_commit的数据会被简单地序列化为 JSON，放入名为notify_tornado的 RabbitMQ 队列，等待投递系统消费。从源码看，send_event_on_commit的核心是transaction.on_commit(lambda: send_event_rollback_unsafe(realm, event, users))——它必须在"正在修改状态的那个数据库事务内部"调用，从而保证只有事务成功提交后才发送事件，事务回滚则事件不发。send_event_rollback_unsafe则按 realm 的 Tornado 端口对用户进行分组（get_realm_tornado_ports/get_user_id_tornado_port），再通过queue_json_publish_rollback_unsafe发布到对应端口的notify_tornado队列。源码注释还约定：这类函数只能从zerver/actions/*.py调用，便于集中查找事件生成代码；每个调用点都应由test_events.py中的测试覆盖，并在zerver/lib/event_schema.py中校验 Schema。

通常情况下，users列表是以下三种之一：

• 单个用户：例如用户级设置变更；
• 整个 realm 的所有人：例如组织级设置变更、新增 realm 表情；
• 会收到某条消息的所有人：例如消息、表情回应、消息编辑等，即某频道的订阅者或某私信会话的参与者。

选择正确的用户 ID 列表是调用方的责任。如果错误地把包含私信内容的事件发给整个组织，会造成安全问题；反之，如果事件没发给足够的客户端，就会出现用户可见的实时同步 Bug。

事件生成过程中最困难的部分，是定义"一致的事件字典"：要清晰、可读、对各类客户端都有用，并方便开发者维护。

事件投递系统（Delivery）

基于 Tornado 的长轮询架构

Zulip 的事件投递（实时推送）系统基于Tornado——它非常适合处理大量保持打开的请求，详情可参考 架构总览。整个系统约 2000 行代码，集中在zerver/tornado/目录，主体是 zerver/tornado/event_queue.py。

投递机制采用长轮询（long-polling）：客户端发起GET /json/events请求，服务器在"有事件可投递"之前不响应这个请求。这种方式相当高效且兼容性好（相比 WebSocket，WebSocket 存在逐渐减少但并非为零的客户端兼容性问题）。

对每个已连接的客户端，事件队列服务器（event queue server）维护一个事件队列（event queue），队列里存放"该投递给这个客户端、但客户端尚未确认（acknowledge）的事件"。忽略错误处理的细节，协议非常简单：

1. 客户端发起GET /json/events；
2. 服务器检查队列中是否有事件：
   • 有：立即把事件作为响应返回；
   • 没有：把该队列记录为"有等待中的客户端"（代码中常称为handler）。
3. 当服务器从notify_tornadoRabbitMQ 队列拉出事件时，就把它投递给目标用户关联的每个队列：
   • 若队列有等待中的客户端：中断长轮询，向等待中的请求返回 HTTP 响应；
   • 若没有等待中的客户端：直接把事件压入队列。

注册与轮询协议

客户端启动时会先调用POST /json/register，服务器为其创建一个新的事件队列，返回queue_id以及初始的last_event_id（可选地，还可以顺带拉取初始数据以节省一个 RTT 并避免竞态，详见下文"初始数据获取"）。注册完成后，客户端只需进入无限循环：用这两个参数反复调用GET /json/events，每次处理完事件后更新last_event_id以确认已收到（Python API 绑定中的call_on_each_event是完整的示例实现）。在处理每个GET /json/events请求时，队列服务器可以安全地删除事件 ID 小于等于客户端last_event_id的事件（事件 ID 只是该队列收到事件的计数器）。

last_event_id参数在无网络故障时并非必需，但它是实现exactly-once 投递的关键：如果没有它，队列服务器只能在"尝试发送事件时就删除事件"，一旦那次 HTTP 响应因 TCP 网络故障没有送达客户端，事件就永久丢失了。

心跳、垃圾回收与持久化

队列服务器是超高流量系统，至少为"每一条投递给每个 Zulip 客户端的消息"处理一次请求。此外，为绕过低质量 NAT 服务器"杀死空闲超过 60 秒的 HTTP 连接"的问题，队列服务器还会在无其他事件到达时，至少每 45 秒向每个队列发送一个心跳事件。源码中的常量可佐证：HEARTBEAT_MIN_FREQ_SECS = 45，心跳事件即dict(type="heartbeat")（见create_heartbeat_event，event_queue.py）。

为避免内存等资源泄漏，队列在客户端默认闲置 10 分钟后会被垃圾回收（GC 每分钟扫描一次，见DEFAULT_EVENT_QUEUE_TIMEOUT_SECS = 60 * 10与EVENT_QUEUE_GC_FREQ_MSECS，event_queue.py），其假设是客户端大概率已断网或不存在。客户端重新回来时会收到 "queue not found" 错误，其处理方式就是重启客户端/刷新浏览器，像启动时一样重新拉取初始数据。由于客户端反正要实现启动流程，这套方案对客户端几乎不增加复杂度。一个额外的好处是：即使队列服务器（队列保存在内存中）崩溃丢失数据，客户端也能自动恢复，就像短暂断网一样（仍需防范 DoS 风险）。垃圾回收系统还带有钩子，对 通知系统 的实现很重要。

值得注意的是，事件队列服务器被设计为将事件队列保存到磁盘并在重启时重新加载（对应ClientDescriptor的to_dict/from_dict序列化机制，见 event_queue.py），并小心地捕获异常，因此此类崩溃非常罕见；但设计上即使发生，也不会留下损坏的过期客户端。

Tornado 侧的事件处理

从 RabbitMQ 队列消费到事件后，核心分发逻辑在process_event：对每个目标用户 ID，取出该用户的全部客户端描述符（ClientDescriptor），凡是accepts_event(event)为真的客户端都调用client.add_event(event)。process_event的调用链可通过zerver/tornado/views.py中的get_events（zerver/tornado/views.py）与fetch_events（event_queue.py）继续深入。ClientDescriptor保存了该客户端注册时的各项能力与过滤条件：event_types、narrow（消息范围过滤，经build_narrow_predicate编译为谓词）、bulk_message_deletion、stream_typing_notifications、simplified_presence_events等，accepts_event正是基于这些字段决定是否把事件投给该客户端。

初始数据获取与原子性保证

客户端启动时通常想从服务器拿到两样东西：

• 各类数据的"当前状态"（initial state）：当前设置、组织用户集合（用于输入联想）、频道、消息等；
• 对这些数据的更新订阅（即一个事件队列）。

理想情况是这两者原子地获取：假设其他用户此时改了名字，那么要么改名发生在拉取之后（初始状态里是旧名字，队列里会有一个改名事件），要么发生在之前（初始状态是新名字，队列里没有改名事件）。绝不出现"初始状态是旧名字、队列里也没有改名事件"这种数据永远对不上的情况。

实现这种原子性可以让 N 个 Zulip 客户端免于处理大量罕见且难以复现的竞态条件——只需在 Zulip 服务器端把这件事一次性做对。

技术上这很有挑战：拉取 Zulip 这种复杂应用的初始状态可能要执行几十次数据库、缓存查询，耗时 100ms 以上，几乎不可能把这些查询原子地完成。Zulip 的解决办法是：用非原子的子过程组合出原子结果。逻辑位于 zerver/views/events_register.py 与 zerver/lib/events.py，registerAPI 请求由 Django 直接处理，流程如下：

1. Django 向 Tornado 发起 HTTP 请求，要求创建一个新的事件队列，并记下其queue_id；
2. Django 非原子地从各个数据源执行各种数据库/缓存查询拉取数据（见fetch_initial_state_data）；
3. Django 第二次向 Tornado 发起 HTTP 请求，取回"自队列创建以来新增到该队列的所有事件"；
4. 最后 Django 把这些事件"应用"到拉取的初始状态上（见apply_events）。例如对改名事件，在realm_user数据结构中找到该用户并更新为新名字。

fetch_initial_state_data有大量参数（见 events.py），包括event_types（为None时拉取支撑 Web 端page_params与/api/v1/register的核心数据；指定时只拉取子集）、client_gravatar、slim_presence、include_subscribers、archived_channels等，并把zulip_version、zulip_feature_level等版本信息无条件放入 state。apply_events则遍历事件，先按fetch_event_types过滤（避免把未订阅类型的事件应用到无关状态上），再逐个交给apply_event处理（如 message 事件会更新state["max_message_id"]）。

测试：为什么需要apply_events协议

上述设计达成了所有目标，代价是必须写一个正确的apply_events函数。这个函数很难写对，因为它处理的场景（竞态条件）在手工测试中几乎从不出现。幸运的是，Zulip 的自动化后端测试有一套专门的测试协议。

测试总览

当你完全确信某个 "action 函数"在"正常操作"下工作正确（通常意味着为对应的 POST/GET 操作写了一个全栈测试），就可以在test_events.py中写测试了。一个test_events测试的实际代码可以非常简洁：

def test_default_streams_events(self) -> None: stream = get_stream("Scotland", self.user_profile.realm) events = self.verify_action(lambda: do_add_default_stream(stream)) check_default_streams("events[0]", events[0]) # (some details omitted)

真正的技巧在于调试这些测试。上述示例做了三件事：

• 准备数据（get_stream）；
• 用verify_action包装一个 action 函数（do_add_default_stream）；
• 用 schema 检查器校验数据（check_default_streams）。

test_events.py文件位于 zerver/tests/test_events.py。

verify_action

所有与apply_events相关的重活都发生在verify_action调用里，它是test_events.py中BaseAction类的测试辅助方法。verify_action通过模拟可能的竞态条件来验证apply_events逻辑在某个 action 函数语境下是否正确。用上面的例子说：把do_add_default_stream产生的事件通过apply_events应用到一个过期的状态副本上，得到的结果应当与"先执行 action 再拉取一份全新状态"完全相同。

具体来说，verify_action依次执行：

1. 调用fetch_initial_state_data获取当前状态；
2. 调用 action 函数（如do_add_default_stream）；
3. 捕获 action 函数产生的事件；
4. 检查这些事件已在 OpenAPI 文档（定义于zerver/openapi/zulip.yaml）中登记；
5. 调用apply_events(state, events)得到"混合状态"（hybrid state）；
6. 再次调用fetch_initial_state_data得到"正常状态"（normal state）；
7. 对比两者。

如果apply_events逻辑一次写对，两个状态完全一致，verify_action通过并返回 action 产生的事件。通常你第一次会写错，导致verify_action失败——它会打印"混合状态"与"正常状态"的 diff 帮助你调试。遇到这种 diff 可能是一场有挑战的调试，建议重读本文档理解apply_events的设计动机，阅读verify_action自身代码，必要时在聊天中求助。

verify_action只有一个必填参数，即 action 函数，通常用 lambda 表达以便传参：

events = self.verify_action(lambda: do_add_default_stream(stream))

它还有几个值得注意的可选参数：

• state_change_expected：如果 action 确实不引起状态变化（例如输入中提示 typing notifications，这类事件是临时的），必须设为False，否则verify_action会抱怨测试没有真正锻炼apply_events逻辑；
• num_events：告诉verify_action该 action 之后hamlet用户会收到几个事件（默认 1）；
• client_gravatar、slim_presence等参数会被透传给fetch_initial_state_data（对相关 action，两个布尔值最好都测一遍）。

高级用法请直接阅读BaseAction（在 zerver/tests/test_events.py 中）的代码。

Schema 检查

test_events.py系统有两种形式的 Schema 检查。第一种是确认你已更新 GET /events API 文档 来记录新事件格式，方便 Zulip 移动端、终端 App 及其他 API 客户端的开发者；细节见 API 文档。

第二种是test_events内部更细粒度的检查：验证这个特定测试产生了预期的事件序列。看示例测试的最后一行：

# ... events = self.verify_action(lambda: do_add_default_stream(stream)) check_default_streams("events[0]", events[0])

verify_action会返回 action 实际产生的事件，test_events的测试纪律要求验证事件格式可预测。理想情况是测试事件与期望数据完全一致，但由于数据库 id 等不可预测因素，只能验证事件的 "Schema"——用check_default_streams这类 schema 检查器校验数据类型。如果要创建新的事件格式，就得在event_schema.py（zerver/lib/event_schema.py）中自己写 schema 检查器，以下是与示例对应的代码：

default_streams_event = event_dict_type( required_keys=[ ("type", Equals("default_streams")), ("default_streams", ListType(DictType(basic_stream_fields))), ] ) check_default_streams = make_checker(default_streams_event)

注意basic_stream_fields未在文档中列出。理解如何编写 schema 检查器的最佳方式是阅读event_schema.py：文件顶部有一大段注释，然后可以快速浏览其余部分学习模式。创建一个新事件的 schema 检查器，不仅让test_events测试更严格，还允许其他工具复用同一检查器去校验 node 测试 fixtures 与 OpenAPI 文档中的事件格式。

Node 端测试

完成后端测试后，还要在web/tests/lib/events.cjs添加一个示例事件，在web/tests/dispatch.test.cjs中为server_events_dispatch.js的对应事件分发逻辑添加测试（该文件已有约 140 处分发用例），并用tools/check-schemas把示例事件与上面声明的两版 schema 做对照验证。

代码覆盖率

最后还需要确保apply_events始终正确：即 Zulip 能生成的每一种事件类型都有相关测试。可以手动运行test-backend --coverage BaseAction，然后检查所有send_event_on_commit调用点都被覆盖。未来计划用自动化手段直接通过检查覆盖率数据来验证这一点。

page_params

在 Zulip Web 应用中，registerAPI 返回的数据通过page_params参数在页面上可用。

消息：初始数据协议的一个例外

一个例外是真正的消息。因为 Zulip 客户端通常在站点其余部分加载完成之后，用单独的 AJAX 调用拉取消息，所以消息不需要包含在初始状态数据里。为正确起见，客户端需要负责丢弃那些"对应消息客户端尚未拉取"的事件。相关机制还可参考 发送消息。

Schema 变更与向后兼容

当改变发送进 Tornado 的事件格式时，必须正确处理向后兼容：

• 新增事件类型或给现有事件类型加字段：只需在 API 文档 中仔细记录变更，务必提升API_FEATURE_LEVEL并在更新的GET /eventsAPI 文档中加入**Changes**条目；同时建议给移动端与终端项目开 issue 通知。如果 Web 端应在浏览器刷新时收到新事件类型的初始状态，把它加进web/src/server_event_types.ts的FETCH_EVENT_TYPES，Web 端会以fetch_event_types参数传给/register。
• 修改可能干扰现有客户端解析逻辑的字段（如改变现有字段的类型/含义、删除字段）：需要非常谨慎，因为 Zulip 支持旧客户端连接新服务器。具体政策见 发布生命周期。技术方案是：为新的格式添加client_capabilities标志，对未声明支持新能力的客户端继续发送旧格式数据。bulk_message_deletion就是一个很好的参考范例（几年后再把该能力设为必需并移除旧路径）。
• 大多数事件类型：Tornado 只是透明透传，event_queue.py无需改动。
• 但若改变了 Tornado 代码自身使用的数据格式（例如重命名message事件中的presence_idle_user_ids字段）：必须小心，因为升级 Tornado 时，队列里可能还存有升级前的事件。因此必须在event_queue.py中写逻辑把旧格式翻译成新格式，否则升级到相关 commit 时 Tornado 可能崩溃。这类逻辑应集中在from_dict函数（用于事件队列格式变更）和client_capabilities条件分支中（例如process_deletion_event对旧格式客户端的逐条删除事件拆分）。与client_capabilities无关的兼容代码应加# TODO/compatibility: ...注释说明何时可以安全删除，主版本发布时会 grep 这些注释。源码中大量此类注释可佐证，例如process_message_update_event中pm_mention_push_disabled_user_ids到dm_mention_push_disabled_user_ids的重命名翻译代码（event_queue.py）。
• Schema 变更是敏感操作：与数据库 Schema 变更一样，必须做认真的手工测试。例如在测试服务器上运行移动端 App 验证它正确处理新事件，或安排新的 Tornado 代码真正处理一个升级前的事件，并通过浏览器控制台确认输出。

实战：如何新增一种事件类型

综合文档与源码，在 Zulip 中新增一种事件类型的完整路径可归纳为：

1. 生成：在zerver/actions/*.py的 action 函数中，于修改状态的数据库事务内部调用send_event_on_commit(realm, event, users)，event字典必须包含type键；确保users列表精确覆盖应收到事件的客户端。
2. 服务端 Schema：在zerver/lib/event_schema.py中为该事件编写event_dict_type风格的 schema 检查器。
3. 后端测试：在zerver/tests/test_events.py中为 action 编写verify_action测试，并用 schema 检查器校验verify_action返回的事件；用test-backend --coverage BaseAction确认send_event_on_commit调用点被覆盖。
4. API 文档：更新GET /events的 OpenAPI 文档 与zerver/openapi/zulip.yaml，必要时提升API_FEATURE_LEVEL；若 Web 端刷新需要初始状态，更新web/src/server_event_types.ts的FETCH_EVENT_TYPES。
5. Node 端：在web/tests/lib/events.cjs加示例事件，在web/tests/dispatch.test.cjs中为server_events_dispatch.js补测试，并用tools/check-schemas校验。
6. 兼容性：若涉及旧客户端或升级场景，在event_queue.py的from_dict/client_capabilities分支中实现格式翻译，并以# TODO/compatibility:注释标注清理时机。

进一步阅读

• 新应用功能教程：一个完整功能如何使用该事件系统的端到端示例
• 架构总览：Tornado 在整体架构中的位置
• 发送消息：消息事件流的专门文档
• 通知系统：依赖队列垃圾回收钩子的移动/邮件通知实现
• API 文档：GET /events等事件相关 API 的 OpenAPI 规范
• 发布生命周期：事件格式变更的兼容性政策

【免费下载链接】zulipZulip server and web application. Open-source team chat that helps teams stay productive and focused.项目地址: https://gitcode.com/GitHub_Trending/zu/zulip