← Cloudflare Queues / queues

Начало работы

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

Для использования Queues понадобится:

1. Зарегистрируйтесь для получения Аккаунт Cloudflare ↗.
2. Установка Node.js ↗.

Вы будете обращаться к очереди из Worker, Worker-производителя. Вам нужно создать хотя бы один Worker-производитель для публикации сообщений в очередь. Если вы используете R2 Bucket Event Notifications, то Worker-производитель вам не понадобится.

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

npm create cloudflare@latest -- producer-worker

Для настройки выберите следующие параметры:

• Для С чего вы хотите начать?, выберите Hello World example.
• Для Какой шаблон вы хотите использовать?, выберите Worker only.
• Для Какой язык вы хотите использовать?, выберите TypeScript.
• Для Хотите использовать git для контроля версий?, выберите Yes.
• Для Хотите развернуть приложение?, выберите No (мы внесём некоторые изменения перед развёртыванием).

Это создаст новый каталог, который будет включать как src/index.ts скрипт Worker и wrangler.jsonc конфигурационный файл. После создания Worker вы создадите очередь для доступа к ней.

Перейдите в только что созданный каталог:

cd producer-worker

Чтобы использовать очереди, создайте хотя бы одну очередь для публикации и получения сообщений.

Чтобы создать очередь, выполните:

npx wrangler queues create <MY-QUEUE-NAME>

Выберите описательное имя, которое отражает типы сообщений, для которых предназначена эта очередь. Понятные имена очередей выглядят так: debug-logs, user-clickstream-data, или password-reset-prod.

Имя очереди должно содержать от 1 до 63 символов. Имя очереди не может содержать специальные символы, кроме дефисов (-), и должно начинаться и заканчиваться буквой или цифрой.

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

Чтобы код внутри Worker мог обращаться к очереди, подключите очередь к Worker, создав привязку. Bindings позволяют вашему Worker обращаться к ресурсам платформы Cloudflare для разработчиков, таким как Queues.

Чтобы создать привязку, откройте только что созданный wrangler.jsonc файл и добавьте в него следующее:

{
	"queues": {
		"producers": [
			{
				"queue": "MY-QUEUE-NAME",
				"binding": "MY_QUEUE"
			}
		]
	}
}

[[queues.producers]]
queue = "MY-QUEUE-NAME"
binding = "MY_QUEUE"

Замените MY-QUEUE-NAME с именем очереди, созданной в шаг 2. Затем замените MY_QUEUE с именем, которое вы хотите присвоить binding. Привязка должна быть допустимым именем переменной JavaScript. Эта переменная используется для обращения к очереди в вашем Worker.

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

1. Принимает запрос, полученный от браузера.
2. Преобразуйте запрос в формат JSON.
3. Запишите запрос напрямую в очередь.

В каталоге проекта Worker откройте src папку и добавьте следующее в ваш index.ts файле:

export default {
  async fetch(request, env, ctx): Promise<Response> {
    const log = {
      url: request.url,
      method: request.method,
      headers: Object.fromEntries(request.headers),
    };
    await env.<MY_QUEUE>.send(log);
    return new Response("Success!");
  },
} satisfies ExportedHandler<Env>;

Замените MY_QUEUE с именем, заданным для привязки в wrangler.jsonc файл.

Также добавьте очередь в Env интерфейс в index.ts.

export interface Env {
   <MY_QUEUE>: Queue;
}

Если эта запись завершится ошибкой, Worker вернёт ошибку (выбросит исключение). Если запись пройдёт успешно, будет возвращено Success с кодом HTTP 200 в качестве кода состояния браузеру.

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

Имея файл Wrangler и index.ts файл настроен, вы готовы опубликовать Worker производителя. Чтобы опубликовать Worker производителя, выполните:

npx wrangler deploy

Вы должны увидеть результат, похожий на приведенный ниже, с *.workers.dev URL по умолчанию.

Uploaded <YOUR-WORKER-NAME> (0.76 sec)
Published <YOUR-WORKER-NAME> (0.29 sec)
  https://<YOUR-WORKER-NAME>.<YOUR-ACCOUNT>.workers.dev

Скопируйте свой *.workers.dev и вставьте его в новую вкладку браузера. Обновите страницу несколько раз, чтобы начать публикацию запросов в очередь. Браузер должен вернуть Success каждый раз после записи запроса в очередь.

Вы создали очередь и Worker-производитель для публикации сообщений в нее. Теперь вы создадите Worker-потребитель, который будет получать опубликованные в очереди сообщения. Без Worker-потребителя сообщения останутся в очереди до истечения срока их хранения, который по умолчанию составляет четыре (4) дня.

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

В этом руководстве вы создадите Worker-потребителя и с его помощью будете журналировать и просматривать сообщения с помощью wrangler tail. Вы создадите Worker-потребитель в том же проекте Worker, где был создан Worker-производитель.

Чтобы создать Worker-потребителя, откройте index.ts файл и добавьте следующее queue обработчик в существующий fetch обработчик:

export default {
  async fetch(request, env, ctx): Promise<Response> {
    const log = {
      url: request.url,
      method: request.method,
      headers: Object.fromEntries(request.headers),
    };
    await env.<MY_QUEUE>.send(log);
    return new Response("Success!");
  },
  async queue(batch, env, ctx): Promise<void> {
    for (const message of batch.messages) {
      console.log("consumed from our queue:", JSON.stringify(message.body));
    }
  },
} satisfies ExportedHandler<Env>;

Замените MY_QUEUE с именем, заданным для привязки в wrangler.jsonc файл.

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

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

При выполнении асинхронных задач внутри обработчика потребителя используйте waitUntil() , чтобы гарантировать обработку ответа функции. Другие асинхронные методы в рамках этого метода не поддерживаются.

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

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

Чтобы подключить очередь к Worker-потребителю, откройте файл Wrangler и добавьте в конец следующее:

{
	"queues": {
		"consumers": [
			{
				"queue": "<MY-QUEUE-NAME>",
				// Required: this should match the name of the queue you created in step 3.
				// If you misspell the name, you will receive an error when attempting to publish your Worker.
				"max_batch_size": 10, // optional: defaults to 10
				"max_batch_timeout": 5 // optional: defaults to 5 seconds
			}
		]
	}
}

[[queues.consumers]]
queue = "<MY-QUEUE-NAME>"
max_batch_size = 10
max_batch_timeout = 5

Замените MY-QUEUE-NAME с очередью, созданной в шаг 2.

В Worker-потребителе вы используете автоматическую пакетную обработку сообщений очереди с помощью max_batch_size и параметр max_batch_timeout . Worker-потребитель будет получать сообщения пакетами по 10 либо каждые 5 секунд, в зависимости от того, что наступит раньше.

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

max_batch_timeout (по умолчанию 5 секунд) помогает сократить время ожидания. Если Worker-отправитель не отправляет в очередь до 10 сообщений, необходимых для вызова Worker-потребителя, Worker-потребитель будет вызываться каждые 5 секунд для получения сообщений, ожидающих в очереди.

Имея файл Wrangler и index.ts файл настроен, опубликуйте Worker потребителя, выполнив команду:

npx wrangler deploy

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

Запустите wrangler tail , чтобы начать ожидание регистрации получаемых сообщений нашим потребителем:

npx wrangler tail

С wrangler tail запущен, откройте URL Worker, который вы открыли в шаг 3.

Вы должны получить Success сообщение в окне браузера.

Если вы получаете Success сообщение, обновите URL несколько раз, чтобы сгенерировать сообщения и отправить их в очередь.

С wrangler tail запущен, ваш Worker-потребитель начнёт регистрировать запросы, создаваемые при обновлении страницы.

Если вы обновили страницу менее 10 раз, появление сообщений может занять несколько секунд, поскольку тайм-аут пакета настроен на 10 секунд. Через 10 секунд сообщения должны появиться в терминале.

Если при обновлении страницы возникают ошибки, проверьте, что имя очереди, созданное в шаг 2 и очередь, указанная в вашем файле Wrangler, совпадают. Убедитесь, что ваш Worker-отправитель возвращает Success и не возвращает ошибку.

Завершив это руководство, вы создали очередь, Worker-производитель, который публикует сообщения в эту очередь, и Worker-потребитель, который получает эти сообщения из неё.

• Подробнее о Cloudflare Workers и приложениях, которые можно создавать на Cloudflare.