プルコンシューマー

プルベースのコンシューマーを使うと、Cloudflare Workers 以外の任意の環境やプログラミング言語から、HTTP 経由でキューをプルできます。上流インフラや長時間実行タスクによってメッセージの消費速度が制限される場合に、プルベースのコンシューマーが役立ちます。

プッシュコンシューマーとプルコンシューマーの選び方

プッシュベースのコンシューマーにするか、プルベースのコンシューマーにするかは、キューの使い方と、キューコンシューマーより上流のインフラ構成によって決まります。

キューの消費を始めるいちばん簡単な方法は、プッシュベースのコンシューマーです。 プッシュベースのコンシューマーは Workers 上で動き、デフォルトではキューへの書き込みに合わせて自動でスケールし、メッセージを消費します。
既存のインフラが Cloudflare Workers の外にあり、そこからメッセージを消費する必要がある場合や、消費速度を慎重に制御したい場合は、プルベースのコンシューマーを使います。プルベースのコンシューマーは、準備ができたときだけ、キューから明示的にプル（その後に確認応答）します。

キューのコンシューマーはいつでも削除して付け直せます。要件が変わったら、プルベースからプッシュベースへ切り替えられます。

プルベースのコンシューマーを設定し、キューからメッセージを受け取るには、次が必要です。

キューで HTTP プルを有効にします。
HTTP クライアント用の有効な認証トークンを作成します。
キューからメッセージバッチをプルします。
バッチ内のメッセージを確認応答するか、再試行します。

1. HTTP プルを有効にする

wrangler CLI または Cloudflare ダッシュボード ↗ で、HTTP プルを有効にしたり、キューをプッシュベースからプルベースへ変更したりできます。Wrangler 設定ファイルからの HTTP プルの有効化は、現在サポートされていません。

wrangler CLI

既存のキューにプルベースのコンシューマーを付けるには、wrangler queues consumer http サブコマンドにキュー名を渡します。

npx wrangler queues consumer http add $QUEUE-NAME

既存のプッシュベースのコンシューマーがある場合は、先に削除する必要があります。既存のコンシューマー設定があるキューで consumer http add を呼ぶと、wrangler はエラーを返します。

wrangler queues consumer worker remove $QUEUE-NAME $SCRIPT_NAME

2. コンシューマーの認証

HTTP プルコンシューマーには、com.cloudflare.api.account.queues_read と com.cloudflare.api.account.queues_write の権限を持つ API トークンが必要です。

読み取りと書き込みの両方が必要です。プルベースのコンシューマーは、受け取ったメッセージを確認応答するために、キューの状態へ書き込む必要があるためです。メッセージの消費はキューを変更します。

API トークンは、HTTP リクエストの Authorization ヘッダーで Bearer トークンとして渡します。形式は Authorization: Bearer $YOUR_TOKEN_HERE です。次の例は、curl HTTP クライアントで API トークンを渡す方法です。

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 }'

1 つのキューに対して、複数のプルベースコンシューマーを同時に認証して実行できます。

API トークンを作成する

API トークンを作成する手順は次のとおりです。

Cloudflare ダッシュボード ↗ にログインします。
My Profile > API Tokens ↗ を開きます。
Create Token を選びます。
ページ下部までスクロールし、Create Custom Token を選びます。
トークンに名前を付けます。例: queue-pull-token。
Permissions で Account を選び、次に Queues を選びます。Edit（読み取り + 書き込み）が選ばれていることを確認します。
（任意）トークンの範囲を All accounts（デフォルト）にするか、特定のアカウントにするかを選びます。
Continue to summary を選び、Create token を選びます。

トークンは一度しか表示されないため、控えておいてください。

3. メッセージをプルする

メッセージをプルするには、Queues REST API へ HTTP POST リクエストを送ります。JSON エンコードした本文で、任意の visibility_timeout と batch_size を指定するか、空の JSON オブジェクト（{}）を送ります。

index.jsjs

// 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 }),
	},
);

index.tsts

// 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}),
)

次の形式で、メッセージの配列（指定した batch_size まで）が返されます。

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

プルコンシューマーは「ショートポーリング」です。配信できるメッセージがあれば、Queues は設定した batch_size までのメッセージをすぐに返します。配信するメッセージがなければ、Queues は空のレスポンスを返します。配信するメッセージがないときに接続を開いたままにする（いわゆる「ロングポーリング」）ことはしません。

各メッセージオブジェクトには 5 つのフィールドがあります。

body - メッセージの公開時の content-type によっては、Base64 エンコードされていることがあります。
id - メッセージの一意で読み取り専用の一時的な識別子です。
timestamp_ms - メッセージがキューに公開された時刻です。Unix epoch ↗ からのミリ秒です。現在のタイムスタンプから引くと、メッセージの経過時間を求められます。
attempts - メッセージの配信が完了まで試行された回数です。max_retries に達すると、メッセージは再配信されず、キューから完全に削除されます。
lease_id - メッセージのエンコードされたリース ID です。lease_id は、メッセージを明示的に確認応答または再試行するために使います。

lease_id を使うと、プルコンシューマーはバッチ内の一部、なし、またはすべてのメッセージを明示的に確認応答したり、再試行に回したりできます。コンシューマーが確認応答も再試行指定もしない場合、visibility_timeout に達すると再配信対象になります。このタイムアウトに達すると、lease_id は無効になります。

キューからプルするとき、batch_size と visibility_timeout の両方を設定できます。

batch_size（デフォルトは 5、最大は 100） - 1 回のプルでコンシューマーに返すメッセージ数です。
visibility_timeout（デフォルトは 30 秒、最大は 12 時間） - バッチで配信されたメッセージを、lease_id に基づいて明示的に確認応答するまでの時間です。このタイムアウトが切れると、メッセージは未確認とみなされ、再配信待ちになります。

同時実行コンシューマー

同じキューから、複数の HTTP クライアントが同時にプルできます。各クライアントは一意のメッセージバッチを受け取り、visibility_timeout が切れるか、再試行に回されるまで、それらのメッセージの「リース」を保持します。

再試行に回されたメッセージはキューに戻され、どのコンシューマーにも配信されます。メッセージは特定のコンシューマーに紐づきません。コンシューマーに識別子はなく、遅い、または停止したコンシューマーがキューの処理を滞らせないようにするためです。

複数コンシューマーは、上流リソースが複数ある場合（例: GPU インフラ）、キューのバックログに応じてオートスケールしたい場合、コストを抑えたい場合などに役立ちます。

4. メッセージを確認応答する

コンシューマーがプルしたメッセージは、確認応答するか、再試行に回す必要があります。

確認応答や再試行指定をするには、Queues REST API に従い、キューの /ack エンドポイントへ HTTP POST リクエストを送ります。確認応答または再試行する lease_id オブジェクトの配列を渡します。

index.jsjs

// 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" }],
		}),
	},
);

index.tsts

// 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"}],
	}),
)

再試行に回すときの遅延秒数は、retries 配列に { lease_id: string, delay_seconds: number } オブジェクトを渡して任意で指定できます。

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

加えて、次の点に注意してください。

コンシューマーで処理しているメッセージについては、すべての lease_id を /ack エンドポイントへのリクエストに含めてください。確認応答しないメッセージは、再配信対象（キューへ戻される）になります。
メッセージを再試行に回すこともできます。たとえば、処理中にエラーが起きた場合や、上流リソースに負荷がある場合です。再試行を明示すると、（長くなり得る）visibility_timeout を待たずに、すぐにキューへ戻します。
バッチの処理が進むにつれて /ack を複数回呼べますが、必要な API 呼び出しを減らすため、確認応答はまとめることを推奨します。

Queues はリース ID に対して寛容です。ビジビリティタイムアウト到達後にコンシューマーがリース ID で確認応答しても、Queues はその確認応答を受け付けます。その間に別のコンシューマーへ配信されていた場合も、そのコンシューマーはエラーなく確認応答できます。

コンテンツタイプ

外部コンシューマーがあるキューへ公開する場合、一部のコンテンツタイプは、JSON オブジェクト内で安全にシリアライズできる形でエンコードされることがあります。

json と bytes のコンテンツタイプでは、RFC 4648 ↗ の Base64 エンコードになります。text タイプは、プレーンな UTF-8 エンコード文字列として送られます。

コンシューマーは、データを扱う前に json と bytes タイプをデコードする必要があります。

次のステップ

Queues の REST API ドキュメントとスキーマを確認します。
Cloudflare API への API 呼び出し方法を学びます。
キューへの書き込みと消費に適用される制限を理解します。