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

> Documentation Index
> Fetch the complete documentation index at: https://x-algorithm.vergego.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Thunder Kafka 摄入

## 这一页回答什么

本页说明 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`）。

## 两种角色

```mermaid
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`。

### 视频资格

```rust
// 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` —— 追平门控与就绪

## 相关页面

- [Thunder 站内帖子内存库](https://x-algorithm.vergego.ai/concepts/thunder-in-network-store/) —— 查询路径与服务端常量
- [PostStore](https://x-algorithm.vergego.ai/entities/post-store/) —— `PostStore` 结构体细节
- [端到端数据流：一条帖子从发布到被推荐](https://x-algorithm.vergego.ai/concepts/end-to-end-dataflow/) —— Kafka 摄入是推荐链路的起点
- [home-mixer 编排层](https://x-algorithm.vergego.ai/concepts/home-mixer-orchestration/) —— `ThunderSource` 如何调用 Thunder
- [系统架构总览](https://x-algorithm.vergego.ai/concepts/system-architecture/) —— Thunder 作为站内候选源的角色

Source: https://x-algorithm.vergego.ai/concepts/thunder-kafka-ingestion/index.mdx