INTEGRITY Документация

← Cloudflare Queues / queues / configuration

JavaScript API

Cloudflare Queues интегрирован с Cloudflare Workers. Для отправки и получения сообщений необходимо использовать Worker.

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

В будущем мы планируем поддержать и другие API, например HTTP-эндпоинты для отправки и получения сообщений. Чтобы сообщить об ошибке или предложить новую функцию, перейдите в Cloudflare Community Forums ↗. Чтобы оставить отзыв, перейдите в #queues ↗ канал Discord.

Производитель

Эти API позволяют Worker-производителю отправлять сообщения в очередь.

Пример записи одного сообщения в очередь:

index.js
export default {
	async fetch(req, env, ctx) {
		await env.MY_QUEUE.send({
			url: req.url,
			method: req.method,
			headers: Object.fromEntries(req.headers),
		});
		return new Response("Sent!");
	},
};
index.ts
interface Env {
  readonly MY_QUEUE: Queue;
}

export default {
  async fetch(req, env, ctx): Promise<Response> {
    await env.MY_QUEUE.send({
      url: req.url,
      method: req.method,
      headers: Object.fromEntries(req.headers),
    });
    return new Response("Sent!");
  },
} satisfies ExportedHandler<Env>;
from workers import Response, WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        await self.env.MY_QUEUE.send({
            "url": request.url,
            "method": request.method,
            "headers": dict(request.headers),
        })
        return Response("Sent!")

API Queues также поддерживает запись нескольких сообщений одновременно:

index.js
const sendResultsToQueue = async (results, env) => {
	const batch = results.map((value) => ({
		body: value,
	}));
	await env.MY_QUEUE.sendBatch(batch);
};
index.ts
const sendResultsToQueue = async (results: Array<unknown>, env: Env) => {
	const batch: MessageSendRequest[] = results.map((value) => ({
		body: value,
	}));
	await env.MY_QUEUE.sendBatch(batch);
};
async def send_results_to_queue(results, env):
    batch = [
        {"body": value}
        for value in results
    ]
    await env.MY_QUEUE.sendBatch(batch)

Queue

Привязка, которая позволяет производителю отправлять сообщения в очередь.

interface Queue<Body = unknown> {
  send(body: Body, options?: QueueSendOptions): Promise<QueueSendResult>;
  sendBatch(messages: Iterable<MessageSendRequest<Body>>, options?: QueueSendBatchOptions): Promise<QueueSendResult>;
  metrics(): Promise<QueueMetrics>;
}

MessageSendRequest

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

interface MessageSendRequest<Body = unknown> {
  body: Body;
  contentType?: QueueContentType;
  delaySeconds?: number;
}

QueueSendOptions

Необязательные параметры, применяемые при отправке сообщения в очередь.

QueueSendBatchOptions

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

QueuesContentType

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

// Default: json
type QueuesContentType = "text" | "bytes" | "json" | "v8";

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

QueueSendResult

Результат успешной операции отправки.

interface QueueSendResult {
	metadata: {
		metrics: QueueMetrics;
	};
}

QueueMetrics

Метрики очереди в реальном времени.

interface QueueMetrics {
	backlogCount: number;
	backlogBytes: number;
	oldestMessageTimestamp: number;
}

Потребитель

Эти API позволяют Worker-потребителю получать сообщения из очереди.

Чтобы определить Worker-потребителя, добавьте queue() функцию в экспорт по умолчанию Worker. Это позволит ему получать сообщения из очереди.

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

  1. queue() функция вернёт управление.
  2. Если queue() функция вернула promise, этот promise будет разрешён.
  3. Любые промисы, переданные в waitUntil() будут разрешены.

Если queue() функция выбросит исключение, либо пока возвращённый ею promise или любой из promise, переданных в waitUntil() были отклонены, весь пакет считается неуспешным и будет повторно обработан в соответствии с настройками повторных попыток потребителя.

index.js
export default {
	async queue(batch, env, ctx) {
		for (const message of batch.messages) {
			console.log("Received", message.body);
		}
	},
};
index.ts
interface Env {
  // Add your bindings here
}

export default {
  async queue(batch, env, ctx): Promise<void> {
    for (const message of batch.messages) {
      console.log("Received", message.body);
    }
  },
} satisfies ExportedHandler<Env>;
from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        for message in batch.messages:
            print("Received", message)

env и ctx поля указаны как описано в документации Workers.

Типы сообщений TypeScript

Вы можете типизировать сообщения очереди с помощью Queue<T> для производителя и ExportedHandler<Env, T> для потребителя.

type MyMessage = {
  id: string;
};

interface Env {
  MY_QUEUE: Queue<MyMessage>;
}

export default {
  async queue(batch) {
    for (const message of batch.messages) {
      console.log(message.body.id);
    }
  },
} satisfies ExportedHandler<Env, MyMessage>;

Для примитивных сообщений используйте Queue<number> или satisfies ExportedHandler<Env, number>. Если вы не укажете тип, message.body это unknown.

Либо, как альтернативу, потребителя очереди можно написать, используя устаревший синтаксис service worker:

addEventListener('queue', (event) => {
	event.waitUntil(handleMessages(event));
});

В синтаксисе service worker event предоставляет те же поля и методы, что и MessageBatch, как описано ниже, в дополнение к waitUntil() ↗.

MessageBatch

Пакет сообщений, отправляемых Worker-потребителю.

interface MessageBatch<Body = unknown> {
  readonly queue: string;
  readonly messages: readonly Message<Body>[];
  ackAll(): void;
  retryAll(options?: QueueRetryOptions): void;
}

Message

Сообщение, отправляемое Worker-потребителю.

interface Message<Body = unknown> {
  readonly id: string;
  readonly timestamp: Date;
  readonly body: Body;
	readonly attempts: number;
  ack(): void;
  retry(options?: QueueRetryOptions): void;
}

QueueRetryOptions

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

interface QueueRetryOptions {
  delaySeconds?: number;
}