17.02-任务队列基本模型

要点

  • 消息队列有三种基本模型：点对点、发布-订阅、请求-响应
  • 点对点模型最常用——生产者发消息，一个消费者取走处理，适合任务分发
  • 发布-订阅模型——一条消息被多个消费者接收，适合事件广播
  • 请求-响应模型——生产者期待消费者返回结果，适合 RPC 场景
  • Cloudflare Queue 原生支持点对点和发布-订阅，请求-响应需要自己实现
  • 选对模型比选对工具更重要——模型错了，后面全错

1. 为什么先讲模型

很多教程直接讲「怎么配置 Queue」「怎么写消费者」，但不讲背后的模型。结果是碰到实际问题时不知道怎么选：

  • 多个消费者抢任务，用哪个模型？
  • 一条消息需要被多个服务处理，怎么办？
  • 生产者想知道任务执行结果，怎么实现？

这些问题都有对应的模型。先理解模型，再看工具怎么用。

2. 点对点模型（Point-to-Point）

基本结构

生产者 → 队列 → 消费者（只有一个取走消息）

生产者往队列里发消息，多个消费者竞争消费。每条消息只被一个消费者取走——先到先得。

这是任务队列最常见的模型。适用场景：

  • 用户上传文件，后台解析——只需要一个 Worker 处理
  • 订单创建后发邮件——只需要一个消费者发
  • 批量 embedding——每条消息独立处理

Cloudflare Queue 实现

Cloudflare Queue 默认就是点对点模型：

// wrangler.jsonc
{
  "queues": {
    "producers": [{
      "queue": "task-queue",
      "binding": "TASK_QUEUE"
    }],
    "consumers": [{
      "queue": "task-queue",
      "max_batch_size": 10
    }]
  }
}

如果有多个 Worker 实例消费同一个队列，消息只会被其中一个实例处理。Cloudflare 保证每条消息至少被投递一次（at-least-once），但不保证只投递一次——消费者需要处理重复消息（幂等性）。

代码示例

// 生产者
await c.env.TASK_QUEUE.send({
  taskId: '123',
  userId: '456',
  action: 'parse_pdf',
})
 
// 消费者
export async function consumer(batch: MessageBatch<TaskMessage>, env: Env) {
  for (const msg of batch.messages) {
    await processTask(msg.body)
    msg.ack()  // 确认处理完成
  }
}

msg.ack() 告诉队列「这条消息处理完了，可以删除」。如果消费者崩溃或没调 ack()，消息会被重新投递。

3. 发布-订阅模型（Publish-Subscribe）

基本结构

         ┌→ 消费者 A（收到消息）
生产者 → 队列 ├→ 消费者 B（收到消息）
         └→ 消费者 C（收到消息）

一条消息被多个消费者接收。每个消费者都会处理这条消息——不是竞争，是广播。

适用场景：

  • 用户上传文档后，多个服务需要响应：搜索索引服务建索引、推荐服务更新向量、通知服务发站内信
  • 订单创建后：库存服务扣库存、积分服务加积分、短信服务发通知
  • 数据变更事件：缓存服务更新缓存、日志服务记录、监控服务报警

Cloudflare Queue 实现

Cloudflare Queue 支持一个队列绑定多个消费者。但更常见的做法是用多个队列——生产者往多个队列各发一条消息：

// 生产者：往多个队列发消息
await env.INDEX_QUEUE.send({ docId: '123', action: 'index' })
await env.RECOMMEND_QUEUE.send({ docId: '123', action: 'update_embedding' })
await env.NOTIFY_QUEUE.send({ docId: '123', action: 'send_notification' })

每个队列有独立的消费者，互不影响。一个消费者失败不影响其他消费者。

代码示例

// 消费者 A：搜索索引
export async function indexConsumer(batch: MessageBatch<DocEvent>, env: Env) {
  for (const msg of batch.messages) {
    await env.SEARCH_INDEX.upsert({
      id: msg.body.docId,
      text: msg.body.text,
    })
    msg.ack()
  }
}
 
// 消费者 B：推荐系统
export async function recommendConsumer(batch: MessageBatch<DocEvent>, env: Env) {
  for (const msg of batch.messages) {
    const embedding = await env.AI.run('@cf/baai/bge-base-en-v1.5', {
      text: msg.body.text,
    })
    await env.VECTOR_DB.upsert({
      id: msg.body.docId,
      values: embedding.data[0],
    })
    msg.ack()
  }
}

两个消费者独立运行，失败互不影响。

4. 请求-响应模型（Request-Reply）

基本结构

生产者 → 请求队列 → 消费者
生产者 ← 响应队列 ← 消费者

生产者发消息后等待消费者返回结果。这不是传统 HTTP 的请求-响应，而是通过两个队列异步通信。

适用场景：

  • 远程过程调用（RPC）——调用另一个服务的函数并等待结果
  • 复杂计算——把计算任务发给专门的 Worker，等待结果
  • 跨服务协作——一个流程涉及多个服务，每步都需要上一步的结果

自己实现

Cloudflare Queue 没有内置请求-响应支持，需要自己用两个队列实现：

// 生产者
async function sendAndWaitReply(taskId: string, env: Env): Promise<string> {
  // 1. 生成关联 ID
  const correlationId = crypto.randomUUID()
 
  // 2. 发请求消息
  await env.REQUEST_QUEUE.send({
    taskId,
    correlationId,
    payload: { action: 'calculate', data: '...' },
  })
 
  // 3. 等待响应（通过轮询 D1 或 KV）
  while (true) {
    const result = await env.DB.prepare(`
      SELECT result FROM task_results WHERE correlation_id = ?
    `).bind(correlationId).first()
 
    if (result) return result.result as string
 
    await new Promise((r) => setTimeout(r, 1000))  // 等 1 秒再查
  }
}
 
// 消费者
export async function requestConsumer(batch: MessageBatch<RequestMessage>, env: Env) {
  for (const msg of batch.messages) {
    const { correlationId, payload } = msg.body
 
    // 处理请求
    const result = await doCalculation(payload)
 
    // 写结果到 D1
    await env.DB.prepare(`
      INSERT INTO task_results (correlation_id, result, created_at)
      VALUES (?, ?, ?)
    `).bind(correlationId, result, new Date().toISOString()).run()
 
    msg.ack()
  }
}

生产者发请求后轮询 D1 等结果。消费者处理完把结果写入 D1。两边通过 correlationId 关联。

这种方式的缺点：生产者需要一直轮询，延迟高。更好的方式是结合 SSE 或 WebSocket 推送结果，但复杂度也更高。

5. 模型对比

模型消息消费适用场景复杂度
点对点一个消费者任务分发、后台处理低
发布-订阅多个消费者事件广播、多服务联动中
请求-响应一个消费者 + 返回结果RPC、复杂计算高

选择建议：

  • 大多数 AI 任务用点对点——上传文件、生成 embedding、调 LLM，都是单消费者处理
  • 需要多服务联动用发布-订阅——用户上传后同时更新搜索索引、推荐系统、通知
  • 需要返回结果用请求-响应——但要接受额外的延迟和复杂度

6. 消息语义

队列的消息交付有三种语义：

At-most-once（最多一次）

消息可能丢失，但不会重复。消费者拿到消息后立即确认，即使处理失败也不重试。

适用场景：日志收集、实时监控——丢几条无所谓，重复反而有问题。

At-least-once（至少一次）

消息不会丢失，但可能重复。消费者处理完必须显式确认（ack()），否则消息会被重新投递。

Cloudflare Queue 默认就是 at-least-once。消费者需要保证幂等性——同一条消息处理多次，结果和处理一次一样。

// 幂等处理
const processed = await env.KV.get(`processed:${msg.body.taskId}`)
if (processed) {
  msg.ack()  // 已经处理过，直接确认
  continue
}
 
// 实际处理
await processTask(msg.body)
 
// 标记为已处理（24 小时后过期）
await env.KV.put(`processed:${msg.body.taskId}`, '1', { expirationTtl: 86400 })
msg.ack()

Exactly-once（恰好一次）

消息不丢失、不重复。实现成本最高——需要分布式事务或事务性发件箱模式。

Cloudflare Queue 不支持 exactly-once。如果业务严格要求，需要在应用层实现幂等性（效果等同于 at-least-once）。

7. 消息顺序

单分区有序

同一个队列的消息，按发送顺序投递。但多个消费者并发处理时，实际执行顺序不确定。

分区键保序

Cloudflare Queue 支持消息分组——相同分组键的消息按顺序投递：

await env.TASK_QUEUE.send(
  { taskId: '1', userId: 'A' },
  { groupId: 'user-A' }  // 同一用户的所有消息按顺序处理
)
await env.TASK_QUEUE.send(
  { taskId: '2', userId: 'A' },
  { groupId: 'user-A' }
)

groupId 相同的消息保证按发送顺序投递。适用场景：

  • 同一用户的操作按顺序执行（先创建再修改）
  • 同一文档的更新按顺序处理（避免乱序覆盖）

不保证全局顺序

不同 groupId 的消息并发处理，没有全局顺序保证。如果需要全局有序，只能用单分区（性能差）。

8. 在 Hono 中使用

把队列集成到 Hono 路由：

// src/index.ts
import { Hono } from 'hono'
import { taskConsumer } from './consumers/task-consumer'
import { notifyConsumer } from './consumers/notify-consumer'
 
type Bindings = {
  TASK_QUEUE: Queue<TaskMessage>
  NOTIFY_QUEUE: Queue<NotifyMessage>
}
 
const app = new Hono<{ Bindings: Bindings }>()
 
// HTTP 路由
app.post('/api/tasks', async (c) => {
  const { action, payload } = await c.req.json()
 
  await c.env.TASK_QUEUE.send({
    taskId: crypto.randomUUID(),
    action,
    payload,
  })
 
  return c.json({ status: 'accepted' }, 202)
})
 
// Worker 入口
export default {
  fetch: app.fetch,
 
  async queue(batch: MessageBatch, env: Bindings) {
    // 根据队列名分发到不同消费者
    switch (batch.queue) {
      case 'task-queue':
        await taskConsumer(batch as MessageBatch<TaskMessage>, env)
        break
      case 'notify-queue':
        await notifyConsumer(batch as MessageBatch<NotifyMessage>, env)
        break
    }
  },
}

一个 Worker 可以同时是多个队列的消费者——通过 batch.queue 判断消息来自哪个队列。

总结

回顾这一节的要点：

  • 消息队列有三种基本模型：点对点、发布-订阅、请求-响应
  • 点对点模型最常用——生产者发消息，一个消费者处理，适合后台任务
  • 发布-订阅模型用于事件广播——一条消息被多个消费者处理
  • 请求-响应模型用于 RPC——生产者等待消费者返回结果，需要自己实现
  • Cloudflare Queue 原生支持点对点，发布-订阅用多队列实现，请求-响应需要轮询或推送
  • 消息交付语义：at-most-once、at-least-once、exactly-once——Cloudflare Queue 是 at-least-once，消费者需要幂等
  • 消息顺序：同 groupId 有序，全局无序
  • Hono 集成：HTTP 路由生产消息，queue() handler 消费消息

下一篇讲消息队列选型——Cloudflare Queue、Redis BullMQ、RabbitMQ，什么场景用什么工具。