← Cloudflare Queues / queues / reference

Как работает Queues

Cloudflare Queues представляет собой гибкую очередь сообщений, которая позволяет ставить сообщения в очередь для асинхронной обработки. Очереди сообщений хорошо подходят для разделения компонентов приложения, например сервисов оформления и выполнения заказов в интернет-магазине. Разделённые сервисы проще анализировать, развёртывать и реализовывать, что позволяет выпускать функции, которые понравятся вашим клиентам, не беспокоясь о синхронизации сложных развёртываний. Очереди также позволяют группировать вызовы к нижестоящим сервисам и API в пакеты и буферизировать их.

В Queues есть четыре основных понятия, которые важно понимать:

1. Queues
2. Производители
3. Потребители
4. Сообщения

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

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

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

Разработчики могут создавать несколько очередей. Это может быть полезно, чтобы:

• Разделяйте разные сценарии использования и требования к обработке: например, очередь логирования и очередь сброса пароля.
• Горизонтально масштабируйте общую пропускную способность (количество сообщений в секунду), используя несколько очередей.
• Настройте разные стратегии пакетной обработки для каждого потребителя, подключённого к очереди.

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

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

Например, если привязать очередь с именем my-first-queue к привязке MY_FIRST_QUEUE, сообщения можно записывать в очередь с помощью вызова send() для привязки:

interface Env {
  readonly MY_FIRST_QUEUE: Queue;
}

export default {
  async fetch(req, env, ctx): Promise<Response> {
    const message = {
      url: req.url,
      method: req.method,
      headers: Object.fromEntries(req.headers),
    };

    await env.MY_FIRST_QUEUE.send(message); // This will throw an exception if the send fails for any reason
    return new Response("Sent!");
  },
} satisfies ExportedHandler<Env>;

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

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

Сообщения, опубликованные в очередь, могут иметь разные форматы в зависимости от того, какая совместимость требуется с вашим потребителем. Тип содержимого по умолчанию: json, это означает, что любой объект, который можно передать в JSON.stringify() будет принято.

Чтобы явно задать тип содержимого или указать альтернативный тип содержимого, передайте contentType параметр в send() метод вашей очереди:

interface Env {
  readonly MY_FIRST_QUEUE: Queue;
}

export default {
  async fetch(req, env, ctx): Promise<Response> {
    const message = {
      url: req.url,
      method: req.method,
      headers: Object.fromEntries(req.headers),
    };
    try {
      await env.MY_FIRST_QUEUE.send(message, { contentType: "json" }); // "json" is the default
      return new Response("Sent!");
    } catch (e) {
      // Catch cases where send fails, including due to a mismatched content type
      const msg = e instanceof Error ? e.message : "Unknown error";
      return Response.json({ error: msg }, { status: 500 });
    }
  },
} satisfies ExportedHandler<Env>;

Чтобы очередь принимала при записи только простые строки, задайте { contentType: "text" } вместо этого:

interface Env {
  readonly MY_FIRST_QUEUE: Queue;
}

export default {
  async fetch(req, env, ctx): Promise<Response> {
    try {
      // This will throw an exception (error) if you pass a non-string to the queue,
      // such as a native JavaScript object or ArrayBuffer.
      await env.MY_FIRST_QUEUE.send("hello there", { contentType: "text" }); // explicitly set 'text'
      return new Response("Sent!");
    } catch (e) {
      const msg = e instanceof Error ? e.message : "Unknown error";
      return Response.json({ error: msg }, { status: 500 });
    }
  },
} satisfies ExportedHandler<Env>;

QueuesContentType В документации API описано, как каждый формат сериализуется в очередь.

Queues поддерживает два типа потребителей:

1. A Worker-получатель, который основан на push: Worker вызывается, когда в очереди появляются сообщения для доставки.
2. A Потребитель HTTP Pull, который основан на pull: потребитель обращается к конечной точке очереди по HTTP, чтобы получать сообщения и затем подтверждать их обработку.

Для очереди можно настроить только один тип потребителя.

Потребителем называется клиент, который подписывается на очередь или потребление сообщения из очереди. В самом простом виде потребитель определяется путём создания queue обработчик в Worker:

interface Env {
  // Add your bindings here, e.g. KV namespaces, R2 buckets, D1 databases
}

export default {
  async queue(batch, env, ctx): Promise<void> {
    // Do something with messages in the batch
    // i.e. write to R2 storage, D1 database, or POST to an external API
    for (const msg of batch.messages) {
      // Process each message
      console.log(msg.body);
    }
  },
} satisfies ExportedHandler<Env>;

Затем подключите этого потребителя к очереди с помощью wrangler queues consumer <queue-name> <worker-script-name> либо путём определения [[queues.consumers]] конфигурацию в вашем конфигурационный файл Wrangler вручную:

{
	"queues": {
		"consumers": [
			{
				"queue": "<your-queue-name>",
				"max_batch_size": 100, // optional
				"max_batch_timeout": 30 // optional
			}
		]
	}
}

[[queues.consumers]]
queue = "<your-queue-name>"
max_batch_size = 100
max_batch_timeout = 30

Важно, что у каждой очереди может быть только один активный потребитель. Это позволяет Cloudflare Queues обеспечивать доставку at least once и сводить к минимуму риск дублирования сообщений сверх этого.

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

• MessageBatch , которая передаётся в ваш queue обработчик включает queue свойство с именем очереди, из которой был прочитан пакет.
• Это уменьшает объём кода, который нужно написать, и позволяет обрабатывать сообщения в зависимости от имени очереди.

Например, конфигурация потребителя, обрабатывающего сообщения из нескольких очередей, будет выглядеть следующим образом:

interface Env {
  // Add your bindings here
}

export default {
  async queue(batch, env, ctx): Promise<void> {
    // MessageBatch has a `queue` property we can switch on
    switch (batch.queue) {
      case "log-queue":
        // Write the batch to R2
        break;
      case "debug-queue":
        // Write the message to the console or to another queue
        break;
      case "email-reset":
        // Trigger a password reset email via an external API
        break;
      default:
        // Handle messages we haven't mentioned explicitly (write a log, push to a DLQ)
        break;
    }
  },
} satisfies ExportedHandler<Env>;

Чтобы удалить очередь из проекта, выполните wrangler queues consumer remove <queue-name> <script-name> а затем удалите нужную очередь ниже [[queues.consumers]] в файле Wrangler.

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

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

Сообщением называется объект, который вы записываете в очередь и считываете из неё.

В очередь можно публиковать любой объект, сериализуемый в JSON. Для большинства разработчиков это означает простые строки или JSON-объекты. Вы можете явно задать тип содержимого при отправке сообщения.

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