月饼 TV 数据处理流程的优化

[HaroldFinchNYC]

数据处理流程优化：从 11.5 小时到 40 分钟的性能提升

作者: MoonCake TV 技术团队 （就我自己）

性能提升: 🚀 17 倍加速 (11.5 小时 → 40 分钟)

---

简单点，说人话，就是我以前的数据处理流程，没有规划，有什么用什么；核心还是太烂了；现在经过和 ai 的多轮对话，终于有了改进

下面是经过 ai 润色的

📊 核心成果

• 处理时间： 从 11.5 小时降至 40 分钟
• 性能提升： 17 倍加速
• 架构简化： 从 4 层服务减少到 2 层
• 数据库精简： 从 3 个数据库整合到 1 个
• 代码简化： 移除了整个 meilisearch-sync worker 服务

---

🎯 问题背景

旧架构的痛点

在优化之前，我们的数据处理流程存在以下严重问题：

1. 处理速度慢： 完整的数据同步需要 11.5 小时
2. 架构复杂： 数据需要经过 4+ 个服务 才能到达用户
3. 数据库冗余： 同时维护 Cloudflare D1、Temporal PostgreSQL 和 meili-search-service PostgreSQL 三个数据库
4. 网络开销大： Temporal Workers → Cloudflare Workers → D1 → meilisearch-sync → meili-search-service 的多次网络跳转
5. 耦合度高： 所有服务都依赖 Cloudflare Workers 作为中间层

性能瓶颈分析

旧架构数据流：
Source APIs (数据源)
    ↓
Temporal Workers (数据抓取)
    ↓ HTTP POST /cf-workers/upsert-batch
Cloudflare Workers (中间层) ← 瓶颈 1: 网络延迟
    ↓ 写入 D1 数据库
Cloudflare D1 (SQLite) ← 瓶颈 2: D1 性能限制
    ↓ GET /cf-workers/search-todos-*
meilisearch-sync Worker ← 瓶颈 3: 额外的同步层
    ↓ POST to meili-search-service
meili-search-service Meilisearch
    ↓
Cloudflare Workers
    ↓
用户查询

关键瓶颈：

• Cloudflare D1 性能限制： SQLite 数据库在大批量写入时性能不足
• 多次网络往返： 每条数据需要经过 4 次网络传输
• 不必要的中间层： Cloudflare Workers 和 meilisearch-sync worker 造成额外开销
• JSON 序列化开销： m3u8_urls 字段需要多次 stringify/parse

---

🚀 优化方案

核心思想：单向数据流 (Uni-Directional Flow)

我们采用了单向数据流的设计理念，彻底简化了数据处理架构：

新架构数据流：
Source APIs (数据源)
    ↓
Temporal Workers (数据抓取 + 解析)
    ↓ HTTP POST /v1/upsert-batch (JWT 认证)
meili-search-service PostgreSQL ---> Meilisearch Server (同一个服务器内部)
    ↓
Cloudflare Workers
    ↓
用户查询 (通过 m3u8-s1 代理)

关键优化措施

1. 移除 Cloudflare D1 数据库

之前：

• Temporal Workers 写入 Cloudflare D1
• D1 作为临时存储，再同步到 meili-search-service

现在：

• Temporal Workers 直接写入 meili-search-service PostgreSQL
• PostgreSQL 作为唯一数据源 (Single Source of Truth)

收益：

• ✅ 减少一次网络往返
• ✅ PostgreSQL 性能远超 D1 (批量写入优化)
• ✅ 无需维护额外的数据库服务

2. 移除 meilisearch-sync Worker

之前：

• 需要专门的 worker 从 D1 读取数据并同步到 Meilisearch
• 需要维护 is_indexed 标志位
• 需要轮询检查待索引数据

现在：

• 完全移除这个 worker
• 数据直接存入 PostgreSQL ，按需索引

收益：

• ✅ 减少一个服务的维护成本
• ✅ 消除轮询开销
• ✅ 简化代码逻辑

3. 优化数据类型：JSONB vs TEXT

之前 (Cloudflare D1)：

// m3u8_urls 存储为 TEXT
m3u8_urls: JSON.stringify(item.m3u8_urls); // 需要序列化

现在 (PostgreSQL)：

// m3u8_urls 存储为 JSONB
m3u8_urls: item.m3u8_urls; // 直接传递对象，无需序列化

收益：

• ✅ PostgreSQL 原生支持 JSONB ，查询和索引更高效
• ✅ 减少 JSON.stringify/parse 的 CPU 开销
• ✅ 可以直接对 JSON 字段进行查询和索引

4. 批量写入优化

新的 PostgreSQL Upsert 逻辑：

INSERT INTO table_name (
  id, field1, field2, field3, ...
) VALUES ($1, $2, $3, $4, ...)
ON CONFLICT (unique_key1, unique_key2)
DO UPDATE SET
  m3u8_urls = EXCLUDED.m3u8_urls,
  updated_at = EXCLUDED.updated_at

关键优化点：

• 使用 ON CONFLICT 实现 upsert （插入或更新）
• 批量处理，减少数据库往返次数
• 复合唯一索引避免重复数据

---

📈 性能对比

处理时间对比

指标	旧架构	新架构	提升比例
总处理时间	11.5 小时	40 分钟	17.25x
网络往返次数	4+ 次	1 次	4x 减少
服务层级	4 层	2 层	2x 简化
数据库数量	3 个	1 个	3x 精简

架构复杂度对比

维度	旧架构	新架构
数据源	Cloudflare D1 + PostgreSQL	PostgreSQL 单一数据源
中间服务	Cloudflare Workers + meilisearch-sync	无
数据流向	循环依赖	单向流动
维护成本	高（多个服务）	低（精简架构）

---

🏗️ 技术实现细节

1. meili-search-service 数据库设计

** 表结构：**

CREATE TABLE IF NOT EXISTS dazahui (
  id SERIAL PRIMARY KEY,
  mc_id VARCHAR(50) UNIQUE NOT NULL,
  title TEXT NOT NULL,
  m3u8_urls JSONB,  -- 关键：使用 JSONB 类型
  .......

  CONSTRAINT unique_xxx UNIQUE(xxx, xxxx)
);

2. Temporal Workers 改造

旧代码 (写入 Cloudflare Workers)：

await fetch("https://cf-workers/upsert-batch", {
  method: "POST",
  body: JSON.stringify({
    items: items.map((item) => ({
      ...item,
      m3u8_urls: JSON.stringify(item.m3u8_urls), // 需要序列化
    })),
  }),
});

新代码 (直接写入 meili-search-service)：

await fetch(`/v1/upsert-batch`, {
  method: "POST",
  body: JSON.stringify({
    items, // m3u8_urls 保持对象格式，无需序列化
  }),
});

关键改进：

• ✅ 直接连接 meili-search-service (减少网络跳转)
• ✅ 无需 JSON.stringify (减少 CPU 开销)
• ✅ 使用专用 JWT 认证 (安全性提升)

3. API 端点设计

数据写入 API (JWT 保护)：

POST /v1/upsert-batch
Authorization: Bearer <MEILI_INTERNAL_JWT_SECRET>

Request:
{
  "items": [
    {
      "title": "电影标题",
      "m3u8_urls": {"第 1 集": "url1", "第 2 集": "url2"},
      "language": "zh",
      ...
    }
  ]
}

Response:
{
  "code": 200,
  "message": "Batch upsert completed",
  "data": {
    "processed": 150,
    "failed": 0
  }
}

---

🎉 优化成果

1. 性能提升

• 处理速度： 从 11.5 小时降至 40 分钟，提升 17 倍
• 吞吐量： 单位时间处理的数据量显著增加
• 响应时间： 用户查询响应时间保持稳定

2. 架构简化

• 服务数量： 从 6 个服务减少到 4 个服务
• 数据库： 从 3 个数据库整合到 1 个 PostgreSQL
• 代码量： 移除了整个 meilisearch-sync worker (~500 行代码)

3. 运维成本降低

• 监控点减少： 无需监控 D1 和 meilisearch-sync worker
• 故障点减少： 服务链路缩短，故障排查更简单
• 资源消耗降低： 减少了 Cloudflare Workers 的调用次数

4. 可维护性提升

• 单向数据流： 数据流向清晰，易于理解和调试
• 单一数据源： PostgreSQL 作为唯一真实数据源，避免数据不一致
• 代码清晰： 移除了复杂的同步逻辑和状态管理

---

💡 关键经验总结

1. 性能优化的本质是减少不必要的中间层

• 每增加一层服务，就增加一次网络往返和序列化开销
• 直接连接往往比多层代理更高效

2. 选择合适的数据类型很重要

• PostgreSQL JSONB vs TEXT：原生支持 vs 手动序列化
• 数据库选择：Cloudflare D1 (SQLite) vs PostgreSQL (批量写入优化)

3. 单向数据流的威力

• 避免循环依赖，数据流向清晰
• 减少同步状态管理的复杂度
• 更容易进行性能优化

4. 批量处理 > 单条处理

• 使用 ON CONFLICT 实现 upsert 批量操作
• 减少数据库往返次数
• 利用数据库的批量写入优化

---

🔮 未来优化方向

Phase 5-6: 查询层优化 (计划中)

虽然数据写入性能已大幅提升，但仍有进一步优化空间：

1. 添加查询 API 到 meili-search-service

   • GET /v1/search-sql?keyword=xxx (SQL LIKE 搜索)
   • GET /v1/random?limit=20 (随机内容)
   • GET /v1/mc_item/:mc_id (获取单个内容)

2. 更新 m3u8-s1 为纯代理层

   • 移除所有 /boss/* 路由
   • 将 /v1/* 端点改为代理到 meili-search-service
   • 保留 Cloudflare KV 缓存层 (边缘计算优势)

3. Meilisearch 自动索引

   • 实现从 PostgreSQL 到 Meilisearch 的自动同步
   • 优化全文搜索性能

---

📚 技术栈

核心技术：

• Temporal.io： 分布式任务调度和工作流编排
• PostgreSQL： 主数据库，支持 JSONB 类型
• Meilisearch： 全文搜索引擎
• Cloudflare Workers + KV： 边缘计算和缓存层
• Hono.js： 高性能 Web 框架

优化技术：

• JSONB 数据类型（ PostgreSQL 原生 JSON 支持）
• ON CONFLICT 批量 Upsert
• 直连架构（减少网络跳转）
• JWT 认证（服务间安全通信）

---

🏁 结论

通过彻底重新设计数据处理架构，我们实现了：

• ✅ 17 倍性能提升 (11.5 小时 → 40 分钟)
• ✅ 架构大幅简化 (4 层 → 2 层服务)
• ✅ 数据库整合 (3 个 → 1 个)
• ✅ 代码精简 (移除整个 meilisearch-sync worker)
• ✅ 运维成本降低 (更少的监控点和故障点)

这次优化充分证明了： 正确的架构设计比单纯的代码优化更能带来质的飞跃。

[itechify]

MoonCake TV 项目在哪，做什么的？

[itechify]

https://github.com/MoonCakeTV/MoonCakeTV