← Cloudflare Queues / queues / configuration

Pull-потребители

Потребитель на основе pull позволяет получать сообщения из очереди по HTTP из любого окружения и на любом языке программирования вне Cloudflare Workers. Такой потребитель полезен, если скорость обработки сообщений ограничена вышестоящей инфраструктурой или длительными задачами.

Как выбрать между push- и pull-потребителем

То, какой тип потребителя выбрать, push-based или pull-based, зависит от того, как вы используете свои очереди, а также от конфигурации инфраструктуры, находящейся выше по потоку от потребителя очереди.

Начиная с push-потребитель это самый простой способ начать работу и получать сообщения из очереди. Потребитель на основе push работает на Workers и по умолчанию автоматически масштабируется, получая сообщения по мере их поступления в очередь.
Используйте pull-потребителя, если нужно получать сообщения из существующей инфраструктуры вне Cloudflare Workers или требуется точно контролировать скорость обработки сообщений. Pull-потребитель должен явно запрашивать сообщения из очереди (pull) и подтверждать их получение только тогда, когда он к этому готов.

Вы можете в любой момент удалить потребителя и подключить нового, что позволяет переключиться с pull-потребителя на push-потребителя при изменении требований.

Чтобы настроить потребителя с получением по запросу (pull-based) и получать сообщения из очереди, необходимо:

Включите HTTP pull для очереди.
Создайте действительный токен аутентификации для HTTP-клиента.
Извлекает пакеты сообщений из очереди.
Подтверждение и/или повторная отправка сообщений в пакете.

1. Включение HTTP pull

Вы можете включить HTTP pull или изменить тип очереди с push на pull через wrangler CLI или через Панель управления Cloudflare ↗. Включение HTTP pull для конфигурационный файл Wrangler больше не поддерживается.

wrangler CLI

Вы можете включить pull-потребителя для любой существующей очереди с помощью wrangler queues consumer http и указания имени очереди.

npx wrangler queues consumer http add $QUEUE-NAME

Если у вас уже есть потребитель на основе push, сначала удалите его. wrangler вернёт ошибку, если вы попытаетесь вызвать consumer http add для очереди с уже существующей конфигурацией потребителя:

wrangler queues consumer worker remove $QUEUE-NAME $SCRIPT_NAME

2. Аутентификация потребителя

Потребителям HTTP Pull требуется API-токен на com.cloudflare.api.account.queues_read и com.cloudflare.api.account.queues_write разрешения.

Как чтение, и write обязательны, поскольку потребителю на основе pull необходимо изменять состояние очереди для подтверждения получаемых сообщений. Получение сообщений изменяет состояние очереди.

API-токены передаются в виде Bearer-токенов в Authorization заголовок HTTP-запроса в формате Authorization: Bearer $YOUR_TOKEN_HERE. В следующем примере показано, как передать токен API с помощью curl HTTP-клиент:

curl "https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull" \
--header "Authorization: Bearer ${QUEUES_TOKEN}" \
--header "Content-Type: application/json" \
--data '{ "visibility_timeout_ms": 10000, "batch_size": 2 }'

Вы можете аутентифицироваться и запускать несколько параллельных pull-потребителей для одной очереди.

Создание API-токенов

Чтобы создать API-токен:

Войдите в Панель управления Cloudflare ↗.
Перейдите в My Profile > API-токены ↗.
Выберите Create Token.
Прокрутите страницу до конца и выберите Create Custom Token.
Присвойте токену имя. Например, queue-pull-token.
В разделе Разрешения разделе выберите Аккаунт и затем Queues. Убедитесь, что вы выбрали Изменить (чтение и запись).
(Необязательно) Выберите Все учётные записи (по умолчанию) или конкретную учётную запись, к которой нужно ограничить область действия токена.
Выберите Continue to summary и затем Создать токен.

Обязательно запишите токен: он отображается только один раз.

3. Получение сообщений

Чтобы получить сообщение, отправьте HTTP POST запрос к REST API для Queues с телом в формате JSON, которое может дополнительно указывать visibility_timeout и batch_size, либо пустой объект JSON ({}):

index.js

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull with the timeout & batch size
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// Optional - you can provide an empty object '{}' and the defaults will apply.
		body: JSON.stringify({ visibility_timeout_ms: 6000, batch_size: 50 }),
	},
);

index.ts

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull with the timeout & batch size
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// Optional - you can provide an empty object '{}' and the defaults will apply.
		body: JSON.stringify({ visibility_timeout_ms: 6000, batch_size: 50 }),
	},
);

import json
from workers import fetch

# POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull with the timeout & batch size

resp = await fetch(
	f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/queues/{QUEUE_ID}/messages/pull",
	method="POST",
	headers={
		"content-type": "application/json",
		"authorization": f"Bearer {QUEUES_API_TOKEN}",
	}, # Optional - you can provide an empty object '{}' and the defaults will apply.
	body=json.dumps({"visibility_timeout_ms": 6000, "batch_size": 50}),
)

Это вернёт массив сообщений (до указанного batch_size) в следующем формате:

{
	"success": true,
	"errors": [],
	"messages": [],
	"result": {
		"message_backlog_count": 10,
		"messages": [
			{
				"body": "hello",
				"id": "1ad27d24c83de78953da635dc2ea208f",
				"timestamp_ms": 1689615013586,
				"attempts": 2,
				"metadata": {
					"CF-sourceMessageSource": "dash",
					"CF-Content-Type": "json"
				},
				"lease_id": "eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0..NXmbr8h6tnKLsxJ_AuexHQ.cDt8oBb_XTSoKUkVKRD_Jshz3PFXGIyu7H1psTO5UwI.smxSvQ8Ue3-ymfkV6cHp5Va7cyUFPIHuxFJA07i17sc"
			},
			{
				"body": "world",
				"id": "95494c37bb89ba8987af80b5966b71a7",
				"timestamp_ms": 1689615013586,
				"attempts": 2,
				"metadata": {
					"CF-sourceMessageSource": "dash",
					"CF-Content-Type": "json"
				},
				"lease_id": "eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0..QXPgHfzETsxYQ1Vd-H0hNA.mFALS3lyouNtgJmGSkTzEo_imlur95EkSiH7fIRIn2U.PlwBk14CY_EWtzYB-_5CR1k30bGuPFPUx1Nk5WIipFU"
			}
		]
	}
}

Pull-потребители используют подход «short polling»: если есть сообщения, готовые к доставке, Queues сразу вернёт ответ с сообщениями, количество которых не превышает настроенный batch_size. Если сообщений для доставки нет, Queues вернёт пустой ответ. Queues не удерживает открытое соединение (часто называемое "long polling"), если сообщений для доставки нет.

Объект сообщения содержит пять полей:

body: это может быть закодировано в base64 в зависимости от тип содержимого, с которым было опубликовано сообщение.
id: уникальный временный идентификатор сообщения, доступный только для чтения.
timestamp_ms: время публикации сообщения в очередь в миллисекундах, прошедших с Эпоха Unix ↗. Это позволяет определить возраст сообщения путём вычитания этого значения из текущей временной метки.
attempts: сколько раз была предпринята попытка полной доставки сообщения. Когда это значение достигает max_retries, сообщение не будет доставлено повторно и будет безвозвратно удалено из очереди.
lease_id: закодированный идентификатор аренды сообщения. lease_id используется для явного подтверждения сообщения или его повторной обработки.

lease_id позволяет вашему pull-потребителю явно подтвердить некоторые, все или ни одного сообщения в пакете либо пометить их для повторной попытки. Если потребитель не подтверждает сообщения и не помечает их для повторной попытки, они будут помечены для повторной доставки после того, как visibility_timeout достигнут. lease_id становится недействительным по истечении этого тайм-аута.

Вы можете настроить как batch_size и visibility_timeout при получении сообщений из очереди:

batch_size (по умолчанию 5, максимум 100): сколько сообщений возвращается потребителю при каждом извлечении.
visibility_timeout (по умолчанию 30 секунд, максимум 12 часов): определяет, сколько времени отводится потребителю на явное подтверждение сообщений, доставленных в пакете, в зависимости от их lease_id. По истечении этого времени ожидания сообщения считаются неподтверждёнными и снова ставятся в очередь для повторной доставки.

Параллельные потребители

У вас может быть несколько HTTP-клиентов, одновременно получающих сообщения из одной очереди: каждый клиент получит уникальный пакет сообщений и сохранит "аренду" на эти сообщения до visibility_timeout истечёт либо пока эти сообщения не будут помечены для повторной обработки.

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

Несколько потребителей могут быть полезны, если у вас несколько вышестоящих ресурсов (например, инфраструктура на GPU) и вы хотите настроить автомасштабирование на основе объём накопленных сообщений очереди, и (или) стоимость.

4. Подтверждение сообщений

Сообщения, полученные потребителем, необходимо либо подтвердить, либо пометить для повтора.

Чтобы подтвердить и (или) пометить сообщения для повторной попытки, выполните запрос HTTP POST запрос к /ack эндпоинт вашей очереди согласно REST API для Queues предоставив массив lease_id объектов для подтверждения и/или повторной обработки:

index.js

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack with the lease_ids
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// If you have no messages to retry, you can specify an empty array - retries: []
		body: JSON.stringify({
			acks: [
				{ lease_id: "lease_id1" },
				{ lease_id: "lease_id2" },
				{ lease_id: "etc" },
			],
			retries: [{ lease_id: "lease_id4" }],
		}),
	},
);

index.ts

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack with the lease_ids
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// If you have no messages to retry, you can specify an empty array - retries: []
		body: JSON.stringify({
			acks: [
				{ lease_id: "lease_id1" },
				{ lease_id: "lease_id2" },
				{ lease_id: "etc" },
			],
			retries: [{ lease_id: "lease_id4" }],
		}),
	},
);

import json
from workers import fetch

# POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack with the lease_ids

resp = await fetch(
	f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/queues/{QUEUE_ID}/messages/ack",
	method="POST",
	headers={
		"content-type": "application/json",
		"authorization": f"Bearer {QUEUES_API_TOKEN}",
	}, # If you have no messages to retry, you can specify an empty array - retries: []
	body=json.dumps({
		"acks": [
			{"lease_id": "lease_id1"},
			{"lease_id": "lease_id2"},
			{"lease_id": "etc"},
		],
		"retries": [{"lease_id": "lease_id4"}],
	}),
)

При пометке сообщения для повтора вы можете дополнительно указать количество секунд задержки, передав { lease_id: string, delay_seconds: number } объект в retries массив:

{
	"acks": [
		{ "lease_id": "lease_id1" },
		{ "lease_id": "lease_id2" },
		{ "lease_id": "lease_id3" }
	],
	"retries": [{ "lease_id": "lease_id4", "delay_seconds": 600 }]
}

Кроме того:

Вам следует передавать каждый lease_id в запросе к /ack эндпоинт, если вы обрабатываете эти сообщения в потребителе. Если вы не подтвердите сообщение, оно будет помечено для повторной доставки (возвращено в очередь).
При необходимости вы можете пометить сообщения для повторной обработки, например, если при обработке сообщения возникла ошибка или наблюдается нагрузка на вышестоящие ресурсы. Явная пометка сообщения для повтора немедленно возвращает его в очередь, вместо ожидания (потенциально долгого) visibility_timeout будет достигнуто.
Вы можете выполнять несколько вызовов /ack эндпоинт по мере обработки пакета сообщений, но мы рекомендуем группировать подтверждения, чтобы сократить количество необходимых вызовов API.

Queues старается быть гибким в отношении lease ID: если потребитель подтверждает сообщение по его lease ID после истекает время видимости, Queues всё равно примет это подтверждение. Если за это время сообщение было доставлено другому потребителю, он тоже сможет подтвердить его без ошибки.

Типы содержимого

При публикации в очередь с внешним потребителем учитывайте, что некоторые типы содержимого могут кодироваться таким образом, чтобы их можно было безопасно сериализовать в объект JSON.

Как для json и bytes типов содержимого это означает, что они будут закодированы в base64 (RFC 4648 ↗). text будет отправлен как обычная строка в кодировке UTF-8.

Вашему потребителю потребуется декодировать json и bytes перед выполнением операций с данными.

Следующие шаги

Просмотрите Документация REST API и схему для Queues.
Подробнее о как выполнять вызовы API в Cloudflare API.
Изучите какие ограничения действуют при получении сообщений из очереди и записи в неё.