mcpField
Q
queues

queues

github.com/vercel开发318135 次安装2 天前更新
代码仓库

关于 queues

面向需要持久化主题和独立消费者的开发者，介绍如何使用 Vercel Queues 构建后台消息处理。内容涵盖消息发布、推送或轮询消费、重试与保留期限，并说明何时应改用 Workflows。

主要功能

  1. 01持久化主题与消费者组
  2. 02推送和轮询消费模式
  3. 03至少一次投递与可配置重试
  4. 04消息延迟、保留期限和幂等选项
  5. 05JavaScript 与 Python SDK 指南
  6. 06Queues 与 Workflows 的选型说明

适用场景

  1. 01缓冲后台任务以供后续处理
  2. 02向多个独立消费者组发送事件
  3. 03通过 Vercel 函数或外部 Worker 消费消息
  4. 04为新消费者组重放仍在保留期内的消息

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/vercel/vercel-plugin --skill queues

安装后在 Claude Code 中输入/queues确认已加载。

附带文件.vercel.approvers

详细介绍

Vercel Queues

你是 Vercel Queues 方面的专家。Vercel Queues 是一种持久化消息主题，为 Vercel 上的后台工作和 Agent 事件提供支持。

它是什么

Vercel Queues（公开测试版）为你提供持久化、仅追加的主题。生产者发布 JSON 消息，每个已订阅的消费者组都会收到每条消息，并采用至少一次投递和自动重试。新的消费者组可以稍后加入，并重放尚未过期的历史消息。Queues 是 Vercel Workflows 底层使用的基本构件；如果你需要控制发布、消费和路由，请直接使用 Queues。

  • 主题：有名称的持久化消息日志，在首次发布时创建
  • 消费者组：独立的订阅者，可接收主题上的每条消息
  • 投递：至少一次；处理程序必须具备幂等性
  • 保留期限：默认 24 小时，最长 7 天；投递最多可延迟至保留期限
  • 模式：推送（由 Vercel 调用你的函数）或轮询（由你自己的工作进程从任意环境拉取消息）

选择 Queues 还是 Workflows

需求使用原因
即发即弃的后台任务、扇出、缓冲Queues直接发布和消费，消费者组相互独立
包含休眠、钩子或人工审批的多步逻辑Workflows（⤳ skill: workflow）基于 Queues 构建的持久化步骤和重放
按 cron 定时调用Cron Jobs（⤳ skill: vercel-functions）基于时间的触发，而非基于消息的触发

快速入门（Next.js App Router）

安装 SDK：

npm install @vercel/queue

从任意路由、Server Action 或函数发布：

// app/api/orders/route.ts
import { send } from '@vercel/queue';

export async function POST(request: Request) {
  const body = await request.json();
  const { messageId } = await send('orders', { orderId: body.orderId, action: 'process' });
  return Response.json({ messageId });
}

使用推送模式处理程序消费消息。处理程序返回时，消息会被确认；抛出异常时，消息会重试：

// app/api/queues/process-order/route.ts
import { handleCallback } from '@vercel/queue';

export const POST = handleCallback(async (message, metadata) => {
  await processOrder(message);
  console.log('processed', metadata.messageId, 'delivery', metadata.deliveryCount);
});

在 vercel.json（或 vercel.ts）中注册消费者，以便 Vercel 将主题路由到该函数：

{
  "functions": {
    "app/api/queues/process-order/route.ts": {
      "experimentalTriggers": [{ "type": "queue/v2beta", "topic": "orders" }]
    }
  }
}

开始本地开发前，先运行 vercel link 和 vercel env pull，以便 SDK 进行身份验证。

发送选项

await send('orders', payload, {
  region: 'sfo1',            // target a specific region
  retentionSeconds: 3600,    // message TTL; min 60, max 604800 (7 days); default 24 hours
  delaySeconds: 60,          // delay first delivery; max 7 days, capped at the TTL
  idempotencyKey: 'order-123', // duplicates within min(retention, 24 hours) are dropped
  headers: { 'x-trace-id': 'abc-123' },
});

如果需要默认值、固定区域或多个客户端，请创建一个 QueueClient：

// lib/queue.ts
import { QueueClient } from '@vercel/queue';

const queue = new QueueClient({ region: 'sfo1' });
export const { send, handleCallback } = queue;

消费者选项和重试

handleCallback(handler, options) 接受：

选项默认值说明
visibilityTimeoutSeconds300消息保持处理中状态的时长；处理程序运行期间，SDK 会延长租约
retry触发 retryAfterSeconds（60s）(error, metadata) => { afterSeconds } | { acknowledge: true } | undefined

达到投递次数阈值后，通过确认来处理毒性消息：

export const POST = handleCallback(processOrder, {
  retry: (error, metadata) => {
    if (metadata.deliveryCount > 5) return { acknowledge: true }; // stop retrying
    return { afterSeconds: Math.min(300, 2 ** metadata.deliveryCount * 5) };
  },
});

metadata 包含 messageId、deliveryCount、createdAt、expiresAt、topicName、consumerGroup 和 region。

对于 Express、Connect 或 Next.js Pages Router 处理程序，请使用 QueueClient 实例中的 queue.handleNodeCallback(async (message, metadata) => ...)，它接收 (req, res)。

其他运行时和框架

  • Python：vercel-queue 使用相同的主题模型进行发布和消费，FastAPI、Flask 和 Django 应用均可使用；Celery 和 Dramatiq 集成见 Python 后端框架相关文档。
  • Nitro / Nuxt：在 nitro.config.ts 中声明 vercel.queues.triggers，并使用 vercel:queue 运行时钩子处理消息；在任何服务端路由中都可以使用 @vercel/queue 提供的 send。
  • 轮询模式：当向 Vercel Function 推送消息不适用时，可在任意环境中由自己的工作进程拉取消息。在 Vercel 上，将始终运行的轮询模式消费者作为守护进程运行（⤳ skill: daemons）。
  • 负载：默认使用 JSON；构造 QueueClient 时，可对二进制数据使用 BufferTransport，或对大型消息体使用 StreamTransport。

错误

@vercel/queue 导出带类型的错误：UnauthorizedError、BadRequestError、MessageNotFoundError 和 QueueEmptyError。重复的幂等键不会抛出错误；重复消息会被静默丢弃。

常见问题

  1. 缺少触发器：没有对应 experimentalTriggers 条目的 handleCallback 路由永远不会收到消息。请在 vercel.json/vercel.ts 中注册每个消费者。
  2. 处理程序不具备幂等性：消息至少会投递一次。请使用 metadata.messageId 或你自己的 idempotencyKey 标记副作用。
  3. 无限重试：如果没有能够确认有问题消息的 retry 策略，也没有触发器 maxDeliveries 上限，永久失败的消息会不断重新投递，直到过期。
  4. 用队列处理多步骤逻辑：如果步骤之间需要休眠、钩子或审批，请使用工作流，而不是手动串联主题。
  5. 本地开发时缺少凭据：请先运行 vercel link 和 vercel env pull；否则 send() 会抛出 Failed to get OIDC token for local development。

参考资料

评论

支持 Markdown