← Cloudflare Queues / queues / reference

Jak Queues fungují

Cloudflare Queues je flexibilní systém frontování zpráv, který umožňuje řadit zprávy pro asynchronní zpracování. Frontování zpráv se skvěle hodí k oddělení jednotlivých částí aplikace, například služby pokladny a vyřizování objednávek u e-shopu. Takto oddělené služby se snáze navrhují, nasazují a implementují, takže můžete nasazovat funkce, které vaše zákazníky potěší, aniž byste museli řešit synchronizaci složitých nasazení. Fronty navíc umožňují dávkovat a bufferovat volání navazujících služeb a API.

U Queues je třeba porozumět čtyřem základním konceptům:

1. Queues
2. Producenti
3. Konzumenti
4. Zprávy

Fronta je vyrovnávací paměť nebo seznam, který se automaticky škáluje podle toho, jak jsou do něj zapisovány zprávy, a umožňuje konzumujícímu Workeru odebírat zprávy z téže fronty.

Queues jsou navržené tak, aby byly spolehlivé: jakmile se zápis zprávy do fronty úspěšně dokončí, zpráva by už neměla být nikdy ztracena. Stejně tak se zprávy z fronty neodstraní, dokud spotřebitel zprávu úspěšně zkonzumoval.

Queues nezaručují, že zprávy budou konzumentovi doručeny ve stejném pořadí, v jakém byly publikovány.

Vývojáři mohou vytvořit více front. Vytvoření více front může být užitečné pro:

• Oddělte jednotlivé případy použití a požadavky na zpracování, například frontu pro logování a frontu pro reset hesla.
• Horizontálně škálujte celkovou propustnost (počet zpráv za sekundu) tím, že pro škálování využijete více front.
• Nastavte odlišné strategie dávkování pro každého konzumenta připojeného k frontě.

U většiny aplikací umožňuje jeden produkující Worker na frontu spolu s jedním konzumujícím Workerem, který z této fronty odebírá zprávy, logicky oddělit zpracování jednotlivých front.

Producent je označení pro klienta, který publikuje nebo vytváří zprávy pro frontu. Producent se konfiguruje pomocí binding frontu k Workeru a zápisem zpráv do fronty prostřednictvím volání této vazby.

Pokud bychom například navázali frontu s názvem my-first-queue na vazbu typu MY_FIRST_QUEUE, zprávy lze do fronty zapisovat voláním send() na vazbu:

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

Fronta může mít více Workerů producenta. Můžete mít například více Workerů producenta, kteří na základě příchozích HTTP požadavků od uživatelů zapisují události nebo logy do sdílené fronty. Celkový počet Workerů producenta, kteří mohou zapisovat do jedné fronty, není nijak omezen.

K jednomu Workeru lze navíc připojit více front. Tento Worker pak může na základě vlastní logiky ve svém kódu rozhodnout, do které fronty zapisovat, případně do kolika najednou.

Zprávy publikované do fronty lze publikovat v různých formátech podle toho, jakou interoperabilitu s konzumentem potřebujete. Výchozí typ obsahu je json, což znamená, že jakýkoli objekt, který lze předat JSON.stringify() budou přijaty.

Chcete-li explicitně nastavit typ obsahu nebo zadat alternativní typ obsahu, předejte parametr contentType možnost do send() metodu vaší fronty:

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

Chcete-li při zápisu do fronty přijímat pouze jednoduché řetězce, nastavte { contentType: "text" } místo toho:

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 Dokumentace API popisuje, jak se jednotlivé formáty serializují do fronty.

Queues podporují dva typy konzumentů:

1. A přijímající Worker, který je typu push: Worker se spustí, jakmile má fronta zprávy k doručení.
2. A Konzument HTTP pull, který je typu pull: konzument volá koncový bod fronty přes HTTP, aby přijal a poté potvrdil zprávy.

Fronta může mít nakonfigurován vždy jen jeden typ konzumenta.

Konzument je označení pro klienta, který odebírá nebo zpracování zprávy z fronty. Ve své nejzákladnější podobě je konzument definován vytvořením queue handler ve Workeru:

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

Tohoto konzumenta poté připojíte k frontě pomocí wrangler queues consumer <queue-name> <worker-script-name> nebo definováním [[queues.consumers]] konfigurace ve vašem Konfigurační soubor Wrangler ručně:

{
	"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

Je důležité si uvědomit, že každá fronta může mít jen jednoho aktivního konzumenta. Díky tomu Cloudflare Queues dosahuje doručení alespoň jednou a minimalizuje riziko duplicitních zpráv nad tento rámec.

Stejný consumer můžete použít pro více front najednou. Handler fronty definující váš consumer Worker se spustí pokaždé, když ho vyvolá některá z front, ke kterým je připojen.

• MessageBatch která je předána vašemu queue handler obsahuje queue vlastnost s názvem fronty, ze které byla dávka přečtena.
• Díky tomu můžete napsat méně kódu a zpracovávat zprávy podle názvu jednotlivých front.

Konzument nakonfigurovaný tak, aby zprávy odebíral z více front, by mohl vypadat například takto:

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

Chcete-li odstranit frontu z projektu, spusťte wrangler queues consumer remove <queue-name> <script-name> a poté odeberte požadovanou frontu pod [[queues.consumers]] v souboru Wrangler.

Fronta může mít konzumenta založeného na HTTP, který zprávy z fronty stahuje (pull), místo aby se odesílaly Workeru (push).

Tímto konzumentem může být jakákoli služba komunikující přes HTTP, která se dokáže připojit k internetu. Projděte si návod pro konzumenty typu pull a zjistíte, jak nakonfigurovat spotřebitele na principu vyžádání (pull) pro frontu.

Zpráva je objekt, který do fronty odesíláte a z fronty přijímáte.

Do fronty lze publikovat jakýkoli objekt serializovatelný do JSON. Pro většinu vývojářů to v praxi znamená jednoduché řetězce nebo JSON objekty. Explicitně můžete nastavit typ obsahu při odesílání zprávy.

Samotné zprávy mohou být při doručení konzumentovi seskupovány do dávek. Ve výchozím nastavení se opakování dávky posuzuje podle zásady vše nebo nic. Pokud se nepodaří zpracovat poslední zprávu v dávce, opakuje se celá dávka. Můžete si také zvolit, že výslovně potvrdit zprávy, jakmile jsou úspěšně zpracovány, a/nebo označit jednotlivé zprávy k opakování.