为队列配置消费者 Worker 时，您还可以定义消息在投递时如何批处理。
批处理可以：
- 减少消费者 Worker 需要被调用的总次数（从而降低成本）。
- 在向外部 API 或服务写入时批处理消息（减少写入次数）。
- 随时间分散负载，尤其是当生产者 Worker 与用户面向的活动相关时。
有两种方式配置消息批处理。在将消费者 Worker 连接到队列时配置批处理。
max_batch_size- 投递给消费者的一批消息的最大大小（默认为 10 条消息）。max_batch_timeout- 队列在将批次投递给消费者之前等待的最大时间（默认为 5 秒）
例如，max_batch_size = 30 且 max_batch_timeout = 10 表示若 30 条消息写入队列，消费者将收到 30 条消息的批次。但是，若这 30 条消息写入队列超过 10 秒，则消费者将收到包含当时队列中消息数量的批次（在此情况下为 1 到 29 条之间）。
在确定要配置的大小和超时设置时，您需要考虑延迟（可以等待多久接收消息？）、整体批大小（写入外部系统时）和成本（更少但更大的批次）。
以下批级别设置可配置，以调整 Queues 向已配置消费者投递批次的方式。
| 设置 | 默认值 | 最小值 | 最大值 |
|---|---|---|---|
最大批大小 max_batch_size |
10 条消息 | 1 条消息 | 100 条消息 |
最大批超时 max_batch_timeout |
5 秒 | 0 秒 | 60 秒 |
您可以在处理批次中的每条消息时显式确认，从而确认批次中的单个消息。显式确认的消息不会重新投递，即使队列消费者在后续消息上失败和/或在处理批次时未能成功返回。
- 您可以在批次内处理每条消息时确认它，若消费者在批处理期间抛出错误，可避免整个批次被重新投递。
- 在调用外部 API、将消息写入数据库或对单个消息执行非幂等（改变状态）操作时，确认单个消息很有用。
要显式确认消息已投递，请调用消息上的 ack() 方法。
export default {
async queue(batch, env, ctx) {
for (const msg of batch.messages) {
// TODO: do something with the message
// Explicitly acknowledge the message as delivered
msg.ack();
}
},
};export default {
async queue(batch, env, ctx): Promise<void> {
for (const msg of batch.messages) {
// TODO: do something with the message
// Explicitly acknowledge the message as delivered
msg.ack();
}
},
} satisfies ExportedHandler<Env>;from workers import WorkerEntrypoint
class Default(WorkerEntrypoint):
async def queue(self, batch):
for msg in batch.messages:
# TODO: do something with the message
# Explicitly acknowledge the message as delivered
msg.ack()您还可以调用 retry() 显式强制消息在后续批次中重新投递。这称为"否定确认"。当您想处理该批次中的其余消息而不抛出会导致整个批次重新投递的错误时，这特别有用。
export default {
async queue(batch, env, ctx) {
for (const msg of batch.messages) {
// TODO: do something with the message that fails
msg.retry();
}
},
};export default {
async queue(batch, env, ctx): Promise<void> {
for (const msg of batch.messages) {
// TODO: do something with the message that fails
msg.retry();
}
},
} satisfies ExportedHandler<Env>;from workers import WorkerEntrypoint
class Default(WorkerEntrypoint):
async def queue(self, batch):
for msg in batch.messages:
# TODO: do something with the message that fails
msg.retry()您还可以在批次级别使用 ackAll() 和 retryAll() 确认或否定确认消息。对投递给消费者 Worker 的消息批次（MessageBatch）调用 ackAll() 的行为与消费者 Worker 成功返回（不抛出错误）相同。
请注意，对 ack()、retry() 及其 ackAll() / retryAll() 等效方法的调用遵循以下优先级规则：
- 若您对消息调用
ack()，随后对ack()或retry()的调用将被静默忽略。 - 若您对消息调用
retry()然后调用ack()：ack()被忽略。在所有情况下，第一个方法调用优先。 - 若您对单条消息调用
ack()或retry()，然后对批次调用ackAll()或retryAll()中的任意/全部，单条消息上的调用优先。即，批次级别调用不适用于该消息（或消息，若进行了多次调用）。
当消息投递失败时，默认行为是在标记投递失败之前重试三次。您可以在配置消费者时设置 max_retries（默认为 3），但在大多数情况下我们建议保留默认值。
达到配置的最大重试次数的消息将从队列中删除，或者若配置了 Dead Letter Queue（DLQ），则写入 DLQ。
当批次中的单条消息投递失败时，除非您已显式确认 该批次中的消息（或消息），否则整个批次将被重试。例如，若投递 10 条消息的批次，但第 8 条消息投递失败，全部 10 条消息将被重试，从而完整地重新投递给消费者。
向队列发布消息时，或标记消息或批次重试 时，您可以选择延迟消息的处理时间。
延迟消息允许您推迟任务，和/或在从队列消费时响应背压。例如，若您调用的上游 API 返回 HTTP 429: Too Many Requests，您可以延迟消息以减慢消费速度，然后再重新处理。
消息最多可延迟 24 小时。
向队列发送消息或一批消息时延迟，可在发送消息时提供 delaySeconds 参数。
// Delay a singular message by 600 seconds (10 minutes)
await env.YOUR_QUEUE.send(message, { delaySeconds: 600 });
// Delay a batch of messages by 300 seconds (5 minutes)
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 300 });
// Do not delay this message.
// If there is a global delay configured on the queue, ignore it.
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 0 });// Delay a singular message by 600 seconds (10 minutes)
await env.YOUR_QUEUE.send(message, { delaySeconds: 600 });
// Delay a batch of messages by 300 seconds (5 minutes)
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 300 });
// Do not delay this message.
// If there is a global delay configured on the queue, ignore it.
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 0 });# Delay a singular message by 600 seconds (10 minutes)
await env.YOUR_QUEUE.send(message, delaySeconds=600)
# Delay a batch of messages by 300 seconds (5 minutes)
await env.YOUR_QUEUE.sendBatch(messages, delaySeconds=300)
# Do not delay this message.
# If there is a global delay configured on the queue, ignore it.
await env.YOUR_QUEUE.sendBatch(messages, delaySeconds=0)您还可以通过在 wrangler CLI 创建队列时传递 --delivery-delay-secs，按队列配置默认的全局延迟：
# Delay all messages by 5 minutes as a default
npx wrangler queues create $QUEUE-NAME --delivery-delay-secs=300从队列消费消息 时，您可以选择显式标记消息重试。消息可以单独重试和延迟，或作为整个批次。
要在批次内延迟单条消息：
export default {
async queue(batch, env, ctx) {
for (const msg of batch.messages) {
// Mark for retry and delay a singular message
// by 3600 seconds (1 hour)
msg.retry({ delaySeconds: 3600 });
}
},
};export default {
async queue(batch, env, ctx): Promise<void> {
for (const msg of batch.messages) {
// Mark for retry and delay a singular message
// by 3600 seconds (1 hour)
msg.retry({ delaySeconds: 3600 });
}
},
} satisfies ExportedHandler<Env>;from workers import WorkerEntrypoint
class Default(WorkerEntrypoint):
async def queue(self, batch):
for msg in batch.messages:
# Mark for retry and delay a singular message
# by 3600 seconds (1 hour)
msg.retry(delaySeconds=3600)要延迟一批消息：
export default {
async queue(batch, env, ctx) {
// Mark for retry and delay a batch of messages
// by 600 seconds (10 minutes)
batch.retryAll({ delaySeconds: 600 });
},
};export default {
async queue(batch, env, ctx): Promise<void> {
// Mark for retry and delay a batch of messages
// by 600 seconds (10 minutes)
batch.retryAll({ delaySeconds: 600 });
},
} satisfies ExportedHandler<Env>;from workers import WorkerEntrypoint
class Default(WorkerEntrypoint):
async def queue(self, batch):
# Mark for retry and delay a batch of messages
# by 600 seconds (10 minutes)
batch.retryAll(delaySeconds=600)您还可以选择为因隐式失败或显式调用 retry() 而重试的任何消息设置默认重试延迟。这在消费者级别设置，push（Worker）和 pull（HTTP）消费者均支持。
可通过 wrangler CLI 配置延迟：
# Push-based consumers
# Delay any messages that are retried by 60 seconds (1 minute) by default.
npx wrangler@latest queues consumer worker add $QUEUE-NAME $WORKER_SCRIPT_NAME --retry-delay-secs=60
# Pull-based consumers
# Delay any messages that are retried by 60 seconds (1 minute) by default.
npx wrangler@latest queues consumer http add $QUEUE-NAME --retry-delay-secs=60延迟也可以在 Wrangler 配置文件 中使用生产者的 delivery_delay 设置（发送时）和/或每个消费者的 retry_delay（重试时）进行配置：
{
"queues": {
"producers": [
{
"binding": "<BINDING_NAME>",
"queue": "<QUEUE-NAME>",
"delivery_delay": 60 // delay every message delivery by 1 minute
}
],
"consumers": [
{
"queue": "my-queue",
"retry_delay": 300 // delay any retried message by 5 minutes before re-attempting delivery
}
]
}
}[[queues.producers]]
binding = "<BINDING_NAME>"
queue = "<QUEUE-NAME>"
delivery_delay = 60
[[queues.consumers]]
queue = "my-queue"
retry_delay = 300若您同时使用 wrangler CLI 和 Wrangler 配置文件 更改与队列或队列消费者相关的设置，最近的配置更改将生效。
请参阅 Queues REST API 文档 了解如何以编程方式配置消息延迟和重试延迟。
消息可以在队列级别默认延迟，或在每条消息（或批次）级别延迟。
- 每条消息/批次延迟设置优先于队列级别设置。
- 发送或重试时设置
delaySeconds: 0将忽略任何队列级别延迟，并使消息在下一批次中投递。 - 以
delaySeconds: <any positive integer>发送到具有较短默认延迟的队列或重试的消息，仍将遵循消息级别设置。
您可以应用退避算法，根据当前投递尝试次数递增延迟消息。
投递给消费者的消息包含 attempts 属性，跟踪已进行的投递尝试次数。
例如，要为消息生成指数退避 ↗，可以创建计算此值的辅助函数：
function calculateExponentialBackoff(attempts, baseDelaySeconds) {
return baseDelaySeconds ** attempts;
}function calculateExponentialBackoff(
attempts: number,
baseDelaySeconds: number,
): number {
return baseDelaySeconds ** attempts;
}def calculate_exponential_backoff(attempts, base_delay_seconds):
return base_delay_seconds ** attempts在消费者中，您将 msg.attempts 的值和所需的延迟因子作为参数传递给 retry() 调用单条消息时的 delaySeconds：
const BASE_DELAY_SECONDS = 30;
export default {
async queue(batch, env, ctx) {
for (const msg of batch.messages) {
// Mark for retry with exponential backoff
msg.retry({
delaySeconds: calculateExponentialBackoff(
msg.attempts,
BASE_DELAY_SECONDS,
),
});
}
},
};const BASE_DELAY_SECONDS = 30;
export default {
async queue(batch, env, ctx): Promise<void> {
for (const msg of batch.messages) {
// Mark for retry with exponential backoff
msg.retry({
delaySeconds: calculateExponentialBackoff(
msg.attempts,
BASE_DELAY_SECONDS,
),
});
}
},
} satisfies ExportedHandler<Env>;from workers import WorkerEntrypoint
BASE_DELAY_SECONDS = 30
class Default(WorkerEntrypoint):
async def queue(self, batch):
for msg in batch.messages:
# Mark for retry and delay a singular message
# by 3600 seconds (1 hour)
msg.retry(
delaySeconds=calculate_exponential_backoff(
msg.attempts,
BASE_DELAY_SECONDS,
)
)- 查看 Queues 的 JavaScript API 文档。
- 了解更多关于 Queues 工作原理。
- 了解队列可用的指标，包括积压和延迟消息计数。