INTEGRITY Dokumentace

← Cloudflare Queues / queues / konfigurace

Rozhraní JavaScript API

Cloudflare Queues je integrováno s Cloudflare Workers. K odesílání a přijímání zpráv musíte použít Worker.

Worker, který umí odesílat zprávy do fronty, se nazývá Worker producenta, zatímco Worker, který umí zprávy z fronty přijímat, se nazývá Worker konzumenta. Stejný Worker může být zároveň producentem i konzumentem, pokud si to přejete.

V budoucnu plánujeme podporu dalších API, například HTTP koncových bodů pro odesílání či příjem zpráv. Chcete-li nahlásit chybu nebo požádat o novou funkci, přejděte na Cloudflare Community Forums ↗. Chcete-li nám poskytnout zpětnou vazbu, přejděte na #queues ↗ kanál Discord.

Producer

Tato rozhraní API umožňují producentskému Workeru odesílat zprávy do fronty.

Příklad odeslání jedné zprávy do fronty:

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!")

Rozhraní Queues API také umožňuje zapisovat více zpráv najednou:

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

Vazba, která producentovi umožňuje odesílat zprávy do fronty.

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

MessageSendRequest

Obalový typ používaný pro odesílání dávek zpráv.

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

QueueSendOptions

Volitelná konfigurace použitá při odesílání zprávy do fronty.

QueueSendBatchOptions

Volitelná konfigurace použitá při odesílání dávky zpráv do fronty.

QueuesContentType

Union typ obsahující platné typy obsahu zprávy.

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

Pokud zadáte neplatný typ obsahu nebo pokud zadaný typ obsahu neodpovídá skutečnému typu obsahu zprávy, operace odeslání selže s chybou.

QueueSendResult

Výsledek úspěšné operace odeslání.

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

QueueMetrics

Metriky fronty v reálném čase.

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

Konzument

Tato rozhraní API umožňují konzumentskému Workeru odebírat zprávy z fronty.

Chcete-li definovat konzumentský Worker, přidejte queue() funkci do výchozího exportu Workeru. To mu umožní přijímat zprávy z fronty.

Ve výchozím nastavení se všechny zprávy v dávce potvrdí, jakmile jsou splněny všechny následující podmínky:

  1. queue() funkce vrátila hodnotu.
  2. Pokud queue() funkce vrátila promise, byl tento promise vyřešen.
  3. Veškeré promises předané do waitUntil() byly vyřešeny.

Pokud queue() funkce vyvolá výjimku, nebo pokud jí vrácený promise či kterýkoli z promise předaných do waitUntil() byly odmítnuty, celá dávka se považuje za neúspěšnou a bude opakovaně zpracována podle nastavení opakování daného konzumenta.

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 a ctx pole jsou jako popsáno v dokumentaci Workers.

Typy zpráv v TypeScriptu

Zprávy fronty můžete typovat pomocí Queue<T> na producentovi a ExportedHandler<Env, T> na spotřebiteli.

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>;

Pro primitivní zprávy použijte Queue<number> nebo satisfies ExportedHandler<Env, number>. Pokud typ nezadáte, message.body je unknown.

Consumer fronty lze alternativně napsat i pomocí (zastaralé) syntaxe service worker:

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

V syntaxi service workeru, event poskytuje stejná pole a metody jako MessageBatch, jak je definováno níže, kromě waitUntil() ↗.

MessageBatch

Dávka zpráv odeslaná Workeru konzumenta.

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

Message

Zpráva odeslaná Workeru konzumenta.

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

QueueRetryOptions

Volitelná konfigurace pro označení zprávy nebo dávky zpráv k opakování.

interface QueueRetryOptions {
  delaySeconds?: number;
}