← Cloudflare Queues / queues / configuration

Пакетная обработка, повторные попытки и задержки

При настройке Worker-получатель для очереди вы также можете определить, как сообщения группируются в пакеты при доставке.

Пакетная обработка позволяет:

1. Уменьшите количество вызовов Worker-потребителя (это может снизить расходы).
2. Позволяет группировать сообщения в пакеты при записи во внешний API или сервис, сокращая количество запросов на запись.
3. Распределять нагрузку во времени, особенно если 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 настроенному потребителю.

Вы можете подтверждать отдельные сообщения в пакете, явно подтверждая каждое сообщение по мере его обработки. Явно подтвержденные сообщения не будут доставлены повторно, даже если потребитель очереди завершится ошибкой на следующем сообщении и (или) не сможет успешно завершить обработку пакета.

• Каждое сообщение можно подтверждать по мере обработки в рамках пакета, что позволяет избежать повторной доставки всего пакета целиком, если в процессе его обработки потребитель выдаёт ошибку.
• Подтверждение отдельных сообщений полезно, если вы обращаетесь к внешним 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() , чтобы явно потребовать повторной доставки сообщения в следующем пакете. Это называется «отрицательным подтверждением» (negative acknowledgement). Это особенно полезно, когда нужно обработать остальные сообщения пакета, не вызывая ошибку, которая привела бы к повторной доставке всего пакета.

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(). Вызов ackAll() для пакета сообщений (MessageBatch) доставленное вашему Worker-потребителю ведёт себя так же, как Worker-потребитель, который успешно завершает выполнение (не выбрасывает ошибку).

Обратите внимание, что вызовы ack(), retry() и их ackAll() / retryAll() эквиваленты подчиняются указанным ниже правилам приоритета:

• Если вы вызовете ack() для сообщения последующие вызовы ack() или retry() игнорируются без предупреждения.
• Если вы вызовете retry() для сообщения, а затем вызвать ack(): ack() игнорируется. Во всех случаях учитывается только первый вызов метода.
• Если вызвать ack() или retry() для отдельного сообщения, а затем любой из следующих вызовов (или оба сразу): ackAll() или retryAll() для пакета, приоритет имеет вызов для отдельного сообщения. То есть вызов на уровне пакета не применяется к этому сообщению (или сообщениям, если было сделано несколько вызовов).

Если доставить сообщение не удается, по умолчанию система повторяет попытку три раза, прежде чем пометить доставку как неуспешную. Вы можете задать max_retries (по умолчанию 3) при настройке потребителя, но в большинстве случаев рекомендуем оставить значение по умолчанию.

Сообщения, достигшие настроенного максимального количества повторов, удаляются из очереди, либо, если очередь недоставленных сообщений (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)

Вы также можете настроить глобальную задержку по умолчанию для конкретной очереди, передав --delivery-delay-secs при создании очереди через wrangler CLI:

# 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 для изменения настроек, связанных с очередью или потребителем очереди, вступит в силу самое последнее изменение конфигурации.

См. Документация по REST API для Queues , чтобы узнать, как программно настраивать задержки сообщений и задержки повторных попыток.

Сообщения можно задерживать по умолчанию на уровне очереди либо для отдельного сообщения (или пакета).

• Настройки задержки на уровне сообщения или пакета имеют приоритет над настройками на уровне очереди.
• Настройка 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 и желаемый коэффициент задержки в качестве аргумента для delaySeconds при вызове retry() для отдельного сообщения:

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,
                )
            )

• Просмотрите JavaScript API документацию по Queues.
• Подробнее о Как работает Queues.
• Изучите доступные метрики для ваших очередей, включая количество необработанных и отложенных сообщений.