17.13-Hono作为任务调度入口

要点

Hono 同时处理 HTTP 请求、Queue 消息、Cron 定时任务——三种入口统一管理
HTTP 接口负责接收请求、校验、入队
Queue 消费者负责后台处理、重试、状态更新
Cron 负责定时扫描、清理、汇总
所有入口共享同一套业务逻辑和环境绑定
这篇是整个第 17 章的收束，把前面讲的能力拼在一起

1. 三种入口的统一

Workers 有三种触发方式：

触发方式	入口方法	用途
HTTP 请求	`fetch()`	用户直接调用
Queue 消息	`queue()`	异步任务处理
Cron 定时	`scheduled()`	周期性任务

Hono 本身只处理 fetch。要接入 Queue 和 Cron，需要在默认导出里包装：

import { Hono } from 'hono'
import { taskConsumer } from './consumers/task-consumer'
import { cronJobs } from './cron/jobs'
 
const app = new Hono<{ Bindings: Env }>()
 
// HTTP 路由
app.post('/api/tasks', async (c) => { /* ... */ })
app.get('/api/tasks/:id', async (c) => { /* ... */ })
 
// 默认导出包装三种入口
export default {
  fetch: app.fetch,
 
  async queue(batch: MessageBatch, env: Env, ctx: ExecutionContext) {
    switch (batch.queue) {
      case 'tasks':
        await taskConsumer(batch as MessageBatch<TaskMessage>, env)
        break
      case 'notifications':
        await notificationConsumer(batch as MessageBatch<NotifyMessage>, env)
        break
    }
  },
 
  async scheduled(controller: ScheduledController, env: Env, ctx: ExecutionContext) {
    switch (controller.cron) {
      case '*/5 * * * *':
        ctx.waitUntil(cronJobs.cleanupStaleTasks(env))
        break
      case '0 * * * *':
        ctx.waitUntil(cronJobs.hourlyRollup(env))
        break
      case '30 3 * * *':
        ctx.waitUntil(cronJobs.dailyCleanup(env))
        break
    }
  },
}

2. 项目结构

src/
├── index.ts              # 入口：Hono app + 默认导出
├── types.ts              # 类型定义
├── routes/               # HTTP 路由
│   ├── tasks.ts          # 任务提交、查询
│   ├── workflows.ts      # 工作流接口
│   └── batch.ts          # 批量任务接口
├── consumers/            # Queue 消费者
│   ├── task-consumer.ts  # 单任务处理
│   ├── workflow-consumer.ts
│   └── batch-consumer.ts
├── cron/                 # Cron 定时任务
│   ├── jobs.ts           # 定时任务实现
│   └── cleanup.ts        # 清理任务
├── services/             # 业务逻辑
│   ├── file-parser.ts
│   ├── vectorizer.ts
│   └── llm-client.ts
└── lib/                  # 工具函数
    ├── queue.ts
    ├── progress.ts
    └── retry.ts

路由层只做参数校验和入队，业务逻辑放在 services/。消费者和 Cron 都调用 services/ 里的函数。

3. HTTP 路由：任务提交

// src/routes/tasks.ts
import { Hono } from 'hono'
import type { Env, TaskMessage } from '../types'
 
export const tasksApp = new Hono<{ Bindings: Env }>()
 
tasksApp.post('/tasks', async (c) => {
  const formData = await c.req.formData()
  const file = formData.get('file') as File
  const type = formData.get('type') as string
 
  if (!file) return c.json({ error: 'No file' }, 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, type, file_key, status, progress, created_at, updated_at)
    VALUES (?, ?, ?, ?, 'pending', 0, ?, ?)
  `).bind(taskId, c.get('user').id, type, fileKey, now, now).run()
 
  // 3. 入队
  await c.env.TASK_QUEUE.send({
    taskId,
    userId: c.get('user').id,
    fileKey,
    fileType: file.type,
    type,
  })
 
  // 4. 返回 202
  return c.json({ taskId, status: 'pending' }, 202)
})
 
tasksApp.get('/tasks/:taskId', async (c) => {
  const task = await c.env.DB.prepare(`
    SELECT id, type, status, stage, progress, error_message, created_at, updated_at
    FROM tasks WHERE id = ? AND user_id = ?
  `).bind(c.req.param('taskId'), c.get('user').id).first()
 
  if (!task) return c.json({ error: 'Not found' }, 404)
  return c.json(task)
})
 
tasksApp.get('/tasks/:taskId/progress', async (c) => {
  const taskId = c.req.param('taskId')
 
  // 先查 KV（实时进度）
  const kvData = await c.env.KV.get(`progress:${taskId}`, 'json')
  if (kvData) return c.json(kvData)
 
  // 再查 D1
  const task = await c.env.DB.prepare(`
    SELECT status, stage, progress FROM tasks WHERE id = ?
  `).bind(taskId).first()
 
  return c.json(task)
})

4. Queue 消费者：任务处理

// src/consumers/task-consumer.ts
import type { MessageBatch } from '@cloudflare/workers-types'
import type { Env, TaskMessage } from '../types'
import { parseDocument } from '../services/file-parser'
import { vectorize } from '../services/vectorizer'
import { updateProgress } from '../lib/progress'
 
export async function taskConsumer(
  batch: MessageBatch<TaskMessage>,
  env: Env
): Promise<void> {
  for (const msg of batch.messages) {
    const { taskId, fileKey, fileType, type } = msg.body
 
    try {
      // 标记处理中
      await env.DB.prepare(
        "UPDATE tasks SET status = 'processing', started_at = ?, updated_at = ? WHERE id = ?"
      ).bind(new Date().toISOString(), new Date().toISOString(), taskId).run()
 
      // 读文件
      const object = await env.BUCKET.get(fileKey)
      if (!object) throw new Error('File not found')
      const buffer = await object.arrayBuffer()
 
      // 根据类型分发
      if (type === 'vectorize') {
        await processVectorize(taskId, buffer, fileType, env)
      } else if (type === 'ocr') {
        await processOCR(taskId, buffer, env)
      } else if (type === 'transcribe') {
        await processTranscribe(taskId, buffer, env)
      } else {
        throw new Error(`Unknown type: ${type}`)
      }
 
      // 标记完成
      await env.DB.prepare(
        "UPDATE tasks SET status = 'completed', progress = 100, completed_at = ?, updated_at = ? WHERE id = ?"
      ).bind(new Date().toISOString(), new Date().toISOString(), taskId).run()
 
      msg.ack()
    } catch (err) {
      const errMsg = err instanceof Error ? err.message : String(err)
      console.error(`Task ${taskId} failed:`, errMsg)
 
      const retryCount = msg.attempts - 1
      if (retryCount < 3 && isRetryable(err)) {
        await env.DB.prepare(
          "UPDATE tasks SET status = 'retrying', retry_count = retry_count + 1, error_message = ?, updated_at = ? WHERE id = ?"
        ).bind(errMsg, new Date().toISOString(), taskId).run()
 
        msg.retry({ delaySeconds: 30 * (retryCount + 1) })
      } else {
        await env.DB.prepare(
          "UPDATE tasks SET status = 'failed', error_message = ?, updated_at = ? WHERE id = ?"
        ).bind(errMsg, new Date().toISOString(), taskId).run()
 
        msg.ack()
      }
    }
  }
}
 
async function processVectorize(taskId: string, buffer: ArrayBuffer, mimeType: string, env: Env) {
  await updateProgress(env, taskId, 0, 'parse')
  const text = await parseDocument(buffer, mimeType)
 
  await updateProgress(env, taskId, 30, 'chunk')
  const chunks = chunkText(text)
 
  await updateProgress(env, taskId, 50, 'embed')
  const vectors = await generateEmbeddings(chunks, env)
 
  await updateProgress(env, taskId, 80, 'index')
  await indexVectors(chunks, vectors, env)
 
  await updateProgress(env, taskId, 100, 'completed')
}

5. Cron 定时任务

// src/cron/jobs.ts
import type { Env } from '../types'
 
export const cronJobs = {
  // 每 5 分钟：清理超时任务
  async cleanupStaleTasks(env: Env): Promise<void> {
    const cutoff = new Date(Date.now() - 30 * 60 * 1000).toISOString()
 
    const stale = await env.DB.prepare(`
      SELECT id FROM tasks WHERE status = 'processing' AND started_at < ?
    `).bind(cutoff).all()
 
    for (const task of stale.results) {
      // 重新入队
      await env.TASK_QUEUE.send({ taskId: task.id })
      await env.DB.prepare(
        "UPDATE tasks SET status = 'pending', updated_at = ? WHERE id = ?"
      ).bind(new Date().toISOString(), task.id).run()
    }
  },
 
  // 每小时：汇总统计
  async hourlyRollup(env: Env): Promise<void> {
    const hourAgo = new Date(Date.now() - 60 * 60 * 1000).toISOString()
 
    const stats = await env.DB.prepare(`
      SELECT
        status,
        COUNT(*) as count,
        AVG(JULIANDAY(completed_at) - JULIANDAY(started_at)) * 86400 as avg_duration
      FROM tasks
      WHERE created_at > ?
      GROUP BY status
    `).bind(hourAgo).all()
 
    // 存入统计表
    for (const row of stats.results) {
      await env.DB.prepare(`
        INSERT INTO task_stats (hour, status, count, avg_duration, created_at)
        VALUES (?, ?, ?, ?, ?)
      `).bind(
        new Date().toISOString().slice(0, 13),
        row.status,
        row.count,
        row.avg_duration,
        new Date().toISOString(),
      ).run()
    }
  },
 
  // 每天凌晨 3:30：清理过期数据
  async dailyCleanup(env: Env): Promise<void> {
    const cutoff = new Date()
    cutoff.setDate(cutoff.getDate() - 30)
 
    // 删除 30 天前的已完成任务
    await env.DB.prepare(`
      DELETE FROM tasks WHERE status = 'completed' AND completed_at < ?
    `).bind(cutoff.toISOString()).run()
 
    // 清理死信
    await env.DB.prepare(`
      DELETE FROM dead_letters WHERE created_at < ?
    `).bind(cutoff.toISOString()).run()
  },
}

6. 配置汇总

// wrangler.jsonc
{
  "name": "task-worker",
  "main": "src/index.ts",
 
  "queues": {
    "producers": [
      { "queue": "tasks", "binding": "TASK_QUEUE" },
      { "queue": "notifications", "binding": "NOTIFY_QUEUE" }
    ],
    "consumers": [
      {
        "queue": "tasks",
        "max_batch_size": 5,
        "max_retries": 3,
        "dead_letter_queue": "tasks-dlq"
      },
      {
        "queue": "notifications",
        "max_batch_size": 10,
        "max_retries": 3
      },
      {
        "queue": "tasks-dlq",
        "max_batch_size": 10,
        "max_retries": 0
      }
    ]
  },
 
  "triggers": {
    "crons": [
      "*/5 * * * *",    // 每 5 分钟
      "0 * * * *",      // 每小时
      "30 3 * * *"      // 每天凌晨 3:30
    ]
  },
 
  "d1_databases": [
    { "binding": "DB", "database_name": "tasks-db", "database_id": "xxx" }
  ],
 
  "r2_buckets": [
    { "binding": "BUCKET", "bucket_name": "task-files" }
  ],
 
  "kv_namespaces": [
    { "binding": "KV", "namespace_id": "xxx" }
  ]
}

7. 完整示例：从提交到完成

用户提交一个文档向量化任务：

1. POST /api/tasks
   - 文件上传到 R2
   - D1 创建任务记录
   - Queue 发送消息
   - 返回 202 + taskId

2. Queue 消费者接收
   - 读 R2 文件
   - 解析 → 切 chunk → embedding → 写入向量库
   - 每步更新 KV 进度
   - 完成后更新 D1 状态

3. 前端轮询 GET /api/tasks/:taskId/progress
   - 先查 KV（实时）
   - 返回进度和阶段

4. Cron 每 5 分钟
   - 扫描超时任务
   - 重新入队

5. 任务完成
   - 前端查询 GET /api/tasks/:taskId
   - 返回结果或下载链接

8. 监控和可观测性

// 任务统计接口
app.get('/api/stats', async (c) => {
  const [total, byStatus, avgDuration] = await Promise.all([
    c.env.DB.prepare('SELECT COUNT(*) as count FROM tasks').first(),
    c.env.DB.prepare(`
      SELECT status, COUNT(*) as count FROM tasks GROUP BY status
    `).all(),
    c.env.DB.prepare(`
      SELECT AVG(JULIANDAY(completed_at) - JULIANDAY(started_at)) * 86400 as avg
      FROM tasks WHERE status = 'completed' AND completed_at > datetime('now', '-1 hour')
    `).first(),
  ])
 
  const statusMap: Record<string, number> = {}
  for (const row of byStatus.results) {
    statusMap[row.status] = row.count
  }
 
  return c.json({
    total: total?.count ?? 0,
    byStatus: statusMap,
    avgDurationSeconds: avgDuration?.avg ?? 0,
  })
})

总结

回顾这一节的要点：

Hono 同时处理 HTTP、Queue、Cron 三种入口——统一管理
HTTP 接口负责接收、校验、入队——返回 202
Queue 消费者负责后台处理——重试、状态更新、进度上报
Cron 负责定时扫描——清理超时任务、汇总统计、清理过期数据
所有入口共享 services/ 层的业务逻辑
配置集中在 wrangler.jsonc——队列、Cron、D1、R2、KV 绑定
监控接口提供任务统计——成功率、平均耗时、队列深度

第 17 章从「为什么需要异步」开始，讲了任务队列模型、选型、状态设计、重试、死信、进度查询、具体任务类型（向量化、批量生成）、工作流编排，最后以 Hono 作为统一入口收束。这些能力组合起来，可以构建完整的异步任务系统。