コンシューマーの同時実行

コンシューマー同時実行（consumer concurrency）では、キューからメッセージを処理するコンシューマー Worker が、キューへの書き込み速度に合わせて水平方向に自動スケールします。

多くのシステムでは、キューへの書き込み速度が、単一コンシューマーの読み取り・処理速度を簡単に上回ります。コンシューマーがメッセージ内容の解析、ストレージやデータベースへの書き込み、サードパーティ（上流）API の呼び出しを行うためです。

プロデューサー側は常にスケールします。上限はキューごとの 1 秒あたりの最大メッセージ数です。

同時実行を有効にする

デフォルトでは、すべてのキューで同時実行が有効です。キューのバックログやエラー率に応じて、コンシューマーは必要に応じて最大同時呼び出し数まで自動でスケールアップします。

メッセージのバッチを処理したあと、Queues は同時コンシューマー数を調整すべきかを確認します。キューに対して呼び出す同時コンシューマー数は、次の要素に基づいて自動スケールします。

キュー上のメッセージ数（バックログ）と、その増加速度。
失敗した呼び出しと成功した呼び出しの比率。失敗した呼び出しとは、queue() ハンドラーが void（何も返さない）ではなく、捕捉していない例外を返した場合です。
そのコンシューマーに設定した max_concurrency の値。

可能な限り、Queues はバックログが指数関数的に増えないようにします。キュー上のメッセージが処理前にメッセージ保持上限に達する事態を減らすためです。

1 秒あたり 100 件を書き込むキューに、100 件のバッチ処理に 5 秒かかる同時コンシューマーが 1 つだけの場合、処理中のメッセージ数はコンシューマーの処理速度を上回って増え続けます。

この状況では、Queues は増えるバックログを検知し、同時コンシューマー Worker の呼び出し数を、定常状態でおおよそ 5 までスケールします。着信メッセージの速度が下がる、コンシューマーの処理が速くなる、またはコンシューマーがエラーを出し始めるまで、この状態が続きます。

コンシューマーが自動スケールしない場合、よくある原因は次のとおりです。

max_concurrency が 1 に設定されています。
コンシューマー Worker がメッセージを処理せず、エラーを返しています。コンシューマーが正常か確認します。
メッセージのバッチを処理中です。Queues はバッチ全体の処理が終わったあとで自動スケールするかを判断するため、バッチ処理中は自動スケールしません。バッチサイズを小さくするか、コンシューマーをリファクタして処理を速くすることを検討します。

上流 API やシステムの制約があるワークフローでは、上流を過負荷にしない代わりにバックログを増やし、全体のレイテンシを受け入れる方がよい場合があります。

コンシューマー Worker の同時実行は、次の 2 通りで設定できます。

ダッシュボードからコンシューマー Worker の同時実行を設定する手順です。

Cloudflare ダッシュボードで Queues ページを開きます。
Queues を開く ↗
キューを選び、Settings を開きます。
Consumer details の Edit Consumer を選びます。
Maximum consumer invocations を 1 から 250 の値に設定します。この値は、キューが使える同時コンシューマー呼び出しの最大数です。

固定の最大値を外すには、auto (recommended) を選びます。

処理速度より速くキューへ書き込んでいる場合、メッセージはそのキューの最大保持期間に達することがあります。上限に達した個別メッセージは期限切れになり、キューから削除されます。

キューごとの同時コンシューマー呼び出し数に固定の上限を付けるには、Wrangler ファイルで max_concurrency を設定します。

{
	"queues": {
		"consumers": [
			{
				"queue": "my-queue",
				"max_concurrency": 1
			}
		]
	}
}

[[queues.consumers]]
queue = "my-queue"
max_concurrency = 1

上限を外すには、対象キューの [[queues.consumers]] 設定から max_concurrency を削除し、npx wrangler deploy で設定を反映します。

複数のコンシューマー Worker が呼び出されると、呼び出しごとに CPU 時間の料金が発生します。

コンシューマーの課金は Workers の標準利用モデルに従います。開発者はリクエストと、そのリクエストで使った CPU 時間に対して課金されます。

バッチ処理に 2 秒かかるコンシューマー Worker が 5,000 万（50,000,000）件のメッセージを処理する場合、同時実行で速く処理しても、1 件ずつ遅く処理しても、総コストは同じです。