← Cloudflare Queues / queues / configuration

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

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

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

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

По умолчанию для всех очередей включена конкурентность. Потребители очереди автоматически масштабируются до максимального числа одновременных вызовов по мере необходимости для управления очередью необработанных сообщений и/или частотой ошибок.

После обработки пакета сообщений Queues проверяет, нужно ли изменить число параллельных потребителей. Количество параллельно запускаемых потребителей очереди автоматически масштабируется с учётом нескольких факторов, в том числе:

• Количество сообщений в очереди (backlog) и скорость его роста.
• Доля неуспешных вызовов (по отношению к успешным). Вызов считается неуспешным, если ваш queue() обработчик возвращает неперехваченное исключение вместо void (ничего).
• Значение max_concurrency , установленного для этого потребителя.

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

Если вы записываете 100 сообщений в секунду в очередь, которую обрабатывает один потребитель, тратящий 5 секунд на пакет из 100 сообщений, количество необработанных сообщений будет расти быстрее, чем потребитель успевает их обрабатывать.

В этом сценарии Queues заметит растущий бэклог и увеличит число параллельных вызовов Worker-потребителя примерно до пяти (5), пока скорость поступления сообщений не снизится, потребитель не начнёт обрабатывать сообщения быстрее или не начнёт выдавать ошибки.

Если ваши потребители не масштабируются автоматически, вероятны следующие причины:

• max_concurrency установлено значение 1.
• Worker-потребитель возвращает ошибки вместо обработки сообщений. Проверьте потребителя, чтобы убедиться в его исправной работе.
• Обрабатывается пакет сообщений. Queues проверяет необходимость автомасштабирования потребителей только после обработки всего пакета сообщений, поэтому во время обработки пакета автомасштабирование не выполняется. Попробуйте уменьшить размер пакетов или доработать потребителя, чтобы он обрабатывал сообщения быстрее.

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

Настроить параллелизм Worker-потребителя можно двумя способами:

1. Настройка параметров параллелизма в панели управления Cloudflare
2. Настройте параметры параллелизма через конфигурационный файл Wrangler

Чтобы настроить параметры параллелизма для Worker-потребителя на панели управления:

1. На панели управления Cloudflare перейдите к разделу Queues страницу.

   Перейдите в Queues ↗

2. Выберите свою очередь > Настройки.

3. Выберите Редактировать потребителя в разделе Consumer details.

4. Задайте Максимальное количество вызовов потребителя значение в диапазоне от 1 и 250. Это значение представляет собой максимальное количество параллельных вызовов потребителя, доступных для вашей очереди.

Чтобы убрать фиксированное максимальное значение, выберите авто (рекомендуется).

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

Чтобы задать фиксированное максимальное количество одновременных вызовов потребителя для очереди, настройте max_concurrency в файле Wrangler:

{
	"queues": {
		"consumers": [
			{
				"queue": "my-queue",
				"max_concurrency": 1
			}
		]
	}
}

[[queues.consumers]]
queue = "my-queue"
max_concurrency = 1

Чтобы снять ограничение, удалите max_concurrency из [[queues.consumers]] конфигурацию для данной очереди и вызвать npx wrangler deploy , чтобы отправить обновление конфигурации.

При вызове нескольких Worker-потребителей каждый вызов Worker влечет за собой Стоимость процессорного времени.

• Если вы намерены обрабатывать все сообщения, записанные в очередь, итоговая стоимость в результате одинакова, даже при включённой параллельной обработке.
• Включение конкурентности просто переносит эти расходы вперёд по времени и помогает избежать ситуаций, когда сообщения достигают ограничение на хранение сообщений.

Тарификация потребителей соответствует Стандартная модель использования Workers то есть с разработчика взимается плата за запрос и за процессорное время, использованное в этом запросе.

Если Worker-потребитель обрабатывает пакет сообщений за 2 секунды, суммарная стоимость обработки 50 миллионов (50,000,000) сообщений будет одинаковой независимо от того, делает ли он это параллельно (быстрее) или по одному (медленнее).