Faust 1.11.0 变更深度解析：窗口表异步回调、HTTP 状态码转发与依赖升级

• 流处理
• 消息队列
• 后端

本篇文章以 Faust 项目根目录下的 Changelog.rst（即最新版变更说明）为主体，逐条拆解 Faust 1.11.0 版本（release-date 标记为 TBA，尚待发布）的核心变更点：依赖版本升级、Table HTTP Router 响应状态码转发、以及窗口化表（Windowed Table）异步on_window_close回调支持。文章同时结合仓库源码（如 faust/tables/base.py、examples/windowed_aggregation.py）与 1.10 系列历史变更，帮助读者理解这些变更背后的实现原理、升级注意事项与实际用法。

1.11.0 变更总览

根据 Changelog.rst，Faust 1.11.0 的变更主要分为三类：

1. 依赖（Requirements）升级
   • 依赖 Mode 4.3.2（Faust 基于 Mode 构建服务与 CLI 基础设施）；
   • 依赖robinhood-aiokafka1.1.5（Faust 底层 Kafka 客户端，见 faust/transport/drivers/aiokafka.py）。
2. Tables：Table HTTP Router 状态码转发
   • Table 的 HTTP 路由（即 Faust Web 应用中用于浏览/管理 Table 的接口，实现位于 faust/web/apps/tables.py）现在会将底层 HTTP 状态码原样转发给响应，方便外部系统准确感知请求结果（如资源不存在、方法不允许等场景），而不再统一返回固定状态。
   • 该贡献来自 Wjatscheslaw Kewlin（GitHub 用户slawak）。
3. Tables：窗口化表异步on_window_close回调
   • 窗口化表（Windowed Table）现在支持异步on_window_close回调，即回调函数可以定义为async def，Faust 在窗口过期时以await方式调用它。这项能力最早在 1.9.0 引入同步版本（见 docs/history/changelog-1.9.rst），1.11.0 补齐了异步支持。
   • 该贡献来自 Ramkumar M（GitHub 用户billaram）。

版本说明：Changelog.rst开篇明确表示，该文档只收录1.10 系列及最新（1.11.0）的变更说明；更早版本的变更记录统一归档在 docs/history/index.rst（History 章节），其中按版本号拆分为changelog-1.10至changelog-0.9共 11 个历史文档。

深入一：窗口化表的异步on_window_close回调

变更带来的能力

1.11.0 之前，on_window_close回调虽然可用，但只支持同步函数。1.11.0 起窗口过期时 Faust 会通过maybe_async(...)统一处理：同步回调会被直接调用，异步回调则会被await，二者皆可。

从源码看，faust/tables/base.py 中的实现如下：

async def on_window_close(self, key: Any, value: Any) -> None: if self._on_window_close: await maybe_async(self._on_window_close(key, value))

也就是说，只要在创建 Table 时传入on_window_close=callback（faust/tables/base.py），并在回调里使用async def，Faust 的窗口清理流程（_expire_keys相关逻辑，见 faust/tables/base.py）就会在窗口过期时自动触发它。

完整的实战示例

仓库中的官方示例 examples/windowed_aggregation.py 演示了完整的用法——它构造了一个基于事件时间（relative_to_field）的 tumbling 窗口表，并在窗口关闭时对窗口内的所有事件做聚合（计数与均值）后发送到下游 Topic：

import faust from datetime import datetime, timedelta class RawModel(faust.Record): date: datetime value: float app = faust.App('windowed-agg', broker='kafka://localhost:9092', version=1, topic_partitions=1) source = app.topic('raw-event', value_type=RawModel) sink = app.topic('agg-event', value_type=faust.Record) def window_processor(key, events): """同步或异步均可；1.11.0 起可改为 async def。""" timestamp = key[1][0] values = [event.value for event in events] count = len(values) mean = sum(values) / count sink.send_soon(value=RawModel(date=timestamp, count=count, mean=mean)) tumbling_table = ( app.Table( 'tumbling_table', default=list, partitions=1, on_window_close=window_processor, # 窗口关闭时回调 ) .tumbling(10, expires=timedelta(seconds=1)) .relative_to_field(RawModel.date) ) @app.agent(source) async def print_windowed_events(stream): async for event in stream: value_list = tumbling_table['events'].value() value_list.append(event) tumbling_table['events'] = value_list

要点说明：

• on_window_close的签名：回调接收(key, value)。对于窗口化表，key是一个二元组(原始key, (窗口起始, 窗口结束))，因此示例中通过key[1][0]取窗口起始时间戳（源码类型定义见 faust/types/tables.py）；
• 异步化建议：如果回调内部涉及网络 IO（如发送消息、调用外部服务），在 1.11.0 中直接写成async def即可，避免阻塞事件循环；
• expires的作用：配合table_cleanup_interval（示例中设为 1.0 秒）控制窗口何时被判定过期并被清理，从而触发回调。相关实现位于 faust/tables/base.py（_should_expire_keys/_maybe_set_key_ttl）；
• 单元测试方面，on_window_close的调用行为在 t/unit/tables/test_base.py 中有覆盖，可作为行为契约参考。

深入二：Table HTTP Router 转发状态码

Faust 为 Table 提供了内置的 HTTP 管理接口（位于 faust/web/apps/tables.py），用于在 Web 层面查看/操作表数据。1.11.0 的变更使其在返回响应时转发真实的 HTTP 状态码，而不是将各类错误统一折叠为同一状态。这对于以下场景尤其重要：

• 客户端需要区分「请求成功（2xx）」「资源不存在（404）」与「参数错误（4xx）」；
• 与监控、告警系统集成时，需要依赖准确的 HTTP 语义。

从 Faust Web 框架的整体设计看，视图层通过 faust/web/views.py 的View基类处理请求，路由注册在 faust/web/apps/router.py；Table 相关视图在此基础上组合而成。升级到 1.11.0 后，无需修改业务代码即可获得正确的状态码语义。

从 1.10 系列看演进：与 1.11 相关的关键变更

Changelog.rst的定位是「最新变更说明」，因此理解 1.11.0 的若干设计（尤其是blocking_timeout）需要回溯 1.10 系列。在 docs/history/changelog-1.10.rst 的 1.10.4 版本中，Faust 提供了「强制启用阻塞检测」的临时方案，并明确说明：

该方案是 1.11 发布前的临时 workaround，将在 1.11 版本中停止工作；1.11 中请改用新的blocking_timeout设置或BLOCKING_TIMEOUT环境变量。

这解释了为什么 1.11.0 的变更列表中虽未逐条列出（受限于 Changelog.rst 当前的 TBA 状态），blocking_timeout仍是升级后必须了解的核心新设置：

# 1.10.4 时代的临时方案（1.11 起失效） F_BLOCKING_TIMEOUT='10.0' F_FORCE_BLOCKING_TIMEOUT=1 faust worker ... # 1.11 起的正式方案：环境变量 BLOCKING_TIMEOUT='10.0' faust worker ... # 或通过 App 构造参数 app = faust.App(..., blocking_timeout=10.0)

blocking_timeout的官方语义（见 docs/includes/settingref.txt 与 faust/types/settings/settings.py）：

• 类型 / 默认值：float或timedelta，默认None（即不启用），环境变量名为BLOCKING_TIMEOUT，对应 CLI 选项--blocking-timeout；
• 工作原理：当指定该值时，Worker 会启动一个基于信号的周期定时器，仅在事件循环被阻塞超过该时长时触发；检测器是一个后台线程，周期性地「撤销上一个 alarm 并重新布防」，伪代码如下：

while True: # cancel previous alarm and arm new alarm signal.signal(signal.SIGALRM, on_alarm) signal.setitimer(signal.ITIMER_REAL, blocking_timeout) # sleep to wakeup just before the timeout await asyncio.sleep(blocking_timeout * 0.96) def on_alarm(signum, frame): logger.warning('Blocking detected: ...')

• 注意事项：Python 自 3.5 起（PEP 475）会自动重试被信号中断的系统调用，因此该方案对纯 Python 代码是安全的；但用户引入的 C 扩展若未自动重试被中断的系统调用，可能受到负面影响；
• Worker 侧接线：blocking_timeout在 faust/worker.py 中作为 Worker 参数接收并在 debug/检测流程中使用。

1.10 系列中与 1.11 一脉相承的其他演进

为了让升级到 1.11 的读者完整理解「1.10 → 1.11」的连续性，以下 1.10 系列变更也值得关注（全部记录在 docs/history/changelog-1.10.rst）：

1. consumer_api_version/broker_api_version分层配置（1.10.0）

   • 新增consumer_api_version与producer_api_version，默认值均指向broker_api_version（默认'auto'，即取客户端与服务端共同支持的最新版本）；
   • 一个典型用法是broker_api_version="0.10"：由于 Kafka 0.11 才引入 headers 支持，该设置可让所有消息不再携带 headers；
   • 相关说明见 docs/includes/settingref.txt、docs/includes/settingref.txt，源码见 faust/types/settings/settings.py。

2. broker_rebalance_timeout（1.10.0）

   • 类型float/timedelta，默认60.0，环境变量BROKER_REBALANCE_TIMEOUT；
   • 含义：Broker 在判定节点「失联并移出集群」前，等待其完成 rebalance 的最长时间；若集群长期处于反复 rebalance 状态，应调大该值，并同步调大broker_heartbeat_interval；注意 session timeout 不得大于broker_request_timeout；
   • 详见 docs/includes/settingref.txt。

3. consumer_max_fetch_size默认值改为 1MB（1.10.1）

   • 默认值1048576字节，即每分区单次最大拉取量；调优时必须估算单节点被分配的总分区数——例如订阅 4 个各 100 分区的 Topic 且只有一个 Worker 时，最大拉取量可达 4×100MB，rebalance 时残留的 socket 缓冲可能触发「rebalancing loop」；
   • 详见 docs/includes/settingref.txt。

4. 模型序列化性能与嵌套类型支持（1.10.0）

   • 序列化提速约 4 倍、反序列化约 2 倍；相关字段改为惰性加载；
   • 支持任意嵌套类型表达式（如List[Dict[str, List[Set[MyModel]]]]）、命名元组字段、模型 Union（Union[ModelX, ModelY]）；
   • 注意一项向后不兼容变更：Model.to_representation()不再递归遍历子模型，而是原样返回对象（依赖 JSON 序列化器在序列化时调用Model.__json__()）；
   • 同时移除了一批不再需要的属性（Model._options下的models/modelattrs/field_coerce/initfield/polyindex，以及FieldDescriptorT下的generic_type/member_type），若有代码依赖这些内部属性需在升级前清理。

5. Agent 回复头默认开启（1.10.0）

   • use_reply_headers默认变为True，影响Agent.ask/.cast/.map/.kvmap/.join的用户；若 Kafka broker 不支持 headers，可显式关闭：@app.agent(use_reply_headers=False)（对应实现见 faust/agents/agent.py）。

6. 新增BooleanField字段类型（1.10.0）

   • 位于 faust/models/fields.py，注册在字段类型映射中（faust/models/fields.py），用于在Record中声明布尔字段。

升级到 1.11.0 的检查清单

综合Changelog.rst与 1.10 系列变更，升级前建议按以下清单核对：

1. 依赖版本：确认使用 Mode 4.3.2 与robinhood-aiokafka1.1.5（或更新的兼容版本）；
2. 临时方案迁移：若在 1.10.4 中使用了F_BLOCKING_TIMEOUT/F_FORCE_BLOCKING_TIMEOUT环境变量启用阻塞检测，请在升级后改用blocking_timeout设置或BLOCKING_TIMEOUT环境变量，旧变量在 1.11 中将失效；
3. 模型内部属性：检查代码是否引用了Model._options.models、field.generic_type等已移除属性，若引用请改用{field: field.related_models for field in model._options.descriptors if field.related_models}等方式替代；
4. 回复头：确认 Kafka broker 版本 ≥ 0.11（支持 headers），否则为相关 Agent 显式设置use_reply_headers=False；
5. 窗口回调：若在窗口关闭回调中执行异步 IO，1.11.0 中可将其改写为async def，与同步回调共存。

以上所有变更的事实依据均可从仓库内文档与源码中验证：最新变更见 Changelog.rst，历史变更见 docs/history/changelog-1.10.rst 及 docs/history/index.rst，设置项完整参考见 docs/includes/settingref.txt，对应实现分布于 faust/tables/base.py、faust/types/settings/settings.py、faust/worker.py 等模块。

• 流处理
• 消息队列
• 后端