---

name: upstash-qstash description: Upstash QStash 专家，用于无服务器消息队列、定时任务和可靠的基于 HTTP 的任务投递，无需管理基础设施。 risk: unknown source: vibeship-spawner-skills (Apache 2.0) date_added: 2026-02-27

Upstash QStash

Upstash QStash 专家，用于无服务器消息队列、定时任务和可靠的基于 HTTP 的任务投递，无需管理基础设施。

原则

• HTTP 是接口 - 只要支持 HTTPS，就能使用 QStash
• 端点必须公开 - QStash 从云端调用你的 URL
• 始终验证签名 - 永远不要信任未验证的 Webhook
• 定时任务是即发即忘的 - QStash 处理 cron
• 重试是内置的 - 但需要根据你的用例配置
• 延迟是免费的 - 可以安排从几秒到几天后的任务
• 回调完成闭环 - 了解投递成功或失败
• 去重防止重复处理 - 使用消息 ID

能力

• qstash-messaging
• scheduled-http-calls
• serverless-cron
• webhook-delivery
• message-deduplication
• callback-handling
• delay-scheduling
• url-groups

范围

• complex-workflows -> inngest
• redis-queues -> bullmq-specialist
• event-sourcing -> event-architect
• workflow-orchestration -> temporal-craftsman

工具

核心

• qstash-sdk
• upstash-console

框架

• nextjs
• cloudflare-workers
• vercel-functions
• aws-lambda
• netlify-functions

模式

• scheduled-jobs
• delayed-messages
• webhook-fanout
• callback-verification

相关

• upstash-redis
• upstash-kafka

模式

基本消息发布

发送消息到端点进行投递

何时使用：需要可靠的异步 HTTP 调用

import { Client } from '@upstash/qstash';

const qstash = new Client({
  token: process.env.QSTASH_TOKEN!,
});

// 简单消息到端点
await qstash.publishJSON({
  url: 'https://myapp.com/api/process',
  body: {
    userId: '123',
    action: 'welcome-email',
  },
});

// 带延迟（1 小时后处理）
await qstash.publishJSON({
  url: 'https://myapp.com/api/reminder',
  body: { userId: '123' },
  delay: 60 * 60,  // 秒
});

// 指定投递时间
await qstash.publishJSON({
  url: 'https://myapp.com/api/scheduled',
  body: { report: 'daily' },
  notBefore: Math.floor(Date.now() / 1000) + 86400,  // 明天
});

定时 Cron 任务

设置周期性定时任务

何时使用：需要无基础设施的周期性后台任务

import { Client } from '@upstash/qstash';

const qstash = new Client({
  token: process.env.QSTASH_TOKEN!,
});

// 创建定时任务
const schedule = await qstash.schedules.create({
  destination: 'https://myapp.com/api/cron/daily-report',
  cron: '0 9 * * *',  // 每天 UTC 9:00
  body: JSON.stringify({ type: 'daily' }),
  headers: {
    'Content-Type': 'application/json',
  },
});

console.log('Schedule created:', schedule.scheduleId);

// 列出所有定时任务
const schedules = await qstash.schedules.list();

// 删除定时任务
await qstash.schedules.delete(schedule.scheduleId);

签名验证

在端点中验证 QStash 消息签名

何时使用：任何接收 QStash 消息的端点（必须！）

// app/api/webhook/route.ts (Next.js App Router)
import { Receiver } from '@upstash/qstash';
import { NextRequest, NextResponse } from 'next/server';

const receiver = new Receiver({
  currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
  nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
});

export async function POST(req: NextRequest) {
  const signature = req.headers.get('upstash-signature');
  const body = await req.text();

  // 始终验证签名
  const isValid = await receiver.verify({
    signature: signature!,
    body,
    url: req.url,
  });

  if (!isValid) {
    return NextResponse.json(
      { error: 'Invalid signature' },
      { status: 401 }
    );
  }

  // 安全处理
  const data = JSON.parse(body);
  await processMessage(data);

  return NextResponse.json({ success: true });
}

投递状态回调

在消息投递成功或失败时获得通知

何时使用：需要跟踪关键消息的投递状态

import { Client } from '@upstash/qstash';

const qstash = new Client({
  token: process.env.QSTASH_TOKEN!,
});

// 带回调发布
await qstash.publishJSON({
  url: 'https://myapp.com/api/critical-task',
  body: { taskId: '456' },
  callback: 'https://myapp.com/api/qstash-callback',
  failureCallback: 'https://myapp.com/api/qstash-failed',
});