17.04-Redis Queue实践

要点

Workers 环境没有原生 Redis，通过 Upstash Redis 的 REST API 接入
Upstash 按请求计费，免费计划每天 10000 次请求
基于 Redis 可以实现延迟任务、优先级队列、任务依赖、限流等高级功能
延迟任务用 Sorted Set——score 是执行时间，Cron 扫描到期任务推入处理队列
优先级队列用多个列表——每个优先级一个列表，按优先级顺序取出
任务依赖用 Hash 记录状态——前置任务完成后触发后续任务

1. 为什么需要 Redis Queue

Cloudflare Queue 功能简单但够用。但有些场景需要更丰富的能力：

延迟任务——「30 分钟后检查订单支付状态」「用户注册 24 小时后发提醒」
优先级队列——VIP 用户的任务优先处理
任务依赖——任务 B 等任务 A 完成后再开始
限流——每秒最多处理 10 个任务，避免打爆外部 API
定时重试——失败后 5 分钟重试，而不是立即重试

这些功能 Cloudflare Queue 不支持或支持有限。Redis 生态的 BullMQ 全都有，但 Workers 跑不了 BullMQ。替代方案：Upstash Redis + 自己实现队列逻辑。

2. Upstash Redis 接入

创建服务

在 Upstash 控制台创建 Redis 实例，拿到 REST URL 和 Token。

类型声明

// src/types.ts
export type Env = {
  UPSTASH_REDIS_REST_URL: string
  UPSTASH_REDIS_REST_TOKEN: string
}

客户端封装

// src/lib/redis.ts
export class RedisClient {
  constructor(
    private url: string,
    private token: string
  ) {}
 
  async command<T>(cmd: string[], opts?: { pipeline?: boolean }): Promise<T> {
    const response = await fetch(this.url, {
      method: 'POST',
      headers: {
        'Authorization': `Bearer ${this.token}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify(cmd),
    })
 
    if (!response.ok) {
      throw new Error(`Redis error: ${response.status} ${await response.text()}`)
    }
 
    return response.json() as Promise<T>
  }
 
  // 常用命令封装
  async set(key: string, value: string, opts?: { ex?: number }): Promise<void> {
    const cmd = ['SET', key, value]
    if (opts?.ex) cmd.push('EX', String(opts.ex))
    await this.command(cmd)
  }
 
  async get(key: string): Promise<string | null> {
    return this.command<string | null>(['GET', key])
  }
 
  async del(key: string): Promise<void> {
    await this.command(['DEL', key])
  }
 
  async rpush(key: string, ...values: string[]): Promise<void> {
    await this.command(['RPUSH', key, ...values])
  }
 
  async lpop(key: string): Promise<string | null> {
    return this.command<string | null>(['LPOP', key])
  }
 
  async zadd(key: string, score: number, member: string): Promise<void> {
    await this.command(['ZADD', key, String(score), member])
  }
 
  async zrangebyscore(key: string, min: number, max: number): Promise<string[]> {
    return this.command<string[]>(['ZRANGEBYSCORE', key, String(min), String(max)])
  }
 
  async zrem(key: string, ...members: string[]): Promise<void> {
    await this.command(['ZREM', key, ...members])
  }
 
  async hset(key: string, field: string, value: string): Promise<void> {
    await this.command(['HSET', key, field, value])
  }
 
  async hget(key: string, field: string): Promise<string | null> {
    return this.command<string | null>(['HGET', key, field])
  }
 
  async hgetall(key: string): Promise<Record<string, string>> {
    return this.command<Record<string, string>>(['HGETALL', key])
  }
}

每次命令都是一次 HTTP 请求。如果一次操作需要多个命令，用 pipeline 合并成一次请求（Upstash 支持）。

3. 基础队列实现

用 Redis List 实现 FIFO 队列：

// src/lib/simple-queue.ts
import { RedisClient } from './redis'
 
export interface Task {
  id: string
  payload: unknown
  createdAt: number
  attempts: number
}
 
export class SimpleQueue {
  constructor(
    private redis: RedisClient,
    private queueName: string
  ) {}
 
  // 添加任务
  async enqueue(payload: unknown, opts?: { delay?: number }): Promise<string> {
    const id = crypto.randomUUID()
    const task: Task = {
      id,
      payload,
      createdAt: Date.now(),
      attempts: 0,
    }
 
    const key = `${this.queueName}:task:${id}`
    await this.redis.set(key, JSON.stringify(task))
 
    if (opts?.delay) {
      // 延迟任务：加入延迟集合
      await this.redis.zadd(
        `${this.queueName}:delayed`,
        Date.now() + opts.delay * 1000,
        id
      )
    } else {
      // 立即任务：加入待处理列表
      await this.redis.rpush(`${this.queueName}:pending`, id)
    }
 
    return id
  }
 
  // 取出任务
  async dequeue(): Promise<Task | null> {
    // 1. 先把到期的延迟任务移入待处理列表
    await this.promoteDelayedTasks()
 
    // 2. 从列表取出一个
    const taskId = await this.redis.lpop(`${this.queueName}:pending`)
    if (!taskId) return null
 
    // 3. 读取任务详情
    const key = `${this.queueName}:task:${taskId}`
    const data = await this.redis.get(key)
    if (!data) return null
 
    return JSON.parse(data) as Task
  }
 
  // 完成任务
  async complete(taskId: string): Promise<void> {
    await this.redis.del(`${this.queueName}:task:${taskId}`)
  }
 
  // 失败任务（重新入队）
  async fail(taskId: string, delay: number): Promise<void> {
    const key = `${this.queueName}:task:${taskId}`
    const data = await this.redis.get(key)
    if (!data) return
 
    const task = JSON.parse(data) as Task
    task.attempts += 1
 
    await this.redis.set(key, JSON.stringify(task))
 
    if (delay > 0) {
      await this.redis.zadd(
        `${this.queueName}:delayed`,
        Date.now() + delay * 1000,
        taskId
      )
    } else {
      await this.redis.rpush(`${this.queueName}:pending`, taskId)
    }
  }
 
  // 把到期的延迟任务移入待处理列表
  private async promoteDelayedTasks(): Promise<void> {
    const now = Date.now()
    const ready = await this.redis.zrangebyscore(
      `${this.queueName}:delayed`,
      0,
      now
    )
 
    for (const id of ready) {
      await this.redis.zrem(`${this.queueName}:delayed`, id)
      await this.redis.rpush(`${this.queueName}:pending`, id)
    }
  }
}

使用：

const redis = new RedisClient(env.UPSTASH_REDIS_REST_URL, env.UPSTASH_REDIS_REST_TOKEN)
const queue = new SimpleQueue(redis, 'tasks')
 
// 添加任务
const taskId = await queue.enqueue({ action: 'parse', fileId: '123' })
 
// 添加延迟任务（30 分钟后）
await queue.enqueue({ action: 'check_payment', orderId: '456' }, { delay: 30 * 60 })
 
// 取出任务（Cron 或消费者调用）
const task = await queue.dequeue()
if (task) {
  try {
    await processTask(task)
    await queue.complete(task.id)
  } catch (err) {
    await queue.fail(task.id, 60)  // 60 秒后重试
  }
}

4. 延迟任务

延迟任务的核心：用 Sorted Set，score 是执行时间戳。Cron 定时扫描到期的任务，推入待处理列表。

// Cron 定时触发（每 10 秒）
export async function scheduled(
  controller: ScheduledController,
  env: Env
): Promise<void> {
  const redis = new RedisClient(env.UPSTASH_REDIS_REST_URL, env.UPSTASH_REDIS_REST_TOKEN)
  const queue = new SimpleQueue(redis, 'tasks')
 
  // 扫描延迟集合，把到期任务移入待处理
  const now = Date.now()
  const ready = await redis.zrangebyscore('tasks:delayed', 0, now)
 
  for (const taskId of ready) {
    await redis.zrem('tasks:delayed', taskId)
    await redis.rpush('tasks:pending', taskId)
  }
 
  // 处理待处理任务
  while (true) {
    const task = await queue.dequeue()
    if (!task) break
 
    try {
      await processTask(task, env)
      await queue.complete(task.id)
    } catch (err) {
      if (task.attempts < 3) {
        await queue.fail(task.id, Math.pow(2, task.attempts) * 60)  // 指数退避
      } else {
        // 超过重试次数，进入死信队列
        await redis.rpush('tasks:dead-letter', task.id)
      }
    }
  }
}

Cron 每 10 秒扫描一次延迟任务。延迟精度取决于扫描频率。

5. 优先级队列

多个优先级——高优先级任务先处理。实现：每个优先级一个列表。

// src/lib/priority-queue.ts
export class PriorityQueue {
  private priorities = ['high', 'medium', 'low']
 
  constructor(
    private redis: RedisClient,
    private queueName: string
  ) {}
 
  async enqueue(payload: unknown, priority: 'high' | 'medium' | 'low' = 'medium'): Promise<string> {
    const id = crypto.randomUUID()
    const task = { id, payload, createdAt: Date.now(), attempts: 0 }
 
    await this.redis.set(`${this.queueName}:task:${id}`, JSON.stringify(task))
    await this.redis.rpush(`${this.queueName}:${priority}`, id)
 
    return id
  }
 
  async dequeue(): Promise<Task | null> {
    // 按优先级顺序检查列表
    for (const priority of this.priorities) {
      const taskId = await this.redis.lpop(`${this.queueName}:${priority}`)
      if (taskId) {
        const data = await this.redis.get(`${this.queueName}:task:${taskId}`)
        if (data) return JSON.parse(data) as Task
      }
    }
    return null
  }
}
 
// 使用
const pq = new PriorityQueue(redis, 'tasks')
await pq.enqueue({ action: 'parse', fileId: '123' }, 'high')    // VIP 用户
await pq.enqueue({ action: 'parse', fileId: '456' }, 'low')     // 普通用户
 
// 取出时 high 优先级先出
const task = await pq.dequeue()  // 先拿到 fileId: 123

6. 任务依赖

任务 B 需要等任务 A 完成后再开始。实现：用 Hash 记录任务状态，任务完成时检查是否有依赖它的后续任务。

// src/lib/dependency-queue.ts
export class DependencyQueue {
  constructor(private redis: RedisClient, private queueName: string) {}
 
  // 添加任务，可以指定前置任务
  async enqueue(
    payload: unknown,
    opts?: { dependsOn?: string[] }
  ): Promise<string> {
    const id = crypto.randomUUID()
    const task = { id, payload, createdAt: Date.now(), attempts: 0 }
 
    await this.redis.set(`${this.queueName}:task:${id}`, JSON.stringify(task))
 
    if (opts?.dependsOn && opts.dependsOn.length > 0) {
      // 记录依赖关系
      for (const depId of opts.dependsOn) {
        await this.redis.hset(`${this.queueName}:deps:${id}`, depId, 'pending')
      }
    } else {
      // 无依赖，直接入队
      await this.redis.rpush(`${this.queueName}:pending`, id)
    }
 
    return id
  }
 
  // 完成任务，触发后续任务
  async complete(taskId: string): Promise<void> {
    // 1. 检查哪些任务依赖这个任务
    const allDeps = await this.redis.command<string[]>(
      ['KEYS', `${this.queueName}:deps:*`]
    )
 
    for (const depKey of allDeps) {
      const dependantId = depKey.split(':').pop()!
      const status = await this.redis.hget(depKey, taskId)
 
      if (status === 'pending') {
        // 标记这个依赖已完成
        await this.redis.hset(depKey, taskId, 'completed')
 
        // 检查是否所有依赖都完成了
        const allDepsStatus = await this.redis.hgetall(depKey)
        const allCompleted = Object.values(allDepsStatus).every(s => s === 'completed')
 
        if (allCompleted) {
          // 所有依赖完成，入队
          await this.redis.rpush(`${this.queueName}:pending`, dependantId)
          await this.redis.del(depKey)
        }
      }
    }
 
    // 2. 删除任务数据
    await this.redis.del(`${this.queueName}:task:${taskId}`)
  }
}
 
// 使用
const dq = new DependencyQueue(redis, 'tasks')
 
// 任务 A
const taskA = await dq.enqueue({ action: 'download', url: '...' })
 
// 任务 B 依赖 A
const taskB = await dq.enqueue({ action: 'process', fileId: '...' }, {
  dependsOn: [taskA],
})
 
// 任务 C 依赖 A 和 B
const taskC = await dq.enqueue({ action: 'notify' }, {
  dependsOn: [taskA, taskB],
})
 
// 完成任务 A 后，B 入队（如果 B 是唯一依赖）
// 完成 A 和 B 后，C 入队

注意：KEYS 命令在生产环境要谨慎使用——会扫描所有 key。大规模场景用 Hash 或 Sorted Set 替代。

7. 限流

限制每秒处理的任务数，避免打爆外部 API。

// src/lib/rate-limited-queue.ts
export class RateLimitedQueue {
  constructor(
    private redis: RedisClient,
    private queueName: string,
    private rateLimit: number  // 每秒最多处理多少个
  ) {}
 
  async dequeue(): Promise<Task | null> {
    // 检查当前秒内的处理数
    const now = Math.floor(Date.now() / 1000)
    const countKey = `${this.queueName}:count:${now}`
    const count = await this.redis.get(countKey)
 
    if (count && parseInt(count) >= this.rateLimit) {
      return null  // 已达限流，等下一秒
    }
 
    // 取出任务
    const taskId = await this.redis.lpop(`${this.queueName}:pending`)
    if (!taskId) return null
 
    const data = await this.redis.get(`${this.queueName}:task:${taskId}`)
    if (!data) return null
 
    // 增加计数
    await this.redis.command(['INCR', countKey])
    await this.redis.command(['EXPIRE', countKey, '2'])  // 2 秒后过期
 
    return JSON.parse(data) as Task
  }
}

配合 Cron 使用——Cron 每秒触发一次，每次最多处理 rateLimit 个任务。

8. 死信队列

超过重试次数的任务进入死信队列，等待人工排查或手动重试。

async function handleDeadLetterQueue(redis: RedisClient, queueName: string): Promise<void> {
  // 取出死信任务
  const taskId = await redis.lpop(`${queueName}:dead-letter`)
  if (!taskId) return
 
  const data = await redis.get(`${queueName}:task:${taskId}`)
  if (!data) return
 
  const task = JSON.parse(data) as Task
 
  // 可以选择：
  // 1. 报警通知
  console.error('Dead letter task:', task)
 
  // 2. 手动重试（重置 attempts）
  // task.attempts = 0
  // await redis.set(`${queueName}:task:${taskId}`, JSON.stringify(task))
  // await redis.rpush(`${queueName}:pending`, taskId)
 
  // 3. 永久保存供排查
  // await redis.rpush(`${queueName}:dead-letter-archive`, data)
}

死信队列用 Cron 定期扫描——检查失败原因，自动重试可恢复的错误，不可恢复的报警。

9. 完整消费者

把前面的组件组合起来：

// src/consumers/redis-consumer.ts
import { RedisClient } from '../lib/redis'
import { PriorityQueue } from '../lib/priority-queue'
import { RateLimitedQueue } from '../lib/rate-limited-queue'
 
export async function redisConsumer(env: Env): Promise<void> {
  const redis = new RedisClient(env.UPSTASH_REDIS_REST_URL, env.UPSTASH_REDIS_REST_TOKEN)
 
  // 方案一：简单队列
  // const queue = new SimpleQueue(redis, 'tasks')
  // const task = await queue.dequeue()
 
  // 方案二：优先级队列
  const queue = new PriorityQueue(redis, 'tasks')
  const task = await queue.dequeue()
 
  // 方案三：限流队列
  // const queue = new RateLimitedQueue(redis, 'tasks', 10)
  // const task = await queue.dequeue()
 
  if (!task) return
 
  try {
    await processTask(task, env)
    await redis.del(`tasks:task:${task.id}`)
  } catch (err) {
    if (task.attempts < 3) {
      task.attempts += 1
      await redis.set(`tasks:task:${task.id}`, JSON.stringify(task))
      // 延迟重试
      await redis.zadd(
        'tasks:delayed',
        Date.now() + Math.pow(2, task.attempts) * 60 * 1000,
        task.id
      )
    } else {
      await redis.rpush('tasks:dead-letter', task.id)
    }
  }
}

总结

回顾这一节的要点：

Workers 环境通过 Upstash Redis REST API 接入 Redis
基于 Redis 可以实现延迟任务（Sorted Set）、优先级队列（多列表）、任务依赖（Hash）、限流（计数 key）
延迟任务用 Cron 扫描到期任务，精度取决于扫描频率
优先级队列按优先级顺序从不同列表取出任务
任务依赖用 Hash 记录前置任务状态，全部完成后触发后续任务
限流用计数 key + EXPIRE，每秒重置
死信队列保存失败任务，供人工排查或自动重试

下一篇讲 Cloudflare Queue 实践——原生集成、Cron Triggers、消费者配置、死信队列。