← Cloudflare Queues / queues / konfigurace

Dávkování, opakování a zpoždění

Při konfiguraci přijímající Worker pro frontu, můžete rovněž definovat, jak jsou zprávy při doručování seskupovány do dávek.

Dávkování umožňuje:

1. Snižte celkový počet vyvolání konzumentského Workeru (což může snížit náklady).
2. Umožňuje slučovat zprávy do dávek při zápisu do externího API nebo služby (a tím omezit počet zápisů).
3. Rozložit zátěž v čase, zejména pokud jsou vaše Workery producenta spojeny s činností orientovanou na uživatele.

Existují dva způsoby, jak nakonfigurovat dávkování zpráv. Dávkování se nastavuje při připojování konzumentského Workeru k frontě.

• max_batch_size - maximální velikost dávky doručené konzumentovi (výchozí hodnota 10 zpráv).
• max_batch_timeout - maximum doba, po kterou fronta čeká před doručením dávky konzumentovi (výchozí hodnota 5 sekund)

Například max_batch_size = 30 a max_batch_timeout = 10 znamená, že pokud je do fronty zapsáno 30 zpráv, konzument obdrží dávku 30 zpráv. Pokud však zápis těchto 30 zpráv do fronty trvá déle než 10 sekund, konzument obdrží dávku zpráv obsahující tolik zpráv, kolik jich bylo ve frontě v daném okamžiku (v tomto případě někde mezi 1 a 29).

Při rozhodování, jaká nastavení velikosti a časového limitu zvolit, zvažte latenci (jak dlouho můžete čekat na přijetí zpráv), celkovou velikost dávky (při zápisu do externích systémů) a náklady (méně, ale větší dávky).

Následující nastavení na úrovni dávky lze upravit tak, aby ovlivnila způsob, jakým Queues doručuje dávky vašemu nakonfigurovanému konzumentovi.

Jednotlivé zprávy v dávce můžete potvrzovat explicitním potvrzením každé zprávy během jejího zpracování. Explicitně potvrzené zprávy se znovu nedoručí, i když konzument fronty selže u některé z následujících zpráv nebo se mu nepodaří úspěšně dokončit zpracování dávky.

• Každou zprávu lze při zpracování v dávce potvrdit, čímž se zabrání opakovanému doručení celé dávky v případě, že konzument během zpracování dávky vyvolá chybu.
• Potvrzování jednotlivých zpráv se hodí, pokud voláte externí API, zapisujete zprávy do databáze nebo provádíte jiné neidempotentní akce (měnící stav) nad jednotlivými zprávami.

Chcete-li zprávu explicitně potvrdit jako doručenou, zavolejte metodu ack() metodu na zprávě.

export default {
	async queue(batch, env, ctx) {
		for (const msg of batch.messages) {
			// TODO: do something with the message
			// Explicitly acknowledge the message as delivered
			msg.ack();
		}
	},
};

export default {
	async queue(batch, env, ctx): Promise<void> {
		for (const msg of batch.messages) {
			// TODO: do something with the message
			// Explicitly acknowledge the message as delivered
			msg.ack();
		}
	},
} satisfies ExportedHandler<Env>;

from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        for msg in batch.messages:
            # TODO: do something with the message
            # Explicitly acknowledge the message as delivered
            msg.ack()

Můžete také zavolat retry() k explicitnímu vynucení opětovného doručení zprávy v následující dávce. Tomu se říká „negativní potvrzení“. To se hodí zejména tehdy, když chcete zpracovat zbytek zpráv v dávce bez vyvolání chyby, která by vynutila opětovné doručení celé dávky.

export default {
	async queue(batch, env, ctx) {
		for (const msg of batch.messages) {
			// TODO: do something with the message that fails
			msg.retry();
		}
	},
};

export default {
	async queue(batch, env, ctx): Promise<void> {
		for (const msg of batch.messages) {
			// TODO: do something with the message that fails
			msg.retry();
		}
	},
} satisfies ExportedHandler<Env>;

from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        for msg in batch.messages:
            # TODO: do something with the message that fails
            msg.retry()

Zprávy můžete potvrzovat nebo záporně potvrzovat také na úrovni celé dávky pomocí ackAll() a retryAll(). Volání ackAll() na dávku zpráv (MessageBatch) doručená vašemu Workeru konzumenta se chová stejně jako Worker konzumenta, který úspěšně skončí (nevyhodí chybu).

Mějte na paměti, že volání ack(), retry() a jejich ackAll() / retryAll() ekvivalenty se řídí těmito pravidly priority:

• Pokud zavoláte ack() na zprávu, následná volání ack() nebo retry() jsou tiše ignorovány.
• Pokud zavoláte retry() na zprávu a poté zavolat ack(): ack() se ignoruje. Ve všech případech vítězí první volání metody.
• Pokud zavoláte ack() nebo retry() na jednu zprávu a poté buď/kteroukoli z ackAll() nebo retryAll() na dávku, má přednost volání na jednotlivou zprávu. To znamená, že volání na úrovni dávky se na tuto zprávu (nebo zprávy, pokud bylo provedeno více volání) nevztahuje.

Pokud se zprávu nepodaří doručit, výchozí chování je zopakovat doručení třikrát, než se označí jako neúspěšné. Nastavit můžete max_retries (výchozí hodnota 3) při konfiguraci konzumenta, ve většině případů však doporučujeme ponechat výchozí hodnotu.

Zprávy, které dosáhnou nastaveného maximálního počtu opakování, budou z fronty odstraněny, nebo pokud je nastavena fronta nedoručených zpráv (DLQ) je nakonfigurována, zapíše se místo toho do DLQ.

Pokud se nepodaří doručit jedinou zprávu v dávce, zopakuje se doručení celé dávky, pokud nemáte výslovně potvrzeno zprávu (nebo zprávy) v dané dávce. Pokud je například doručena dávka 10 zpráv, ale doručení 8. zprávy selže, zopakuje se doručení všech 10 zpráv a konzumentovi se znovu doručí celá dávka.

Při publikování zpráv do fronty nebo při označení zprávy nebo dávky k opakování, můžete zprávy na určitou dobu zdržet a jejich zpracování odložit.

Zpoždění zpráv umožňuje odložit úlohy na později nebo reagovat na backpressure při zpracovávání fronty. Pokud například upstream API, které voláte, vrátí HTTP 429: Too Many Requests, můžete zprávy zdržet a zpomalit tak jejich zpracování před dalším pokusem.

Zprávy lze zpozdit až o 24 hodin.

Při odesílání zprávy nebo dávky zpráv do fronty můžete zpoždění nastavit pomocí parametru delaySeconds parametr při odesílání zprávy.

// Delay a singular message by 600 seconds (10 minutes)
await env.YOUR_QUEUE.send(message, { delaySeconds: 600 });

// Delay a batch of messages by 300 seconds (5 minutes)
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 300 });

// Do not delay this message.
// If there is a global delay configured on the queue, ignore it.
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 0 });

// Delay a singular message by 600 seconds (10 minutes)
await env.YOUR_QUEUE.send(message, { delaySeconds: 600 });

// Delay a batch of messages by 300 seconds (5 minutes)
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 300 });

// Do not delay this message.
// If there is a global delay configured on the queue, ignore it.
await env.YOUR_QUEUE.sendBatch(messages, { delaySeconds: 0 });

# Delay a singular message by 600 seconds (10 minutes)
await env.YOUR_QUEUE.send(message, delaySeconds=600)

# Delay a batch of messages by 300 seconds (5 minutes)
await env.YOUR_QUEUE.sendBatch(messages, delaySeconds=300)

# Do not delay this message.
# If there is a global delay configured on the queue, ignore it.
await env.YOUR_QUEUE.sendBatch(messages, delaySeconds=0)

Výchozí globální prodlevu pro celou frontu můžete nastavit také předáním --delivery-delay-secs při vytváření fronty prostřednictvím wrangler CLI:

# Delay all messages by 5 minutes as a default
npx wrangler queues create $QUEUE-NAME --delivery-delay-secs=300

Když zpracování zpráv z fronty, můžete se rozhodnout výslovně označit zprávy k opakování. Zprávy lze opakovat a zdržet jednotlivě nebo jako celou dávku.

Jak zpozdit jednotlivou zprávu v rámci dávky:

export default {
	async queue(batch, env, ctx) {
		for (const msg of batch.messages) {
			// Mark for retry and delay a singular message
			// by 3600 seconds (1 hour)
			msg.retry({ delaySeconds: 3600 });
		}
	},
};

export default {
	async queue(batch, env, ctx): Promise<void> {
		for (const msg of batch.messages) {
			// Mark for retry and delay a singular message
			// by 3600 seconds (1 hour)
			msg.retry({ delaySeconds: 3600 });
		}
	},
} satisfies ExportedHandler<Env>;

from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        for msg in batch.messages:
            # Mark for retry and delay a singular message
            # by 3600 seconds (1 hour)
            msg.retry(delaySeconds=3600)

Jak zpozdit dávku zpráv:

export default {
	async queue(batch, env, ctx) {
		// Mark for retry and delay a batch of messages
		// by 600 seconds (10 minutes)
		batch.retryAll({ delaySeconds: 600 });
	},
};

export default {
	async queue(batch, env, ctx): Promise<void> {
		// Mark for retry and delay a batch of messages
		// by 600 seconds (10 minutes)
		batch.retryAll({ delaySeconds: 600 });
	},
} satisfies ExportedHandler<Env>;

from workers import WorkerEntrypoint

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        # Mark for retry and delay a batch of messages
        # by 600 seconds (10 minutes)
        batch.retryAll(delaySeconds=600)

Výchozí prodlevu opakování můžete nastavit také u zpráv, jejichž doručení se opakuje kvůli implicitnímu selhání nebo při volání retry() explicitně. Nastavuje se na úrovni konzumenta a je podporováno jak u push konzumentů (Worker), tak u pull konzumentů (HTTP).

Zpoždění lze nastavit pomocí wrangler CLI:

# Push-based consumers
# Delay any messages that are retried by 60 seconds (1 minute) by default.
npx wrangler@latest queues consumer worker add $QUEUE-NAME $WORKER_SCRIPT_NAME --retry-delay-secs=60

# Pull-based consumers
# Delay any messages that are retried by 60 seconds (1 minute) by default.
npx wrangler@latest queues consumer http add $QUEUE-NAME --retry-delay-secs=60

Zpoždění lze nastavit také v Konfigurační soubor Wrangler hodnotou delivery_delay nastavení pro producenty (při odesílání) a/nebo retry_delay (při opakování) pro jednotlivého konzumenta:

{
	"queues": {
		"producers": [
			{
				"binding": "<BINDING_NAME>",
				"queue": "<QUEUE-NAME>",
				"delivery_delay": 60 // delay every message delivery by 1 minute
			}
		],
		"consumers": [
			{
				"queue": "my-queue",
				"retry_delay": 300 // delay any retried message by 5 minutes before re-attempting delivery
			}
		]
	}
}

[[queues.producers]]
binding = "<BINDING_NAME>"
queue = "<QUEUE-NAME>"
delivery_delay = 60

[[queues.consumers]]
queue = "my-queue"
retry_delay = 300

Pokud používáte současně wrangler CLI a Konfigurační soubor Wrangler ke změně nastavení souvisejících s frontou nebo spotřebitelem fronty, projeví se poslední provedená změna konfigurace.

Viz Dokumentace REST API pro Queues a zjistíte, jak programově nakonfigurovat zpoždění zpráv a zpoždění opakování.

Zprávy lze standardně zpozdit na úrovni fronty, nebo jednotlivě (případně po dávkách).

• Nastavení zpoždění pro jednotlivé zprávy nebo dávky má přednost před nastavením na úrovni fronty.
• Nastavení delaySeconds: 0 na zprávu při odesílání nebo opakování ignoruje veškerá zpoždění na úrovni fronty a způsobí, že zpráva bude doručena v další dávce.
• Zpráva odeslaná nebo opakovaně odeslaná s delaySeconds: <any positive integer> do fronty s kratším výchozím zpožděním bude i tak respektovat nastavení na úrovni zprávy.

Můžete použít algoritmus zpětného odkladu (backoff), který postupně prodlužuje zpoždění zpráv podle aktuálního počtu pokusů o jejich doručení.

Každá zpráva doručená konzumentovi obsahuje attempts vlastnost, která sleduje počet provedených pokusů o doručení.

Chcete-li například vygenerovat exponenciální prodleva ↗ pro zprávu, můžete si vytvořit pomocnou funkci, která to za vás vypočítá:

function calculateExponentialBackoff(attempts, baseDelaySeconds) {
	return baseDelaySeconds ** attempts;
}

function calculateExponentialBackoff(
	attempts: number,
	baseDelaySeconds: number,
): number {
	return baseDelaySeconds ** attempts;
}

def calculate_exponential_backoff(attempts, base_delay_seconds):
    return base_delay_seconds ** attempts

V konzumentovi poté předáte hodnotu msg.attempts a požadovaný faktor zpoždění jako argument pro delaySeconds při volání retry() na jednotlivou zprávu:

const BASE_DELAY_SECONDS = 30;

export default {
	async queue(batch, env, ctx) {
		for (const msg of batch.messages) {
			// Mark for retry with exponential backoff
			msg.retry({
				delaySeconds: calculateExponentialBackoff(
					msg.attempts,
					BASE_DELAY_SECONDS,
				),
			});
		}
	},
};

const BASE_DELAY_SECONDS = 30;

export default {
	async queue(batch, env, ctx): Promise<void> {
		for (const msg of batch.messages) {
			// Mark for retry with exponential backoff
			msg.retry({
				delaySeconds: calculateExponentialBackoff(
					msg.attempts,
					BASE_DELAY_SECONDS,
				),
			});
		}
	},
} satisfies ExportedHandler<Env>;

from workers import WorkerEntrypoint

BASE_DELAY_SECONDS = 30

class Default(WorkerEntrypoint):
    async def queue(self, batch):
        for msg in batch.messages:
            # Mark for retry and delay a singular message
            # by 3600 seconds (1 hour)
            msg.retry(
                delaySeconds=calculate_exponential_backoff(
                    msg.attempts,
                    BASE_DELAY_SECONDS,
                )
            )

• Zkontrolujte JavaScript API dokumentaci pro Queues.
• Další informace o Jak Queues fungují.
• Seznamte se s dostupné metriky pro vaše fronty, včetně počtu zpráv čekajících ve frontě a zpožděných zpráv.