- Queues の配送モデル(at-least-once・バッチ配送・順序非保証)と、それを前提とした設計指針
- キューの作成と、producer バインディングからの
send()/sendBatch() - consumer Worker の
queueハンドラと、バッチサイズ / タイムアウトの調整 ack()/retry()による個別制御と、リトライ・Dead Letter Queue の設計- 「受け付けたら即応答し、重い処理は裏で行う」API パターンの実装
Queues のモデル — producer と consumer を分離する
Cloudflare Queues は、Workers に統合されたメッセージキューです。HTTP リクエストを受けた Worker がその場で重い処理(画像変換、外部 API 呼び出し、メール送信など)を実行すると、応答はその分遅くなり、失敗時のリカバリも複雑になります。Queues を使うと「受け付ける側」と「処理する側」を分離できます。登場人物は 4 つです。
| 概念 | 役割 |
|---|---|
| キュー | メッセージを溜めるバッファ。書き込みに応じて自動でスケールする |
| producer | キューへメッセージを書き込むクライアント。Worker にバインディングを設定して send() を呼ぶ |
| consumer | キューからメッセージを受け取るクライアント。queue ハンドラを持つ Worker(push 型)が基本 |
| メッセージ | キューに載せるデータ。既定の content type は json で、JSON 化できるオブジェクトを送れる |
最小構成はこれだけです。producer はバインディング経由で送り、consumer は fetch の代わりに(あるいは同じ Worker 内に併設して)queue ハンドラを実装します。
// producer 側 — バインディング経由でメッセージを送る
await env.MY_QUEUE.send({ url: req.url, method: req.method });
// consumer 側 — queue ハンドラでバッチを受け取る
export default {
async queue(batch, env, ctx): Promise<void> {
for (const msg of batch.messages) {
console.log(msg.body);
}
},
} satisfies ExportedHandler<Env>;
配送保証は at-least-once(少なくとも 1 回)です。メッセージは必ず 1 回以上配送されますが、まれに 2 回以上届くことがあります。また、発行した順序どおりに consumer へ届くことは保証されません。したがって「同じメッセージを 2 回処理しても壊れない」冪等な設計が基本になります。ドキュメントが推奨するのは、メッセージ書き込み時に一意な ID を生成し、データベースの主キーや外部 API の冪等キーとして使い、重複を受け側で弾く方法です。
接続関係には明確なルールがあります。1 つのキューに書き込む producer Worker の数に制限はありませんが、1 つのキューに接続できる consumer Worker は 1 つだけです。逆に、1 つの consumer Worker を複数のキューに接続することはでき、その場合は batch.queue(バッチの取得元キュー名)で処理を分岐できます。なお HTTP で任意の環境からメッセージを取得する pull 型 consumer もありますが、本章では Workers による push 型に絞ります。
バッチ配送 — サイズとタイムアウトの調整
consumer へのメッセージ配送はバッチ単位です。呼び出し回数を減らしてコストを下げ、外部 API への書き込みをまとめられます。バッチの形は consumer 設定の 2 つの値で決まり、どちらかの条件に先に達した時点で配送されます。
| 設定 | 意味 | 既定値 | 範囲 |
|---|---|---|---|
max_batch_size | 1 バッチの最大メッセージ数 | 10 件 | 1〜100 件 |
max_batch_timeout | バッチ配送まで待つ最大秒数 | 5 秒 | 0〜60 秒 |
たとえば max_batch_size = 30、max_batch_timeout = 10 なら、30 件溜まった時点か、10 秒経過した時点のどちらか早い方でバッチが届きます。キューが空のときは queue ハンドラ自体が呼ばれないので、無駄な起動は発生しません。調整の軸は「どれだけ待てるか(レイテンシ)」「外部システムへ書き込む単位」「呼び出し回数(コスト)」の 3 つです。
producer 側もまとめ書きできます。sendBatch() は 1 回の呼び出しで最大 100 件(合計 256 KB まで)を書き込みます。また delaySeconds オプションで、メッセージが consumer へ配送可能になるまでの遅延を最大 24 時間まで指定できます。
// 最大 100 件(合計 256 KB まで)をまとめて送る
await env.MY_QUEUE.sendBatch(items.map((item) => ({ body: item })));
// 10 分後から配送されるメッセージ
await env.MY_QUEUE.send(task, { delaySeconds: 600 });
リトライ・ack / retry の個別制御・Dead Letter Queue
既定の挙動はバッチ単位の all-or-nothing です。queue ハンドラが正常に return すれば(返した Promise と waitUntil() に渡した Promise がすべて解決すれば)バッチ全体が確認応答(ack)され、例外を投げればバッチ全体がリトライ対象になります。リトライ回数は consumer 設定の max_retries(既定 3 回)で制御し、上限に達したメッセージはキューから削除されます。
ただし「10 件中 8 件目だけ失敗」のようなケースで全 10 件を再配送されると、処理済みの 7 件が重複実行されます。これを避けるのが個別制御です。
ack() で確定させている(薄い ○)ので、下段に残るのは失敗した 1 件だけです。増えるコードは ack() の 1 行、減る再配送は 9 件ぶんです。msg.ack()— そのメッセージを処理済みとして確定する。以降ハンドラが例外を投げても再配送されないmsg.retry(options?)— そのメッセージだけを次のバッチで再配送させる(negative acknowledgement)batch.ackAll()/batch.retryAll(options?)— バッチ全体に対する同等操作
優先順位のルールは「先勝ち・個別優先」です。同じメッセージに対する 2 回目以降の ack() / retry() は無視され、個別に呼んだメッセージにはバッチ全体への ackAll() / retryAll() は適用されません。
retry() には delaySeconds を渡せます。各メッセージには配送試行回数 attempts(1 始まり)が付いているので、組み合わせると指数バックオフを実装できます。上流 API が 429 Too Many Requests を返したときに消費速度を落とす、といった使い方が典型です。
const BASE_DELAY_SECONDS = 30;
export default {
async queue(batch, env, ctx): Promise<void> {
for (const msg of batch.messages) {
// attempts は 1 から始まる配送試行回数
msg.retry({
delaySeconds: BASE_DELAY_SECONDS ** msg.attempts,
});
}
},
} satisfies ExportedHandler<Env>;
max_retries を使い切ったメッセージの行き先が Dead Letter Queue(DLQ)です。consumer 設定の dead_letter_queue にキュー名を指定すると、リトライ上限に達したメッセージは削除される代わりにそのキューへ書き込まれます(指定した名前のキューが存在しなければ自動作成されます)。DLQ もただのキューなので、consumer を接続して内容を検査・再処理できます。consumer のない DLQ に入ったメッセージは 4 日で削除される点に注意してください。なお、リトライは 1 回ごとに読み取りオペレーションとして課金対象になります。
attempts を 1 ずつ増やしながらキューへ戻る周回に入ります。この周回を max_retries 回ぶん使い切って初めて DLQ へ移り、そこに consumer を繋がなければ 4 日で失われます。プランと制限(2026-08 時点)
Queues は Workers Free / Paid の両プランで利用できます。課金単位は「オペレーション」で、64 KB ごとの書き込み・読み取り・削除がそれぞれ 1 オペレーションです。通常、1 メッセージの配送には書き込み 1 + 読み取り 1 + 削除 1 の計 3 オペレーションかかります。
| 項目 | Workers Free | Workers Paid |
|---|---|---|
| 無料枠 | 10,000 オペレーション/日 | 1,000,000 オペレーション/月 |
| 超過料金 | —(枠内のみ) | $0.40 / 100 万オペレーション |
| メッセージ保持期間 | 24 時間(変更不可) | 既定 4 日、最長 14 日まで設定可 |
Free プランでも本章のハンズオンは完走できますが、保持期間 24 時間固定・1 日 10,000 オペレーションという制約があるため、実運用では Workers Paid が前提と考えてください。主な技術的制限は次のとおりです。
- メッセージサイズ: 最大 128 KB(約 100 バイトの内部メタデータ込み)
- スループット: キューあたり 5,000 メッセージ/秒。超過すると
send()/sendBatch()がToo Many Requests例外を投げる - キュー数: アカウントあたり 10,000。バックログ: キューあたり 25 GB
- consumer の実行時間: 1 呼び出しあたり wall time 15 分。CPU 時間は既定 30 秒(
limits.cpu_msで最大 5 分まで拡張可) - 同時 consumer 呼び出し: 最大 250。バックログの増加に応じて自動でスケールアウトし、
max_concurrencyで上限を固定できる
ジョブを受け付けて即座に 202 Accepted を返し、実処理はキューへ逃がす API の「受付側」を作ります。所要時間は約 15 分です。
Queues は 2026-08 時点で Workers Free プランでも利用できます(10,000 オペレーション/日・保持期間 24 時間固定)。継続的に使う場合は Workers Paid プラン(月 100 万オペレーション込み)を前提にしてください。
- Worker プロジェクトを作成し、キューを作る。テンプレートは「Hello World example」→「Worker only」→「TypeScript」を選択する
npm create cloudflare@latest -- job-api cd job-api npx wrangler queues create job-queueキュー名は 1〜63 文字、英数字とダッシュのみ(先頭・末尾は英数字)で、作成後に変更できません。用途がわかる名前を付けましょう。
wrangler.jsoncに producer バインディングを追加する。bindingの値が Worker コード内の変数名になる{ "queues": { "producers": [ { "queue": "job-queue", "binding": "JOB_QUEUE" } ] } }src/index.tsを受付 API に書き換える。send()の Promise が解決した時点でキューへの書き込みは永続化されているので、安心して即応答できるtype JobMessage = { name: string; fail?: boolean; receivedAt: number; }; interface Env { JOB_QUEUE: Queue<JobMessage>; } export default { async fetch(req, env, ctx): Promise<Response> { const url = new URL(req.url); if (req.method === "POST" && url.pathname === "/jobs") { const payload = (await req.json()) as { name?: string; fail?: boolean }; await env.JOB_QUEUE.send({ name: payload.name ?? "unnamed", fail: payload.fail ?? false, receivedAt: Date.now(), }); return Response.json({ status: "accepted" }, { status: 202 }); } return new Response("POST /jobs にジョブを送ってください", { status: 404 }); }, } satisfies ExportedHandler<Env>;- ローカルで起動して動作確認する。Queues は Miniflare によるローカルシミュレーションで動く(
wrangler dev --remoteは非対応)npx wrangler dev # 別ターミナルから curl -X POST http://localhost:8787/jobs \ -H "Content-Type: application/json" \ -d '{"name":"resize-image"}'
{"status":"accepted"} が返れば成功です。まだ consumer がいないので、送ったメッセージはキューに溜まったままです(consumer がなければ保持期間経過後に削除されます)。次のハンズオンで処理側を作ります。
同じ Worker に queue ハンドラを追加して、キューに溜まったジョブを裏で処理します。成功したメッセージだけを ack() で個別確定する実践形です。所要時間は約 15 分です。
src/index.tsのfetchハンドラはそのままに、queueハンドラと処理関数を追加するexport default { // ...9-1 で書いた fetch ハンドラはそのまま... async queue(batch, env, ctx): Promise<void> { for (const msg of batch.messages) { try { await processJob(msg.body); msg.ack(); // 成功したメッセージだけ個別に確定 } catch (err) { console.error(`attempt ${msg.attempts} failed:`, err); msg.retry({ delaySeconds: 5 * msg.attempts }); } } }, } satisfies ExportedHandler<Env, JobMessage>; async function processJob(job: JobMessage): Promise<void> { if (job.fail) { throw new Error(`job "${job.name}" failed`); // 9-3 で使う失敗スイッチ } await new Promise((resolve) => setTimeout(resolve, 1000)); // 重い処理のダミー console.log("processed:", job.name); }ExportedHandler<Env, JobMessage>と型引数を渡すと、msg.bodyがJobMessage型になります。wrangler.jsoncに consumer 設定を追加する(producersと並べて書く){ "queues": { "producers": [ { "queue": "job-queue", "binding": "JOB_QUEUE" } ], "consumers": [ { "queue": "job-queue", "max_batch_size": 10, "max_batch_timeout": 5 } ] } }npx wrangler devを再起動し、9-1 の curl を数回実行する。最大 5 秒(max_batch_timeout)待つと、dev のターミナルにprocessed: resize-imageのログがまとめて流れる- デプロイして本番でも確認する。URL は deploy 時の出力に表示される
*.workers.devのものに置き換えるnpx wrangler deploy npx wrangler tail # 別ターミナルから本番 URL へ curl -X POST https://job-api.<YOUR_SUBDOMAIN>.workers.dev/jobs \ -H "Content-Type: application/json" \ -d '{"name":"send-report"}'
curl が即座に 202 を返し、数秒後に wrangler tail 側へ processed: send-report が流れれば、「受け付け即応答 + 裏で処理」パターンの完成です。
処理に失敗し続けたジョブを握りつぶさず、DLQ へ退避して後から調査できるようにします。所要時間は約 10 分です。
- DLQ 用のキューを作る:
npx wrangler queues create job-dlq(consumer 設定に書いた名前のキューが存在しない場合は自動作成もされますが、明示的に作っておくと確実です) - consumer 設定に
max_retriesとdead_letter_queueを追加して再デプロイする{ "queues": { "consumers": [ { "queue": "job-queue", "max_batch_size": 10, "max_batch_timeout": 5, "max_retries": 3, "dead_letter_queue": "job-dlq" } ] } } - 9-2 で仕込んだ失敗スイッチを使い、わざと失敗するジョブを送る
curl -X POST http://localhost:8787/jobs \ -H "Content-Type: application/json" \ -d '{"name":"broken-job","fail":true}' - ログを観察する。
attempt 1 failedからattempt 4 failed(初回 + リトライ 3 回)まで出力された後、そのメッセージはjob-dlqへ書き込まれ、再配送されなくなる。npx wrangler queues info job-dlqでキューの情報を確認できる
DLQ もただのキューなので、consumer を接続すれば「失敗ジョブを Slack に通知する」「内容を修正して元のキューに戻す」といった運用が組めます。consumer のない DLQ のメッセージは 4 日で消えることを忘れないでください。
つまずきポイント
複数の Worker を同じキューの consumer にしようとすると、デプロイ時にエラーになります。処理を分けたいときはキュー自体を分けるのが基本です。逆に 1 つの consumer Worker を複数キューに接続するのは可能で、batch.queue で分岐できます。
10 件中 1 件の失敗でハンドラが例外を投げると、処理済みの 9 件も含めて再配送されます。メール送信や決済のような非冪等な処理では、成功のたびに msg.ack() を呼んで個別に確定してください。at-least-once 配送なので、そもそも冪等キーによる重複排除も併用するのが安全です。
forEach では非同期処理を待てない
batch.messages.forEach() は async 関数の完了を待ちません。for...of か await Promise.all(batch.messages.map(...)) を使ってください。また、ハンドラ return 後に完了させたい処理(ログ送信など)は ctx.waitUntil() に渡すのが唯一サポートされた方法で、渡し忘れた未解決の Promise は完了しないことがあります。
send() は例外を投げる。waitUntil() 経由だと失敗が握りつぶされる
send() / sendBatch() は、128 KB 超のメッセージ、スループット超過(Too Many Requests)、バックログ 25 GB 超過(Storage Limit Exceeded)などで例外を投げます。応答をブロックしたくないからと ctx.waitUntil(env.MY_QUEUE.send(...)) とすると、これらのエラーは暗黙に無視されます。確実に受け付けたことを応答したいなら await + try...catch で扱ってください。
理解度チェック
Q1. Queues の配送保証モデルは? その前提で consumer 側に求められる設計は?
at-least-once(少なくとも 1 回)です。まれに同じメッセージが 2 回以上届くため、メッセージに一意な ID を持たせ、DB の主キーや外部 API の冪等キーとして使って重複を弾く冪等な設計が求められます。順序も保証されない点に注意します。
Q2. max_batch_size: 100、max_batch_timeout: 30 のとき、バッチはいつ配送される?
「100 件溜まる」か「30 秒経過」のどちらか早い方が満たされた時点です。30 秒経過時にキューにあるのが 40 件なら 40 件のバッチが届きます。キューが空ならハンドラは呼ばれません。
Q3. 10 件のバッチ処理中に 8 件目で例外が発生した。既定の挙動と、それを避ける方法は?
既定では 10 件すべてがリトライ対象になり、処理済みの 7 件も再配送されます。処理に成功したメッセージごとに msg.ack() を呼べば、それらは再配送されなくなります。失敗した 1 件だけ msg.retry() でマークする方法もあります。
Q4. max_retries を使い切ったメッセージはどうなる?
dead_letter_queue が設定されていればそのキューへ書き込まれ、設定がなければ完全に削除されます。DLQ は通常のキューとして consumer を接続でき、consumer がない場合メッセージは 4 日で削除されます。
Q5. 同じメッセージに msg.retry() を呼んだ後に msg.ack() を呼ぶと?
ack() は無視され、メッセージは再配送されます。ack / retry は「先に呼んだ方が勝ち」で、さらに個別メッセージへの呼び出しはバッチ全体への ackAll() / retryAll() より優先されます。
まとめと次章へ
- Queues は producer(
send()/sendBatch())と consumer(queueハンドラ)を分離する、at-least-once 配送のメッセージキュー。順序は保証されないため冪等設計が前提 - 配送はバッチ単位。
max_batch_size(既定 10 件)とmax_batch_timeout(既定 5 秒)の早い方で届く - 失敗時は
max_retries(既定 3 回)までバッチ単位で再配送。msg.ack()/msg.retry()で個別制御でき、delaySecondsとattemptsでバックオフも組める - リトライ上限を超えたメッセージは Dead Letter Queue へ退避して調査・再処理する
- 「受け付け即応答 + 裏で処理」は、producer の
send()をawaitしてから 202 を返すだけで実現できる
キューは「1 発の非同期処理」に向きますが、複数ステップの処理を途中経過ごと永続化してリトライしたい場合は、次章の Workflows がより適したツールです。キューで作った受付パターンはそのまま Workflows の起動にも応用できます。