deno.com
本页内容

使用队列

Deno 运行时包含一个队列 API,支持卸载更大的工作负载进行异步处理,并保证至少一次交付排队消息。队列可用于卸载 Web 应用程序中的任务,或调度未来的工作单元。

您将主要与队列一起使用的 API 位于 Deno.Kv 命名空间中,作为 enqueuelistenQueue

入队消息 跳转到标题

要将消息入队进行处理,请在 Deno.Kv 实例上使用 enqueue 方法。在下面的示例中,我们展示了如何将通知入队以进行交付。

queue_example.ts
// Describe the shape of your message object (optional)
interface Notification {
  forUser: string;
  body: string;
}

// Get a reference to a KV instance
const kv = await Deno.openKv();

// Create a notification object
const message: Notification = {
  forUser: "alovelace",
  body: "You've got mail!",
};

// Enqueue the message for immediate delivery
await kv.enqueue(message);

您可以通过在毫秒中指定 delay 选项来将消息入队以便稍后交付。

// Enqueue the message for delivery in 3 days
const delay = 1000 * 60 * 60 * 24 * 3;
await kv.enqueue(message, { delay });

您还可以指定 Deno KV 中的一个键,如果您的消息因任何原因未交付,您的消息值将存储在该键下。

// Configure a key where a failed message would be sent
const backupKey = ["failed_notifications", "alovelace", Date.now()];
await kv.enqueue(message, { keysIfUndelivered: [backupKey] });

// ... disaster strikes ...

// Get the unsent message
const r = await kv.get<Notification>(backupKey);
// This is the message that didn't get sent:
console.log("Found failed notification for:", r.value?.forUser);

监听消息 跳转到标题

您可以使用 Deno.Kv 实例上的 listenQueue 方法配置一个 JavaScript 函数,该函数将处理添加到队列中的项。

listen_example.ts
// Define the shape of the object we expect as a message in the queue
interface Notification {
  forUser: string;
  body: string;
}

// Create a type guard to check the type of the incoming message
function isNotification(o: unknown): o is Notification {
  return (
    ((o as Notification)?.forUser !== undefined &&
      typeof (o as Notification).forUser === "string") &&
    ((o as Notification)?.body !== undefined &&
      typeof (o as Notification).body === "string")
  );
}

// Get a reference to a KV database
const kv = await Deno.openKv();

// Register a handler function to listen for values - this example shows
// how you might send a notification
kv.listenQueue((msg: unknown) => {
  // Use type guard - then TypeScript compiler knows msg is a Notification
  if (isNotification(msg)) {
    console.log("Sending notification to user:", msg.forUser);
    // ... do something to actually send the notification!
  } else {
    // If the message is of an unknown type, it might be an error
    console.error("Unknown message received:", msg);
  }
});

带 KV 原子事务的队列 API 跳转到标题

您可以将队列 API 与 KV 原子事务结合使用,以在同一事务中原子地入队消息并修改键。

kv_transaction_example.ts
const kv = await Deno.openKv();

kv.listenQueue(async (msg: unknown) => {
  const nonce = await kv.get(["nonces", msg.nonce]);
  if (nonce.value === null) {
    // This messaged was already processed
    return;
  }

  const change = msg.change;
  const bob = await kv.get(["balance", "bob"]);
  const liz = await kv.get(["balance", "liz"]);

  const success = await kv.atomic()
    // Ensure this message was not yet processed
    .check({ key: nonce.key, versionstamp: nonce.versionstamp })
    .delete(nonce.key)
    .sum(["processed_count"], 1n)
    .check(bob, liz) // balances did not change
    .set(["balance", "bob"], bob.value - change)
    .set(["balance", "liz"], liz.value + change)
    .commit();
});

// Modify keys and enqueue messages in the same KV transaction!
const nonce = crypto.randomUUID();
await kv
  .atomic()
  .check({ key: ["nonces", nonce], versionstamp: null })
  .enqueue({ nonce: nonce, change: 10 })
  .set(["nonces", nonce], true)
  .sum(["enqueued_count"], 1n)
  .commit();

队列行为 跳转到标题

消息传递保证 跳转到标题

运行时保证至少一次传递。这意味着对于大多数入队消息,listenQueue 处理程序将为每条消息调用一次。在某些故障场景中,处理程序可能会为同一消息多次调用以确保传递。设计您的应用程序时,正确处理重复消息非常重要。

您可以将队列与 KV 原子事务原语结合使用,以确保您的队列处理程序 KV 更新对于每条消息只执行一次。请参阅带 KV 原子事务的队列 API

自动重试 跳转到标题

当您的入队消息准备好交付时,将调用 listenQueue 处理程序来处理它们。如果您的处理程序抛出异常,运行时将自动重试调用处理程序,直到成功或达到最大重试次数。一旦 listenQueue 处理程序调用成功完成,消息就被认为是成功处理了。如果处理程序在重试时持续失败,消息将被丢弃。

消息传递顺序 跳转到标题

运行时尽力按照消息入队的顺序传递消息。但是,没有严格的顺序保证。有时,消息可能会乱序传递,以确保最大吞吐量。

Deno Deploy 上的队列 跳转到标题

Deno Deploy 提供队列 API 的全球化、无服务器、分布式实现,专为高可用性和吞吐量而设计。您可以使用它来构建可扩展以处理大型工作负载的应用程序。

即时隔离环境启动 跳转到标题

在 Deno Deploy 中使用队列时,当消息准备好进行处理时,隔离环境会自动按需启动,以调用您的 listenQueue 处理程序。在 Deno Deploy 应用程序中,定义 listenQueue 处理程序是启用队列处理的唯一要求,无需额外配置。

队列大小限制 跳转到标题

未交付队列消息的最大数量限制为 100,000 条。enqueue 方法在队列满时将失败并报错。

定价详情和限制 跳转到标题

  • enqueue 的处理方式与其他 Deno.Kv 写操作相同。入队消息会消耗 KV 存储和写入单元。
  • 通过 listenQueue 交付的消息会消耗请求和 KV 写入单元。
  • 有关更多信息,请参阅定价详情

用例 跳转到标题

队列在许多不同的场景中都很有用,但在构建 Web 应用程序时,您可能会看到一些常见的用例。

卸载异步进程 跳转到标题

有时,由客户端发起(如发送通知或 API 请求)的任务可能需要足够长的时间,您不希望客户端等待该任务完成才返回响应。其他时候,客户端根本不需要响应,例如当客户端向您的应用程序发送 Webhook 请求时,因此无需等待底层任务完成即可返回响应。

在这些情况下,您可以将工作卸载到队列中,以保持 Web 应用程序的响应性并立即向客户端发送反馈。要查看此用例的实际示例,请查看我们的Webhook 处理示例

调度未来的工作 跳转到标题

队列(以及像这样的队列 API)的另一个有用应用是调度工作在将来的适当时间发生。也许您想在客户下订单一天后向其发送通知,以发送满意度调查。您可以调度一个队列消息在 24 小时后交付,并设置一个监听器在该时间发送通知。

要查看调度未来通知的示例,请查看我们的通知示例

您找到所需内容了吗?

隐私政策