← Cloudflare Queues / queues / konfigurace

Pull consumers

Konzument založený na pull vám umožňuje načítat zprávy z fronty přes HTTP z libovolného prostředí nebo programovacího jazyka mimo Cloudflare Workers. Tento typ konzumenta se hodí, pokud je rychlost zpracování zpráv omezena nadřazenou infrastrukturou nebo dlouho běžícími úlohami.

To, zda nakonfigurujete konzumenta typu push, nebo typu pull, závisí na tom, jak fronty používáte, a na konfiguraci infrastruktury, která předchází vašemu konzumentovi fronty.

• Počínaje konzument typu push je nejjednodušší způsob, jak začít a konzumovat z fronty. Konzument typu push běží na Workers a ve výchozím nastavení se automaticky škáluje a odebírá zprávy, jak jsou zapisovány do fronty.
• Pull-based consumer použijte v případě, že potřebujete odebírat zprávy z existující infrastruktury mimo Cloudflare Workers, nebo když potřebujete přesně řídit rychlost, jakou se zprávy odebírají. Pull-based consumer musí zprávy z fronty explicitně stáhnout voláním pull (a poté je potvrdit), a to pouze ve chvíli, kdy je na to připravený.

Konzumenta fronty můžete kdykoli odebrat a připojit nového, což vám umožní přejít z konzumenta typu pull na typ push, pokud se změní vaše požadavky.

Chcete-li nakonfigurovat konzumenta založeného na pull a přijímat zprávy z fronty, musíte:

1. Povolte HTTP pull pro frontu.
2. Vytvořte platný autentizační token pro HTTP klienta.
3. Načítání dávek zpráv z fronty.
4. Potvrďte a/nebo znovu odešlete zprávy v rámci dávky.

HTTP pull můžete povolit, nebo frontu změnit z typu push na typ pull pomocí wrangler CLI nebo přes Cloudflare dashboard ↗. Povolení HTTP pull u Konfigurační soubor Wrangler již není podporován.

Konzumenta typu pull můžete povolit u kterékoli existující fronty pomocí wrangler queues consumer http dílčích příkazů a zadáním názvu fronty.

npx wrangler queues consumer http add $QUEUE-NAME

Pokud již máte existujícího konzumenta založeného na push, musíte ho nejprve odstranit. wrangler vrátí chybu, pokud se pokusíte zavolat consumer http add na frontu s existující konfigurací spotřebitele:

wrangler queues consumer worker remove $QUEUE-NAME $SCRIPT_NAME

Konzumenti HTTP Pull vyžadují API token hodnotou com.cloudflare.api.account.queues_read a com.cloudflare.api.account.queues_write oprávnění.

Obě čtou a zápis jsou nutná, protože konzument typu pull potřebuje zapisovat do stavu fronty, aby potvrdil přijetí zpráv. Odběr zpráv frontu mění.

API tokeny se předávají jako Bearer tokeny v Authorization hlavičce HTTP požadavku ve formátu Authorization: Bearer $YOUR_TOKEN_HERE. Následující příklad ukazuje, jak předat token API pomocí curl HTTP klient:

curl "https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull" \
--header "Authorization: Bearer ${QUEUES_TOKEN}" \
--header "Content-Type: application/json" \
--data '{ "visibility_timeout_ms": 10000, "batch_size": 2 }'

K jedné frontě můžete autentizovat a spustit více souběžných konzumentů typu pull.

Chcete-li vytvořit token API:

1. Přihlaste se do Cloudflare dashboard ↗.
2. Přejděte na My Profile > API Tokens ↗.
3. Vyberte Create Token.
4. Přejděte na konec stránky a vyberte Create Custom Token.
5. Zadejte název tokenu, například queue-pull-token.
6. V části Oprávnění sekci vyberte Účet a poté Queues. Ujistěte se, že jste vybrali Úprava (čtení a zápis).
7. (Volitelné) Vyberte Všechny účty (výchozí) nebo konkrétní účet, na který chcete token omezit.
8. Vyberte Pokračovat k souhrnu a poté Vytvoření tokenu.

Token si poznamenejte, zobrazí se pouze jednou.

Chcete-li stáhnout zprávu, odešlete požadavek HTTP POST na REST API pro Queues s tělem ve formátu JSON, které volitelně určuje visibility_timeout a batch_size, nebo prázdný objekt JSON ({}):

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull with the timeout & batch size
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// Optional - you can provide an empty object '{}' and the defaults will apply.
		body: JSON.stringify({ visibility_timeout_ms: 6000, batch_size: 50 }),
	},
);

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull with the timeout & batch size
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// Optional - you can provide an empty object '{}' and the defaults will apply.
		body: JSON.stringify({ visibility_timeout_ms: 6000, batch_size: 50 }),
	},
);

import json
from workers import fetch

# POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/pull with the timeout & batch size

resp = await fetch(
	f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/queues/{QUEUE_ID}/messages/pull",
	method="POST",
	headers={
		"content-type": "application/json",
		"authorization": f"Bearer {QUEUES_API_TOKEN}",
	}, # Optional - you can provide an empty object '{}' and the defaults will apply.
	body=json.dumps({"visibility_timeout_ms": 6000, "batch_size": 50}),
)

Tímto se vrátí pole zpráv (až do zadaného batch_size) v následujícím formátu:

{
	"success": true,
	"errors": [],
	"messages": [],
	"result": {
		"message_backlog_count": 10,
		"messages": [
			{
				"body": "hello",
				"id": "1ad27d24c83de78953da635dc2ea208f",
				"timestamp_ms": 1689615013586,
				"attempts": 2,
				"metadata": {
					"CF-sourceMessageSource": "dash",
					"CF-Content-Type": "json"
				},
				"lease_id": "eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0..NXmbr8h6tnKLsxJ_AuexHQ.cDt8oBb_XTSoKUkVKRD_Jshz3PFXGIyu7H1psTO5UwI.smxSvQ8Ue3-ymfkV6cHp5Va7cyUFPIHuxFJA07i17sc"
			},
			{
				"body": "world",
				"id": "95494c37bb89ba8987af80b5966b71a7",
				"timestamp_ms": 1689615013586,
				"attempts": 2,
				"metadata": {
					"CF-sourceMessageSource": "dash",
					"CF-Content-Type": "json"
				},
				"lease_id": "eyJhbGciOiJkaXIiLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIn0..QXPgHfzETsxYQ1Vd-H0hNA.mFALS3lyouNtgJmGSkTzEo_imlur95EkSiH7fIRIn2U.PlwBk14CY_EWtzYB-_5CR1k30bGuPFPUx1Nk5WIipFU"
			}
		]
	}
}

Pull consumers fungují na principu takzvaného "short pollingu": pokud jsou k dispozici zprávy k doručení, Queues vrátí odpověď okamžitě, a to až do počtu zpráv odpovídajícího nastavenému batch_size. Pokud nejsou k dispozici žádné zprávy k doručení, Queues vrátí prázdnou odpověď. Queues neudržuje otevřené spojení (často označované jako „long polling“), pokud nejsou k dispozici žádné zprávy k doručení.

Každý objekt zprávy má pět polí:

1. body - může být zakódováno pomocí base64 v závislosti na typ obsahu, se kterým byla zpráva publikována.
2. id - jedinečný, pouze pro čtení a dočasný identifikátor zprávy.
3. timestamp_ms - kdy byla zpráva publikována do fronty v milisekundách od Unixová epocha ↗. Díky tomu můžete určit stáří zprávy odečtením od aktuálního časového razítka.
4. attempts - kolikrát se zpráva pokusila doručit celá. Jakmile tato hodnota dosáhne max_retries, zpráva nebude znovu doručena a bude z fronty trvale odstraněna.
5. lease_id - zakódované ID zápůjčky zprávy. lease_id slouží k explicitnímu potvrzení nebo opakování zprávy.

lease_id umožňuje vašemu pull konzumentovi explicitně potvrdit některé, žádné nebo všechny zprávy v dávce, nebo je označit k opakování. Pokud konzument zprávy nepotvrdí ani neoznačí k opakování, budou označeny k opětovnému doručení, jakmile visibility_timeout je dosaženo. lease_id po dosažení tohoto časového limitu již není platný.

Nakonfigurovat můžete obojí batch_size a visibility_timeout při stahování zpráv z fronty:

• batch_size (výchozí hodnota 5, maximum 100) určuje, kolik zpráv se konzumentovi vrátí při každém pull požadavku.
• visibility_timeout (výchozí hodnota 30 sekund, maximum 12 hodin) určuje, jak dlouho má konzument na explicitní potvrzení zpráv doručených v dávce na základě jejich lease_id. Jakmile tento časový limit vyprší, zprávy se považují za nepotvrzené a zařadí se znovu do fronty k opakovanému doručení.

Ze stejné fronty může souběžně načítat více HTTP klientů: každý z nich obdrží jedinečnou dávku zpráv a podrží si na tyto zprávy „lease" až do visibility_timeout vyprší, nebo dokud nejsou tyto zprávy označeny k opakování.

Zprávy označené k opakování se vrátí zpět do fronty a mohou být doručeny libovolnému konzumentovi. Zprávy jsou ne vázané na konkrétního spotřebitele, protože spotřebitelé nemají identitu a aby pomalý nebo zaseknutý spotřebitel nezdržoval zpracování zpráv ve frontě.

Více konzumentů se hodí v případech, kdy máte více upstream prostředků (například GPU infrastrukturu), kde chcete automaticky škálovat na základě backlog fronty, případně náklady.

Zprávy, které konzument stáhne, je třeba buď potvrdit, nebo označit k opakování.

Chcete-li potvrdit přijetí zpráv nebo je označit k opakovanému doručení, odešlete požadavek HTTP POST požadavek na /ack endpoint vaší fronty podle REST API pro Queues zadáním pole lease_id objektů pro potvrzení a/nebo opakování:

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack with the lease_ids
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// If you have no messages to retry, you can specify an empty array - retries: []
		body: JSON.stringify({
			acks: [
				{ lease_id: "lease_id1" },
				{ lease_id: "lease_id2" },
				{ lease_id: "etc" },
			],
			retries: [{ lease_id: "lease_id4" }],
		}),
	},
);

// POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack with the lease_ids
let resp = await fetch(
	`https://api.cloudflare.com/client/v4/accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack`,
	{
		method: "POST",
		headers: {
			"content-type": "application/json",
			authorization: `Bearer ${QUEUES_API_TOKEN}`,
		},
		// If you have no messages to retry, you can specify an empty array - retries: []
		body: JSON.stringify({
			acks: [
				{ lease_id: "lease_id1" },
				{ lease_id: "lease_id2" },
				{ lease_id: "etc" },
			],
			retries: [{ lease_id: "lease_id4" }],
		}),
	},
);

import json
from workers import fetch

# POST /accounts/${CF_ACCOUNT_ID}/queues/${QUEUE_ID}/messages/ack with the lease_ids

resp = await fetch(
	f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}/queues/{QUEUE_ID}/messages/ack",
	method="POST",
	headers={
		"content-type": "application/json",
		"authorization": f"Bearer {QUEUES_API_TOKEN}",
	}, # If you have no messages to retry, you can specify an empty array - retries: []
	body=json.dumps({
		"acks": [
			{"lease_id": "lease_id1"},
			{"lease_id": "lease_id2"},
			{"lease_id": "etc"},
		],
		"retries": [{"lease_id": "lease_id4"}],
	}),
)

Při označování zprávy k opakování můžete volitelně určit počet sekund, o které se má doručení zpozdit, a to zadáním { lease_id: string, delay_seconds: number } objekt v retries pole:

{
	"acks": [
		{ "lease_id": "lease_id1" },
		{ "lease_id": "lease_id2" },
		{ "lease_id": "lease_id3" }
	],
	"retries": [{ "lease_id": "lease_id4", "delay_seconds": 600 }]
}

Kromě toho:

• Měli byste uvádět každý lease_id v požadavku na /ack endpoint, pokud tyto zprávy zpracováváte ve svém konzumentovi. Pokud zprávu nepotvrdíte, bude označena k opětovnému doručení (vrácena zpět do fronty).
• Zprávy můžete volitelně označit k opakovanému doručení, například pokud při jejich zpracování dojde k chybě nebo je přetížený navazující systém. Explicitní označení zprávy k opakování ji ihned vrátí zpět do fronty místo čekání na (potenciálně dlouhý) visibility_timeout dosaženo.
• Můžete provést více volání /ack endpoint postupně při zpracování dávky zpráv, doporučujeme však potvrzení seskupovat, aby se snížil počet potřebných volání API.

Queues je k lease ID benevolentní: pokud consumer potvrdí zprávu podle jejího lease ID po je dosažen limit viditelnosti, Queues potvrzení přesto přijme. Pokud byla zpráva v mezičase doručena jinému spotřebiteli, bude ji rovněž moci potvrdit bez chyby.

Při publikování do fronty, která má externího konzumenta, mějte na paměti, že některé typy obsahu mohou být zakódovány tak, aby je bylo možné bezpečně serializovat do objektu JSON.

Pro obě json a bytes typů obsahu to znamená, že budou zakódovány pomocí base64 (RFC 4648 ↗). text typu bude odeslán jako čistý řetězec kódovaný v UTF-8.

Váš konzument bude muset dekódovat json a bytes typů před prací s daty.

• Zkontrolujte Dokumentace REST API a schéma pro Queues.
• Další informace o jak provádět volání API do Cloudflare API.
• Seznamte se s jaké limity platí při odběru i zápisu do fronty.