Skip to content
X 推荐算法

Thunder Kafka 摄入

Thunder 从 Kafka 实时摄入帖子：feeder（v1）转码，serving（v2）写库，追平后才就绪，删除墓碑与分片裁剪

Updated

这一页回答什么

本页说明 Thunder 如何从 Kafka 实时摄入帖子的创建事件和删除事件，内容包括：两种角色如何分工、事件如何反序列化、启动时如何追平、旧帖和删除墓碑如何裁剪。

核心结论

  1. 同一个二进制有两种角色。is_serving = false 时，进程是 feeder：它消费原始 Thrift 帖子事件，转成精简的 proto InNetworkEvent，写入 innetwork_post topic。is_serving = true 时，进程是 serving 实例：它消费 innetwork_post，写入内存中的 PostStore（thunder/kafka_utils.rs:71-161）。
  2. 每个 serving 实例都读取全量数据。每次启动时，consumer group id 都拼接一个随机 UUID（kafka_utils.rs:72-73），因此各实例不分摊分区。实例没有已提交的 offset，按 auto_offset_reset（默认 earliest）从头读取（thunder/args.rs:9-10）。每个实例都持有一份完整的两天窗口。
  3. feeder 过滤三类事件：直接丢弃 nullcast 帖（不进入时间线的定向推广帖）；忽略超过保留期的帖子的删除事件；引用帖被删除时，删除引用它的那条帖子（thunder/kafka/tweet_events_listener.rs:220-263）。
  4. 视频资格门槛是 5 秒：帖子只有一个媒体，且该媒体是时长不少于 5000 ms 的视频时，feeder 才标记 has_video（tweet_events_listener.rs:55-74）。旧版引用的常量 MIN_VIDEO_DURATION_MS 已删除，代码改用字面量。
  5. 追平后才对外服务：每个线程在“剩余 lag 小于分区数 × 批大小”时发送一次信号。main 收齐全部线程的信号后，才调用 finalize_init() 并进入就绪状态（thunder/kafka/tweet_events_listener_v2.rs:182-229；thunder/main.rs:94-116）。
  6. 追平后写入串行执行：稳态时，所有摄入线程共用一个许可数为 1 的信号量，同一时刻只有一个批次在写库（tweet_events_listener_v2.rs:52、206-210）。旧版的许可数是 3。
  7. 裁剪分 5 片轮转：后台每 2 分钟裁剪一次。每次只处理 user_id % 5 等于本轮编号的作者，外加删除墓碑队列（thunder/posts/post_store.rs:479-606；thunder/config.rs:14）。

两种角色

flowchart LR
    RAW[("tweet_events<br/>Thrift TweetEvent")] -->|"feeder · is_serving=false"| FEED["tweet_events_listener<br/>转码 + 过滤"]
    FEED -->|"proto InNetworkEvent"| INE[("innetwork_post")]
    INE -->|"serving · is_serving=true"| SRV["tweet_events_listener_v2<br/>反序列化 + 写库"]
    SRV --> STORE[("PostStore")]
角色 is_serving 消费 topic 入口 输出 offset
feeder false tweet_events（Thrift） start_tweet_event_processing 写入 innetwork_post 关闭自动提交；每批发送完成后手动提交（tweet_events_listener.rs:340-370）
serving true innetwork_post（proto） start_tweet_event_processing_v2 写入 PostStore 自动提交；每次启动时 group id 唯一

topic 名与地址常量位于 kafka_utils.rs:18-23。仓库中的地址是占位符 example.invalid。

认证方式

innetwork_post 的消费端和生产端各自可选两种认证方式：提供 --in-network-events-consumer-mtls-zone / --in-network-events-producer-mtls-zone 时，使用该 zone 的 mTLS；否则使用 SCRAM 用户名和密码（kafka_utils.rs:25-52、75-97、135-158；args.rs:72-76）。两端的认证方式相互独立。SASL 密码从环境变量 SASL_PASSWORD / PRODUCER_SASL_PASSWORD 或命令行参数读取（kafka_utils.rs:60-69）。

多线程与分区

两个 listener 的分配方式相同：按 kafka_num_threads（默认 32）启动线程，把分区按块平均分给各线程，partitions_per_thread = 分区数.div_ceil(线程数)（tweet_events_listener.rs:97-101；tweet_events_listener_v2.rs:28-32）。分区数默认值为：feeder 120（tweet_events_num_partitions），serving 32（kafka_tweet_events_v2_num_partitions）（args.rs:30-31、54-55）。

每个线程有一个 consumer，并启动一个分区 lag 监控任务，周期为 lag_monitor_interval_secs（默认 30 秒）（tweet_events_listener_v2.rs:79-85；args.rs:39-40）。线程意外退出或 consumer 创建失败时，进程直接 panic（tweet_events_listener_v2.rs:96-106）。

反序列化

thunder/deserializer.rs 提供三个函数，分别解析 Thrift TweetEvent、Thrift Event 和 proto InNetworkEvent。kafka/utils.rs 的 deserialize_kafka_messages() 是通用批量解析器：单条消息解析失败时，它只记录 KAFKA_MESSAGES_FAILED_PARSE 并跳过该条（thunder/kafka/utils.rs:20-39）。

Thrift 结构体的 Rust 定义位于新增的 thunder/schema/ 下（tweet.rs、tweet_events.rs、media_entity.rs 等）。这些文件由 Thrift IDL 生成，共约 2.9 万行。

feeder：转码与过滤

process_message_batch()（tweet_events_listener.rs:189-338）逐条处理事件：

  • TweetCreateEvent：nullcast 为真时跳过（tweet_events_listener.rs:220-224）。其余事件转成 LightPost，字段包括帖子 id、作者 id、created_at_secs、回复对象（in_reply_to_status_id / in_reply_to_user_id）、转发来源（share.source_status_id / source_user_id）、conversation_id、has_video。is_retweet 取决于 share 是否存在，is_reply 取决于 reply 是否存在（tweet_events_listener.rs:226-244）。
  • TweetDeleteEvent：帖子发布时间早于保留期时忽略（tweet_events_listener.rs:246-260）；否则记入待删列表。
  • QuotedTweetDeleteEvent：把引用了被删帖的那条帖子 quoting_tweet_id 记入待删列表（tweet_events_listener.rs:261-263）。
  • 忽略其他事件类型。

随后，feeder 把每条创建事件封装成 InNetworkEvent::TweetCreateEvent，把每条删除事件封装成 InNetworkEvent::TweetDeleteEvent，deleted_at 填当前时间。它逐条异步发送，并等待全部发送完成（tweet_events_listener.rs:269-323）。feeder 不持有 PostStore。

视频资格

// thunder/kafka/tweet_events_listener.rs:55-74（节选）
let [first_media] = media.as_slice() else { return false; };
let Some(MediaInfo::VideoInfo(video_info)) = first_media.media_info.as_ref() else {
    return false;
};
video_info.duration_millis.map(|d| d >= 5_000).unwrap_or(false)

let [first_media] = ... 要求帖子恰好有一个媒体。因此，多图帖和图加视频的帖子都不算视频帖。

serving：写入内存库

拆分批次

deserialize_batch()（tweet_events_listener_v2.rs:113-161）把 InNetworkEvent 分成创建列表和删除列表。is_reply 有兜底判断：事件自带 is_reply、有 in_reply_to_post_id 或有 in_reply_to_user_id，任一条件成立即视为回复（tweet_events_listener_v2.rs:145-147）。

写库

process_tweet_events_v2()（tweet_events_listener_v2.rs:163-239）循环执行以下步骤：

  1. 每次调用 poll(batch_size)，kafka_batch_size 默认为 1000（args.rs:12-13）。
  2. 尚未追平时，读取各分区 lag。总 lag 小于 分区数 × batch_size 时，标记为已追平并准备发送信号（tweet_events_listener_v2.rs:182-197）。
  3. 缓冲区攒够一批后，在 spawn_blocking 中依次调用 insert_posts() 和 mark_as_deleted()（tweet_events_listener_v2.rs:212-222）。
  4. 追平之后，每批写入前先获取全局信号量的许可（许可数 1）；追平之前不获取许可（tweet_events_listener_v2.rs:206-210）。
  5. 本批写完后，如果第 2 步已标记追平，向 main 发送一次信号（tweet_events_listener_v2.rs:224-229）。

追平期间不限流，让内存库尽快填满。追平后，所有线程串行写库，把 CPU 让给查询。许可数从 3 改为 1，说明上游进一步减少了稳态摄入对查询的干扰（推测）。

启动门控

在 is_serving 模式下，main 循环 kafka_num_threads 次 rx.recv()。收齐信号后，它调用 finalize_init()，记录 INIT_DURATION_SECONDS，再启动统计日志和自动裁剪（main.rs:94-114）。is_serving = false（feeder）时，进程不等待，直接就绪。

删除事件的处理

mark_as_deleted()（post_store.rs:111-125）对每个删除事件执行三个动作：

  1. 从 posts 删除正文。之后，任何队列中指向该帖的 TinyPost 回查都会失败，查询自然跳过它。
  2. 在 deleted_posts 中记录墓碑。之后迟到的创建事件会在 insert_posts_internal 中被拒绝（post_store.rs:163-165）。
  3. 在特殊作者 DELETE_EVENT_KEY = 0 的原创队列中追加一条 TinyPost(帖子 id, deleted_at)（config.rs:10）。这条记录用于按时间裁剪墓碑。

finalize_init() 最后再把墓碑表中的帖子从 posts 清除一遍，以处理追平期间“删除先到、创建后到”的乱序（post_store.rs:142-154）。

保留与裁剪

  • 写入时过滤：insert_posts() 只保留 created_at 在过去 retention_seconds 内且不在未来的帖子，并按时间升序写入（post_store.rs:127-140）。保留期默认为 172800 秒，即 2 天（args.rs:48-49）。
  • 每个队列的上限：单个作者的单个队列超过 5000 条时，新帖入队会挤出最旧的一条（post_store.rs:181-185；config.rs:12）。
  • 启动裁剪：finalize_init() 先给所有队列排序，再依次执行 5 轮裁剪，覆盖全部分片（post_store.rs:142-154）。
  • 周期裁剪：start_auto_trim(2) 每 2 分钟执行一轮，错过的 tick 直接跳过（main.rs:109；post_store.rs:479-499）。

trim_old_posts(trim_iter)（post_store.rs:501-606）在 spawn_blocking 中对三个队列表各执行一次以下步骤：

  1. 先处理 DELETE_EVENT_KEY 队列：弹出超过保留期的墓碑，并同时从 posts 和 deleted_posts 删除（post_store.rs:526-538）。
  2. 其余作者只处理 user_id % 5 == trim_iter % 5 的那一片（post_store.rs:547-549）。
  3. 对选中的作者，从队首弹出“超过保留期”或“队列超过 5000 条”的帖子，并从 posts 删除（post_store.rs:552-563）。
  4. 队列容量超过长度的两倍时，收缩到长度的 1.5 倍；移除队列为空的作者（post_store.rs:565-577）。

分片的代价是：同一作者每 5 轮才轮到一次，即每 10 分钟一次。因此，过期帖在内存中最多多停留约 10 分钟（按代码推断）。下游另有 48 小时的 AgeFilter，所以这不影响信息流。

未接入的部分

thunder/o2/mod.rs 定义了一个 S3 兼容对象存储客户端 O2Client。args.rs 也有 --o2-bucket（默认 innetwork-posts）和 --o2-prefix（args.rs:78-82）。但当前 thunder 代码中没有任何地方调用它。源码未说明它的用途，推测是为快照加载预留。

设计决策

决策 理由
分离 feeder 与 serving Thrift 解析和过滤只执行一次；serving 实例只解码精简的 proto
serving 使用唯一 group id 并从头读取 每个实例都需要完整的两天窗口，不能分摊分区
追平期不限流，稳态串行写入 启动后尽快可用；上线后把 CPU 让给查询
追平后才就绪 避免半空的内存库返回不完整的站内候选
删除时立即删除正文并记录墓碑 查询无需再查墓碑表；迟到的创建事件也能被拦截
墓碑挂在特殊作者 0 下 复用按时间裁剪队列的逻辑，墓碑自然过期
裁剪分 5 片 单轮裁剪只扫描五分之一的作者，缩短每轮持锁时间
单条消息解析失败时跳过 个别坏消息不中断整批
处理线程退出时 panic 摄入停止等于数据过期；宁可让编排系统重启进程

FAQ

Q：删除事件早于创建事件到达时会怎样？ A：mark_as_deleted 先记录墓碑。之后的创建事件在 insert_posts_internal 中因墓碑而被跳过（post_store.rs:163-165）。finalize_init 再兜底清除一遍。

Q：重启一个 serving 实例需要多久？ A：实例需要从 innetwork_post 头部重放到追平。耗时取决于 topic 保留的数据量，并记录在 INIT_DURATION_SECONDS 中。源码中没有具体数值。

Q：feeder 和 serving 是同一个二进制吗？ A：是。thunder 是一个二进制，由 --is-serving 决定角色（args.rs:57-58；kafka_utils.rs:71、116）。

源码锚点

  • thunder/kafka_utils.rs:18-161 —— topic、认证、按角色分流
  • thunder/args.rs:6-91 —— Kafka、保留期、线程与分区默认值
  • thunder/kafka/tweet_events_listener.rs:55-74 —— 视频资格
  • thunder/kafka/tweet_events_listener.rs:189-338 —— feeder 转码、过滤、发送
  • thunder/kafka/tweet_events_listener_v2.rs:42-239 —— serving 线程、追平信号、写库
  • thunder/posts/post_store.rs:111-154 —— 删除墓碑、写入过滤、启动收尾
  • thunder/posts/post_store.rs:479-606 —— 自动裁剪与分片
  • thunder/main.rs:90-116 —— 追平门控与就绪

相关页面

X 推荐算法

Type to search…

↑↓ navigate↵ selectEsc close