本页内容
使用队列
Deno 运行时包含一个队列 API,支持卸载更大的工作负载进行异步处理,并保证至少一次交付排队消息。队列可用于卸载 Web 应用程序中的任务,或调度未来的工作单元。
您将主要与队列一起使用的 API 位于 Deno.Kv
命名空间中,作为 enqueue
和 listenQueue
。
入队消息 跳转到标题
要将消息入队进行处理,请在 Deno.Kv
实例上使用 enqueue
方法。在下面的示例中,我们展示了如何将通知入队以进行交付。
// 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 函数,该函数将处理添加到队列中的项。
// 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 原子事务结合使用,以在同一事务中原子地入队消息并修改键。
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 小时后交付,并设置一个监听器在该时间发送通知。
要查看调度未来通知的示例,请查看我们的通知示例。