17.03-消息队列选型

要点

Workers 环境首选 Cloudflare Queue——原生集成，零运维，按消息计费
需要 Redis 生态（BullMQ、延迟任务、优先级队列）时用 Upstash Redis——REST API 接入
传统消息队列（RabbitMQ、Kafka）不适合 Workers——需要自建服务或外部托管
选型看四个维度：集成难度、功能匹配、成本、运维负担
没有银弹——小项目用 Cloudflare Queue 足够，复杂场景才需要更重的方案

1. 先看 Workers 环境有什么

Workers 运行在 Cloudflare 边缘网络上，没有自己的服务器，不能随便装软件。能用的队列方案：

方案	集成方式	适用场景
Cloudflare Queue	原生绑定	大多数 Workers 项目
Upstash Redis + BullMQ	REST API	需要 BullMQ 高级功能
自建 RabbitMQ/Kafka	HTTP 调用	已有基础设施的团队
第三方托管（Aiven、Confluent）	HTTP 调用	需要 Kafka 协议

大多数项目用 Cloudflare Queue 就够了。需要 BullMQ 的延迟任务、优先级队列等高级功能时，再考虑 Upstash Redis。传统消息队列在 Workers 环境接入成本高，除非团队已有基础设施，否则不建议引入。

2. Cloudflare Queue

优势

原生集成——在 wrangler.jsonc 里配置 binding，Worker 里直接用 env.QUEUE.send()
零运维——Cloudflare 托管，不需要管服务器、集群、扩容
按消息计费——每百万条消息 $0.40（免费计划 100 万条/月），不用为空闲资源付费
和 D1、R2、KV 无缝配合——都在 Cloudflare 生态里，延迟最低

限制

功能简单——没有延迟任务、优先级队列、消息分组（后来加了 groupId）、死信队列（后来加了 DLQ）
消息大小限制——单条消息最大 128 KB
吞吐量限制——每个队列每秒最多 5000 条消息
消费者并发——每批最多 10 个消息，最多 3 个并发批次

适用场景

后台任务分发（文件处理、AI 生成）
事件驱动架构（用户注册后发欢迎邮件）
批量处理（文档向量化、数据同步）

代码速览

// wrangler.jsonc
{
  "queues": {
    "producers": [{ "queue": "tasks", "binding": "TASK_QUEUE" }],
    "consumers": [{
      "queue": "tasks",
      "max_batch_size": 5,
      "max_retries": 3,
      "dead_letter_queue": "tasks-dlq"
    }]
  }
}

// 生产者
await env.TASK_QUEUE.send({ taskId: '123', action: 'process' })
 
// 消费者
export default {
  async queue(batch: MessageBatch<TaskMessage>, env: Env) {
    for (const msg of batch.messages) {
      await process(msg.body)
      msg.ack()
    }
  }
}

3. Upstash Redis + BullMQ

为什么需要 BullMQ

Cloudflare Queue 功能简单，有些场景不够用：

延迟任务——「30 分钟后检查订单支付状态」
优先级队列——VIP 用户的任务优先处理
任务依赖——任务 B 等任务 A 完成后再开始
限流——每秒最多处理 10 个任务
重试策略——更细粒度的重试控制

这些功能 BullMQ 都有。BullMQ 是基于 Redis 的任务队列库，功能丰富。

Upstash Redis

Workers 环境没有原生 Redis，但 Upstash 提供 REST API 访问 Redis——不需要维护 Redis 服务器。

import { Client } from '@upstash/redis'
 
const redis = new Client({
  url: env.UPSTASH_REDIS_REST_URL,
  token: env.UPSTASH_REDIS_REST_TOKEN,
})
 
// 基本操作
await redis.set('key', 'value')
const value = await redis.get('key')

Upstash 按请求计费——每次 REST API 调用算一次请求。免费计划每天 10000 次请求。

BullMQ 接入

BullMQ 原生需要 Node.js 的 Redis 客户端（ioredis）。Workers 环境没有 ioredis，需要 BullMQ 的轻量版本或者自己封装。

更实际的做法：用 Upstash 的 Redis 作为存储，自己实现简化的任务队列逻辑。

// 简化的任务队列实现
class SimpleQueue {
  constructor(private redis: Client, private queueName: string) {}
 
  // 添加任务
  async enqueue(task: { id: string; payload: any; delay?: number }): Promise<void> {
    const key = `${this.queueName}:task:${task.id}`
    await redis.hset(key, {
      id: task.id,
      payload: JSON.stringify(task.payload),
      status: 'pending',
      createdAt: Date.now(),
    })
 
    if (task.delay) {
      // 延迟任务：加入延迟集合
      await redis.zadd(`${this.queueName}:delayed`, Date.now() + task.delay, task.id)
    } else {
      // 立即任务：加入列表
      await redis.rpush(`${this.queueName}:pending`, task.id)
    }
  }
 
  // 取出任务
  async dequeue(): Promise<any | null> {
    // 先检查延迟任务是否到期
    const now = Date.now()
    const ready = await redis.zrangebyscore(
      `${this.queueName}:delayed`, 0, now
    )
    for (const id of ready) {
      await redis.zrem(`${this.queueName}:delayed`, id)
      await redis.rpush(`${this.queueName}:pending`, id)
    }
 
    // 从列表取出一个
    const taskId = await redis.lpop(`${this.queueName}:pending`)
    if (!taskId) return null
 
    const key = `${this.queueName}:task:${taskId}`
    const data = await redis.hgetall(key)
    return data ? { id: data.id, payload: JSON.parse(data.payload) } : null
  }
 
  // 完成任务
  async complete(taskId: string): Promise<void> {
    await redis.hset(`${this.queueName}:task:${taskId}`, 'status', 'completed')
  }
}

这是一个极简实现。生产环境需要考虑：分布式锁、超时处理、重试逻辑、死信队列。

成本对比

方案	计费方式	100 万任务成本
Cloudflare Queue	$0.40 / 百万消息	$0.40
Upstash Redis	$10 / 月（1000 万请求）	~$10

Upstash 贵 25 倍，但功能更强。如果只需要简单的任务分发，Cloudflare Queue 性价比更高。

4. 传统消息队列

RabbitMQ

RabbitMQ 功能丰富——路由、优先级、延迟消息、事务。但 Workers 接入需要：

自建或托管 RabbitMQ 服务器
通过 HTTP API 或 AMQP 协议调用
自己处理认证、重连、消息确认

Cloudflare 没有原生 RabbitMQ 支持。需要通过外部 HTTP API 调用：

// 通过 RabbitMQ HTTP API 发消息
await fetch('https://rabbitmq.example.com/api/exchanges/%2F/tasks/publish', {
  method: 'POST',
  headers: {
    'Authorization': 'Basic ' + btoa('guest:guest'),
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    payload: JSON.stringify({ taskId: '123' }),
    payload_encoding: 'string',
  }),
})

消费消息也需要轮询 HTTP API，或者自建一个 Consumer Worker 常驻运行（Workers 不适合）。

不推荐在纯 Workers 项目里用 RabbitMQ——接入成本高、运维复杂。如果团队已有 RabbitMQ 基础设施，可以考虑。

Kafka

Kafka 适合大数据场景——高吞吐、消息回溯、多消费者组。但 Workers 接入更难：

Kafka 协议复杂，Workers 没有原生客户端
需要 Kafka 集群（Confluent Cloud、Aiven 等托管服务）
消费者需要长连接，Workers 不适合

不推荐在 Workers 项目里用 Kafka——除非团队已有 Kafka 基础设施且需要其特有能力。

5. 选型决策树

你的项目跑在 Workers 上吗？
├─ 否 → 按传统方式选型（RabbitMQ / Kafka / Redis）
└─ 是 → 需要延迟任务、优先级、任务依赖吗？
    ├─ 否 → Cloudflare Queue（首选）
    └─ 是 → 团队已有 Redis 基础设施吗？
        ├─ 是 → BullMQ + 已有 Redis
        └─ 否 → 能用 Upstash Redis 接受成本吗？
            ├─ 是 → Upstash Redis + 自实现队列
            └─ 否 → 回到 Cloudflare Queue，接受限制

6. 混合方案

实际项目里可以混合使用多种队列：

// 简单任务用 Cloudflare Queue
await env.TASK_QUEUE.send({ taskId: '123', action: 'parse' })
 
// 延迟任务用 Upstash Redis
await redis.zadd('delayed-tasks', Date.now() + 30 * 60 * 1000, JSON.stringify({
  taskId: '456',
  action: 'check_payment',
}))
 
// Cron 扫描延迟任务，推入 Cloudflare Queue
export async function scheduledCron(env: Env) {
  const now = Date.now()
  const ready = await env.REDIS.zrangebyscore('delayed-tasks', 0, now)
 
  for (const taskJson of ready) {
    await env.REDIS.zrem('delayed-tasks', taskJson)
    const task = JSON.parse(taskJson)
    await env.TASK_QUEUE.send(task)
  }
}

Cron 定时扫描 Upstash 的延迟任务，到期后推入 Cloudflare Queue 处理。这样既用了 Cloudflare Queue 的低成本，又有了延迟任务的能力。

7. 迁移考虑

如果项目从小做起，先用 Cloudflare Queue，后来需要更复杂的功能，迁移到 Upstash + 自实现队列的成本：

消息格式不变——都是 JSON
生产者接口不变——queue.send() 换成自己的 enqueue()
消费者逻辑不变——处理消息的代码不需要改
需要补充：延迟任务、重试逻辑、死信队列

迁移成本可控。关键是消费者逻辑（处理消息的代码）和队列基础设施解耦——消费者不直接调用队列 API，而是接收消息参数。

总结

回顾这一节的要点：

Workers 环境首选 Cloudflare Queue——原生集成，零运维，按消息计费
需要高级功能（延迟任务、优先级、任务依赖）时用 Upstash Redis + 自实现队列
传统消息队列（RabbitMQ、Kafka）不适合纯 Workers 项目——接入成本高
选型看四个维度：集成难度、功能匹配、成本、运维负担
可以混合方案——Cron + Cloudflare Queue + Upstash Redis 各司其职
消费者逻辑和队列基础设施解耦，方便迁移

下一篇讲 Redis Queue 实践——Upstash Redis 接入、自实现队列、延迟任务、优先级。