异步文件处理队列

要点

文件解析、OCR、音视频转码耗时长，放在 HTTP 请求里同步执行会超时
Cloudflare Workers 有 CPU 时间限制（免费 10ms、付费 30s），大文件处理不够用
Cloudflare Queue 提供生产者-消费者模型，HTTP 接口写消息，消费者 Worker 在后台处理
D1 存任务记录（状态、进度、结果路径），Queue 做消息分发
前端通过轮询或 SSE 获取进度，完成后下载结果

1. 为什么需要异步

前面几篇讲了文件上传、PDF 解析、图片 OCR、音频转文本。这些操作有一个共同特点：慢。一个 50MB 的 PDF 解析可能需要十几秒，一段音频的语音转写可能要几十秒。

同步处理会碰到三个问题：

Workers CPU 时间限制——免费计划 10ms，付费计划 30 秒，大文件解析很容易超限
HTTP 超时——客户端连接通常 30-60 秒超时，用户不会等半分钟
阻塞并发——一个耗时请求占用 Worker 实例，影响其他请求

解决思路：收到请求后马上返回，把耗时工作丢到后台。前端拿到任务 ID，定期查询进度。

同步：上传 → 解析 → OCR → 返回（可能 60 秒 → 超时）
异步：上传 → 返回 taskId → 后台处理 → 前端轮询/SSE 获取结果

2. Queue 基础配置

Cloudflare Queue 的工作模式：生产者写入消息，消费者取出处理。

生产者（HTTP Worker）→ Queue → 消费者（Queue Worker）

在 wrangler.jsonc 中配置：

// wrangler.jsonc
{
  "queues": {
    "producers": [{
      "binding": "FILE_QUEUE",
      "queue": "file-processing"
    }],
    "consumers": [{
      "queue": "file-processing",
      "max_batch_size": 5,
      "max_batch_timeout": 10,
      "max_retries": 3,
      "dead_letter_queue": "file-processing-dlq"
    }]
  }
}

max_batch_size：每次拉取多少条消息一起处理
max_retries：失败后最多重试几次
dead_letter_queue：重试耗尽后消息的去向

需要注意的限制：单条消息最大 128 KB（消息里只放文件引用，内容从 R2 读），每批最多 10 个消息。

类型声明：

// src/types.ts
export type Env = {
  Bindings: {
    DB: D1Database
    BUCKET: R2Bucket
    FILE_QUEUE: Queue<FileTaskMessage>
  }
}
 
export type FileTaskMessage = {
  taskId: string
  userId: string
  fileKey: string
  fileType: string
}

3. 任务状态设计

任务从提交到完成，经历 pending → processing → completed / failed 四个状态，存到 D1 供前端随时查询。

CREATE TABLE tasks (
  id TEXT PRIMARY KEY,
  user_id TEXT NOT NULL,
  file_key TEXT NOT NULL,
  file_type TEXT NOT NULL,
  status TEXT NOT NULL DEFAULT 'pending',
  progress INTEGER NOT NULL DEFAULT 0,
  result_key TEXT,
  error_message TEXT,
  retry_count INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL,
  updated_at TEXT NOT NULL
);
 
CREATE INDEX idx_tasks_user_status ON tasks(user_id, status);

progress：0-100 的百分比，前端用来展示进度条
result_key：处理结果在 R2 中的路径

4. 生产者：提交任务

HTTP 接口做三件事：上传文件到 R2、写任务记录到 D1、发消息到 Queue，然后立刻返回。

// src/index.ts
import { Hono } from 'hono'
import type { Env, FileTaskMessage } from './types'
 
const app = new Hono<{ Bindings: Env['Bindings'] }>()
 
app.post('/api/tasks', async (c) => {
  const formData = await c.req.formData()
  const file = formData.get('file') as File
 
  if (!file || file.size === 0) {
    return c.json({ error: 'No file provided' }, 400)
  }
 
  const taskId = crypto.randomUUID()
  const now = new Date().toISOString()
 
  // 1. 文件上传到 R2
  const fileKey = `uploads/${taskId}/${file.name}`
  await c.env.BUCKET.put(fileKey, file.stream(), {
    httpMetadata: { contentType: file.type },
  })
 
  // 2. D1 创建任务记录
  await c.env.DB.prepare(`
    INSERT INTO tasks (id, user_id, file_key, file_type, status, created_at, updated_at)
    VALUES (?, ?, ?, ?, 'pending', ?, ?)
  `).bind(taskId, 'anonymous', fileKey, file.type, now, now).run()
 
  // 3. Queue 写入消息
  await c.env.FILE_QUEUE.send({ taskId, userId: 'anonymous', fileKey, fileType: file.type })
 
  // 202 Accepted——请求已接受，尚未处理完
  return c.json({ taskId, status: 'pending' }, 202)
})

5. 消费者：Queue 处理

消费者在 queue 事件处理器中实现，Workers 有消息时自动调用：

// src/index.ts（续）
export default {
  async fetch(request: Request, env: Env['Bindings']): Promise<Response> {
    return app.fetch(request, env)
  },
 
  async queue(batch: MessageBatch<FileTaskMessage>, env: Env['Bindings']): Promise<void> {
    for (const message of batch.messages) {
      const { taskId, fileKey, fileType } = message.body
 
      try {
        // 标记为处理中
        await env.DB.prepare(
          "UPDATE tasks SET status = 'processing', updated_at = ? WHERE id = ?"
        ).bind(new Date().toISOString(), taskId).run()
 
        // 从 R2 取文件
        const object = await env.BUCKET.get(fileKey)
        if (!object) throw new Error(`File not found: ${fileKey}`)
 
        // 按文件类型分发处理
        let result: { resultKey: string }
        if (fileType === 'application/pdf') {
          result = await processPDF(object, taskId, env)
        } else if (fileType.startsWith('image/')) {
          result = await processImage(object, taskId, env)
        } else if (fileType.startsWith('audio/')) {
          result = await processAudio(object, taskId, env)
        } else {
          throw new Error(`Unsupported: ${fileType}`)
        }
 
        // 处理成功
        await env.DB.prepare(
          "UPDATE tasks SET status='completed', progress=100, result_key=?, updated_at=? WHERE id=?"
        ).bind(result.resultKey, new Date().toISOString(), taskId).run()
 
        message.ack()
 
      } catch (error) {
        const errMsg = error instanceof Error ? error.message : String(error)
        const retryCount = message.attempts - 1
 
        await env.DB.prepare(
          'UPDATE tasks SET retry_count=?, error_message=?, updated_at=? WHERE id=?'
        ).bind(retryCount, errMsg, new Date().toISOString(), taskId).run()
 
        if (isRetryable(error) && retryCount < 3) {
          // 不 ack，Queue 按指数退避重新投递
          message.retry({ delaySeconds: 30 * (retryCount + 1) })
        } else {
          // 不可重试或重试耗尽
          await env.DB.prepare(
            "UPDATE tasks SET status='failed', updated_at=? WHERE id=?"
          ).bind(new Date().toISOString(), taskId).run()
          message.ack()
        }
      }
    }
  }
} satisfies ExportedHandler<Env['Bindings']>
 
function isRetryable(error: unknown): boolean {
  if (!(error instanceof Error)) return true
  return ['timeout', 'rate limit', 'temporarily unavailable'].some(
    msg => error.message.toLowerCase().includes(msg)
  )
}

关键区别：message.ack() 表示处理成功，Queue 会移除这条消息。不 ack 或调用 message.retry() 则会重新投递。重试耗尽的消息自动进入 dead_letter_queue。

6. 进度上报

处理函数在执行过程中更新 D1 的 progress 字段：

// src/processors/pdf.ts
export async function processPDF(
  object: R2ObjectBody, taskId: string, env: { DB: D1Database; BUCKET: R2Bucket }
): Promise<{ resultKey: string }> {
  const buffer = await object.arrayBuffer()
  const totalSteps = 4
 
  const updateProgress = async (step: number) => {
    const progress = Math.round((step / totalSteps) * 100)
    await env.DB.prepare(
      'UPDATE tasks SET progress=?, updated_at=? WHERE id=?'
    ).bind(progress, new Date().toISOString(), taskId).run()
  }
 
  await updateProgress(1)
  const text = extractText(buffer)
 
  await updateProgress(2)
  const structure = parseStructure(text)
 
  await updateProgress(3)
  const summary = generateSummary(structure)
 
  await updateProgress(4)
  const resultKey = `results/${taskId}/output.json`
  await env.BUCKET.put(resultKey, JSON.stringify({ text, structure, summary }))
 
  return { resultKey }
}

大文件逐页处理时，每 N 页更新一次进度，避免频繁写 D1。

7. 查询进度与获取结果

// 查询任务状态
app.get('/api/tasks/:taskId', async (c) => {
  const task = await c.env.DB.prepare(
    'SELECT id, status, progress, result_key, error_message, created_at, updated_at FROM tasks WHERE id=?'
  ).bind(c.req.param('taskId')).first()
 
  if (!task) return c.json({ error: 'Task not found' }, 404)
 
  return c.json({
    taskId: task.id, status: task.status, progress: task.progress,
    resultKey: task.result_key, errorMessage: task.error_message,
  })
})
 
// 下载结果
app.get('/api/tasks/:taskId/result', async (c) => {
  const task = await c.env.DB.prepare(
    'SELECT status, result_key FROM tasks WHERE id=?'
  ).bind(c.req.param('taskId')).first()
 
  if (!task || task.status !== 'completed') {
    return c.json({ error: 'Task not completed' }, 409)
  }
 
  const result = await c.env.BUCKET.get(task.result_key!)
  if (!result) return c.json({ error: 'Result not found' }, 404)
 
  c.header('Content-Type', result.httpMetadata?.contentType || 'application/json')
  return c.body(result.body)
})

8. SSE 实时通知

轮询简单但有延迟。如果前端需要更实时的进度更新，可以用 Server-Sent Events。Hono 提供了 streamSSE：

import { streamSSE } from 'hono/streaming'
 
app.get('/api/tasks/:taskId/stream', async (c) => {
  return streamSSE(c, async (stream) => {
    let lastStatus = ''
 
    while (true) {
      const task = await c.env.DB.prepare(
        'SELECT status, progress, error_message FROM tasks WHERE id=?'
      ).bind(c.req.param('taskId')).first()
 
      if (!task) break
 
      const current = `${task.status}:${task.progress}`
      if (current !== lastStatus) {
        lastStatus = current
        await stream.writeSSE({
          event: 'progress',
          data: JSON.stringify({
            status: task.status, progress: task.progress,
          }),
        })
      }
 
      if (task.status === 'completed' || task.status === 'failed') break
      await stream.sleep(1000)
    }
  })
})

前端用 EventSource 接收：

const source = new EventSource(`/api/tasks/${taskId}/stream`)
 
source.addEventListener('progress', (event) => {
  const data = JSON.parse(event.data)
  updateProgressBar(data.progress)
 
  if (data.status === 'completed' || data.status === 'failed') {
    source.close()
  }
})

SSE 的局限：Workers 免费版每个 IP 最多 100 个并发连接，大批量场景轮询更稳妥。如果需要双向通信（比如取消任务），需要用 WebSocket。进度推送场景下 SSE 够用。

9. 错误重试与死信队列

重试策略的核心是区分「临时性故障」和「永久性错误」：

可重试：网络超时、R2 暂时不可用、外部 API 限流——等一会儿可能就好了
不可重试：文件格式不支持、文件损坏——重试多少次都没用

第 5 节的消费者代码已经展示了这个逻辑。isRetryable() 函数通过错误信息判断是否值得重试，可重试的调用 message.retry({ delaySeconds })，不可重试的直接标记 failed。

dead_letter_queue 接住重试耗尽的消息。写一个单独的消费者处理死信，记录日志、标记最终失败状态、必要时通知用户手动重试。

10. 完整流程

把各部分串起来，数据流如下：

用户提交文件 → POST /api/tasks
    ├── 文件 → R2
    ├── 任务记录 → D1（status=pending）
    └── 消息 → Queue
    ↓
返回 202 { taskId }
    ↓
Queue 消费者拉取消息
    ├── status → processing
    ├── R2 读取 → 解析/OCR/转码（每步更新 progress）
    ├── 结果 → R2
    └── status → completed
    ↓
前端：GET /api/tasks/:taskId（轮询）
  或：GET /api/tasks/:taskId/stream（SSE）
    ↓
完成后：GET /api/tasks/:taskId/result

前端调用：

// 提交
const formData = new FormData()
formData.append('file', file)
const { taskId } = await fetch('/api/tasks', { method: 'POST', body: formData }).then(r => r.json())
 
// 监听进度
const source = new EventSource(`/api/tasks/${taskId}/stream`)
source.addEventListener('progress', (e) => {
  const { status, progress } = JSON.parse(e.data)
  updateProgressBar(progress)
  if (status === 'completed' || status === 'failed') source.close()
})
 
// 完成后下载
const res = await fetch(`/api/tasks/${taskId}/result`)
const blob = await res.blob()

总结

HTTP 接口只做三件事：收文件存 R2、写任务记录到 D1、发消息到 Queue，然后返回 202
消费者在后台从 Queue 拉消息，按步骤处理文件并更新进度
前端通过轮询或 SSE 获取进度，完成后从 R2 下载结果
重试区分临时性故障和永久性错误，死信队列兜底重试耗尽的消息
消息体只放文件引用（fileKey），文件内容从 R2 读，不塞进 128KB 的消息限制里

如果处理逻辑需要 GPU 推理等 Workers 做不了的事，可以把消费者指向外部服务，Queue 只做消息中转。