Queues、Cron 与 Workflows

Queues:消息队列

创建与绑定

wrangler queues create send-email

# wrangler.toml 生产者
[[queues.producers]]
queue = "send-email"
binding = "EMAIL_Q"

# 消费者(可以是另一个 Worker)
[[queues.consumers]]
queue = "send-email"
max_batch_size = 10
max_batch_timeout = 5   # 最多攒 5s 或 10 条触发
max_retries = 3
dead_letter_queue = "send-email-dlq"

生产消息

app.post('/signup', async (c) => {
  const { email } = await c.req.json();
  // 立即返回给用户,邮件走队列异步发
  await c.env.EMAIL_Q.send({ type: 'welcome', to: email });
  return c.json({ ok: true });
});

// 批量发
await c.env.EMAIL_Q.sendBatch([
  { body: { type: 'news', to: 'a@x' } },
  { body: { type: 'news', to: 'b@x' } },
]);

消费消息

export default {
  async queue(batch: MessageBatch, env: Env) {
    for (const msg of batch.messages) {
      try {
        await sendEmail(msg.body);
        msg.ack();  // 成功
      } catch (e) {
        msg.retry({ delaySeconds: 60 });  // 1 分钟后重试
      }
    }
  },
};

Cron Triggers:定时任务

[triggers]
crons = [
  "*/5 * * * *",         # 每 5 分钟
  "0 2 * * *",            # 每天凌晨 2 点
  "0 0 * * 1",            # 每周一零点
]

export default {
  async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
    if (event.cron === '0 2 * * *') {
      ctx.waitUntil(runDailyReport(env));
    }
  },
};

Workflows:持久化长流程

2025 GA 的 Cloudflare Workflows——像 AWS Step Functions 但原生 JS。把一个跨分钟/小时的流程写成代码,每一步自动持久化,失败自动重试,断点续跑。

定义

import { WorkflowEntrypoint, WorkflowEvent, WorkflowStep } from 'cloudflare:workers';

export class OrderFulfillment extends WorkflowEntrypoint<Env, { orderId: string }> {
  async run(event: WorkflowEvent<{ orderId: string }>, step: WorkflowStep) {
    const order = await step.do('fetch order', async () => {
      return await this.env.DB.prepare('SELECT * FROM orders WHERE id=?')
        .bind(event.payload.orderId).first();
    });

    await step.do('charge card', { retries: { limit: 3, delay: '30 seconds' } },
      async () => {
        await stripeCharge(order);
      },
    );

    await step.sleep('wait for warehouse', '30 minutes');

    await step.do('ship', async () => {
      await callShippingAPI(order);
    });

    await step.do('notify user', async () => {
      await this.env.EMAIL_Q.send({ type: 'shipped', to: order.email });
    });
  }
}

触发

app.post('/checkout', async (c) => {
  const order = await createOrder(c);
  const instance = await c.env.ORDER_WF.create({ params: { orderId: order.id } });
  return c.json({ workflowId: instance.id });
});

关键语义

三者选型

实战组合:用户注册后的全套异步

本章小结