跳转至

使用 Edge Functions、Cron 和队列处理大规模任务

文章背景与核心概要

在构建处理大量数据的应用程序时，单体式执行往往会导致超时、系统崩溃以及糟糕的用户体验。与其依赖更大型的服务器，最优的解决方案是将庞大的任务拆分为小型、可管理的可控单元。本文探讨了如何利用三层架构模式来实现可靠、可扩展的任务处理：即使用 Supabase Edge Functions 提供无服务器计算，使用 Cron 进行定时调度，以及利用数据库 队列（Queues） 来实现可靠的任务分发。

通过结合具体构建 NFL（美国国家橄榄球联盟）新闻聚合器的案例，本文详细展示了如何利用这一架构进行内容抓取、智能路由、AI 辅助分析以及后台任务处理，从而有效隔离故障、避免级联崩溃，并完美契合无服务器（Serverless）环境的各项限制。

三层架构模式

The Three-Layer Pattern

该架构的运作方式类似于一条装配线，通过将复杂的逻辑拆分为职责单一的专注步骤：

The architecture operates like an assembly line, breaking down complexity into focused, single-responsibility steps:

  1. 收集（Collection）： Cron 定时任务触发 Edge Functions，用于发现新涌入的工作并填充主队列。
  2. 分发（Distribution）： Cron 定常例程根据来源需求，将任务从集中式队列路由到专用的处理队列中。
  3. 处理（Processing）： 专用的工作器从各自的分配队列中拉取任务，并执行特定任务（例如网页抓取、AI 分类或向量嵌入生成）。
  1. Collection: Cron jobs trigger Edge Functions that discover incoming work and populate the main queues.
  2. Distribution: Cron routines route tasks from centralized queues into specialized processing queues based on source requirements.
  3. Processing: Specialized workers pull tasks from their assigned queues, performing dedicated tasks (e.g., HTML scraping, AI classification, or vector embedding generation).

真实案例：构建 NFL 新闻聚合器

Real Example: Building an NFL News Aggregator

构建一个监控数十个新闻网站、每天处理数百篇文章、调用 OpenAI 进行内容分析、生成向量嵌入并将所有内容高效存储的仪表板，很快就会变得十分复杂。如果没有适当的故障隔离机制，单个网页的损坏就可能导致整个处理流水线崩溃。

Building a dashboard that monitors dozens of news sites, processes hundreds of daily articles, invokes OpenAI for content analysis, generates vector embeddings, and stores everything efficiently can quickly become complex. Without proper fault isolation, a single broken webpage can crash the entire processing pipeline.

利用 Supabase 集成的技术栈，可以提供具备强大弹性的架构来处理此类数据流。

Using Supabase's integrated stack provides a resilient architecture to handle this data flow.

搭建基础架构

Setting Up the Foundation

Supabase 的核心是一个可扩展且值得信赖的 PostgreSQL 数据库。应用程序的架构包含内容表、队列表、实体映射以及关系表：

At the core of Supabase is a scalable, dependable PostgreSQL database. The application's schema includes content tables, queue tables, entity mappings, and relationship tables:

create table articles (
  url text unique not null,
  headline text,
  content text,
  embedding vector(1536)
);

收集：发现新内容

Collection: Finding New Content

收集层按计划的时间间隔发现原始内容。每 30 分钟触发一次的 Cron 任务将启动收集器：

The collection layer discovers raw content on a scheduled interval. A cron job triggers every 30 minutes to initiate the collector:

SELECT cron.schedule(
  'nfl-collector',
  '*/30 * * * *',
  $$SELECT net.http_post(url := 'https://your-project.supabase.co/functions/v1/collect-content')$$)
);

抓取与过滤

Scraping and Filtering

在 Edge Function 内部，过滤器可确保仅摄入相关的新闻文章，同时忽略促销材料和视频：

Inside the Edge Function, filters ensure only relevant news articles are ingested while promotional material and videos are ignored:

function isRelevantArticle(url: string): boolean {
  return url.includes('/news/') && !url.includes('/video/')
}

安全地处理相对 URL 并动态进行条目去重：

Handle relative URLs securely and deduplicate entries dynamically:

if (href.startsWith('/')) {
  href = BASE_URL + href
}

const seen = new Set<string>()
if (!seen.has(href)) {
  seen.add(href)
  articles.push({ url: href, site: 'nfl' })
}

与其在内存中处理复杂的去重逻辑，不如将重复检查卸载到数据库约束层：

Instead of handling complex deduplication logic in-memory, offload duplicate checks to the database constraint layer:

const { error } = await supabase.from('articles').insert({ url, site })
if (error && !error.message.includes('duplicate')) {
  console.error(`Error inserting: ${url}`, error)
}

分发：智能路由

Distribution: Smart Routing

不同的内容来源需要自定义的解析结构。分发器每 5 分钟运行一次，将未经处理的文章路由到特定来源的队列中：

Different content sources require custom parsing structures. The distributor runs every 5 minutes to route unprocessed articles to source-specific queues:

SELECT cron.schedule(
  'distributor',
  '*/5 * * * *',
  $$SELECT net.http_post(url := 'https://your-project.supabase.co/functions/v1/distribute-work')$$)
);

分发函数查询未处理的记录并对其进行相应的路由：

The distribution function queries unprocessed records and routes them accordingly:

const { data } = await supabase
  .from('articles')
  .select('url, site')
  .is('headline', null) // Missing headline means unprocessed
  .limit(50)

if (article.site === "nfl") {
  await supabase.from("nfl_queue").insert({ url: article.url });
} else if (article.site === "espn") {
  await supabase.from("espn_queue").insert({ url: article.url });
}

处理：繁重的工作

Processing: The Heavy Lifting

高优先级的信息源（例如 NFL 新闻）可以更频繁地进行处理——例如通过 Cron 每 15 秒处理一次：

High-priority feeds like NFL news can be processed more frequently—for instance, every 15 seconds via cron:

SELECT cron.schedule(
  'nfl-processor',
  '*/15 * * * *',
  $$SELECT net.http_post(url := 'https://your-project.supabase.co/functions/v1/process-nfl')$$)
);

队列消费与解析

Queue Consumption & Parsing

处理程序按顺序逐个拉取条目，以舒适地契合 Edge Function 的超时限制：

Processors pull items sequentially to comfortably fit within Edge Function timeout limits:

const { data } = await supabase
  .from('nfl_queue')
  .select('id, url')
  .eq('processed', false)
  .order('created_at')
  .limit(1)

const headline = $('h1').first().text().trim()
const content = $('.article-body').text().trim() || $('article').text().trim()

finally 安全模式

The finally Safety Pattern

为了防止在发生故障的有效载荷上陷入无限重试循环，请务必在 finally 块中将队列项标记为已处理：

To prevent infinite retry loops on failing payloads, always mark queue items as processed in a finally block:

try {
  // Process article content, generate embeddings, and analyze via AI
} finally {
  await supabase.from("nfl_queue")
    .update({ processed: true })
    .eq("id", item.id);
}

使用 Sentry 进行监控

Monitoring with Sentry

将 Sentry 集成到你的 Edge Functions 中，以便捕获详细的故障上下文而不会丢失追踪功能：

Integrate Sentry into your Edge Functions to capture detailed failure context without losing tracking capabilities:

import { captureException, init } from 'https://deno.land/x/sentry/index.js'

init({
  dsn: Deno.env.get('SENTRY_DSN'),
  environment: Deno.env.get('ENVIRONMENT') || 'production',
})

try {
  const content = await scrapeArticle(url);
  const analysis = await classifyArticle(headline, content);
  await storeArticle(article, analysis);
} catch (error) {
  captureException(error, {
    tags: {
      function: "nfl-processor",
      site: article.site
    },
    extra: {
      url: article.url,
      queueId: queueItem.id
    }
  });
  console.error(`Failed to process ${url}:`, error);
} finally {
  await supabase.from("nfl_queue")
    .update({ processed: true })
    .eq("id", queueItem.id);
}

处理用户交互

Processing User Interactions

用户生成的事件（例如点击、分享或收藏）不应阻塞用户界面。应当对交互进行异步排队并批量处理：

User-generated events (e.g., clicks, shares, or saves) should not block the user interface. Queue interactions asynchronously and process them in batches:

await supabase.from('interaction_queue').insert({
  article_url: url,
  user_id: userId,
  interaction_type: 'share',
})

每 2 分钟运行一次后台 Cron 任务，即可在后台干净利落地计算趋势并更新评分指标：

Run a background cron job every 2 minutes to calculate trends and update score metrics cleanly in the background:

SELECT cron.schedule(
  'process-interactions',
  '*/2 * * * *',
  $$SELECT net.http_post(url := 'https://your-project.supabase.co/functions/v1/process-interactions')$$)
);

AI 驱动的内容评分

AI-Powered Content Scoring

为了自动凸显重要新闻，可以使用来自 OpenAI 的结构化 JSON 输出对内容进行分类：

To automatically surface important news, classify content using structured JSON outputs from OpenAI:

const prompt = `Analyze this headline: "${headline}"
Return JSON: {"context": "trade|injury|etc", "score": 1-9}`;

const result = await openai.chat.completions.create({
  model: "gpt-3.5-turbo",
  response_format: { type: "json_object" }
});

针对高开销操作的后台任务

Background Tasks for Expensive Operations

对于诸如批量 AI 分析或向量嵌入生成等资源密集型任务，可以在发送即时响应后，让 Edge Functions 异步处理请求：

For resource-heavy tasks like bulk AI analysis or vector embedding generation, let Edge Functions handle requests asynchronously after sending an immediate response:

const article = await scrapeAndStore(url);

const backgroundTasks = [
  generateEmbedding(article),
  analyzeWithAI(article),
  updateRelatedContent(article)
];

Promise.all(backgroundTasks).catch(error => {
  captureException(error, {
    tags: { operation: "background-tasks" },
    extra: { articleUrl: url }
  });
});

await markAsProcessed(queueItem.id);

为什么这种方法行之有效

Why This Works

通过拥抱无服务器的限制而不是与之对抗，这种模式提供了强大的可扩展性： - 时间限制通过逐个处理项目得到尊重。 - 速率限制通过自定义的 cron 间隔进行管理。 - 故障被隔离，从而防止了级联故障。 - Supabase 原语消除了对外部队列管理器或独立作业调度程序的需要，以初创公司的简洁性提供了企业级的可靠性。

By embracing serverless constraints rather than fighting them, this pattern delivers robust scalability: - Time limits are respected by processing items individually. - Rate limits are managed via custom cron intervals. - Failures are isolated, preventing cascading outages. - Supabase primitives eliminate the need for external queue managers or standalone job schedulers, offering enterprise reliability with startup simplicity.