Cloudflare ハンズオン教科書|2026-08 版
第 9 章

Queues — 非同期メッセージング

重い処理はキューに逃がす。producer / consumer を両方書く

この章で学ぶこと

Queues のモデル — producer と consumer を分離する

Cloudflare Queues は、Workers に統合されたメッセージキューです。HTTP リクエストを受けた Worker がその場で重い処理(画像変換、外部 API 呼び出し、メール送信など)を実行すると、応答はその分遅くなり、失敗時のリカバリも複雑になります。Queues を使うと「受け付ける側」と「処理する側」を分離できます。登場人物は 4 つです。

概念役割
キューメッセージを溜めるバッファ。書き込みに応じて自動でスケールする
producerキューへメッセージを書き込むクライアント。Worker にバインディングを設定して send() を呼ぶ
consumerキューからメッセージを受け取るクライアント。queue ハンドラを持つ Worker(push 型)が基本
メッセージキューに載せるデータ。既定の content type は json で、JSON 化できるオブジェクトを送れる
Queues なし — クライアントは処理の完了まで待つ クライアント Worker 重い処理 画像変換・送信 POST /jobs 同期実行 処理完了後に応答 Queues あり — send() が解決した時点で応答を返せる クライアント 受付 Worker producer キュー job-queue consumer queue ハンドラ POST /jobs send() バッチ配送 202 を即返す 裏で重い処理
図 9-1 上下の列は対応しています(4 列目=重い処理が走る場所)。変わるのは「クライアントへ戻る矢印がどこから出るか」で、上段は右端の処理が終わるまで戻れず、下段は 2 列目から即戻れます。差し込まれたのは 3 列目のキュー 1 つだけで、それだけで応答時間から処理時間が丸ごと消えます。

最小構成はこれだけです。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_size1 バッチの最大メッセージ数10 件1〜100 件
max_batch_timeoutバッチ配送まで待つ最大秒数5 秒0〜60 秒

たとえば max_batch_size = 30、max_batch_timeout = 10 なら、30 件溜まった時点か、10 秒経過した時点のどちらか早い方でバッチが届きます。キューが空のときは queue ハンドラ自体が呼ばれないので、無駄な起動は発生しません。調整の軸は「どれだけ待てるか(レイテンシ)」「外部システムへ書き込む単位」「呼び出し回数(コスト)」の 3 つです。

設定: max_batch_size = 10 件 / max_batch_timeout = 5 秒 ① 到着が速いとき — 件数が先に埋まる 10 件到達 → 配送 5 秒を待たずに切り出す ○ = メッセージ 1 件 0 秒 3 秒 5 秒 ② 到着が遅いとき — 時間が先に来る 5 秒経過 → 4 件で配送 10 件に届かないまま締め切り 0 秒 5 秒
図 9-2 横軸は時間、○ ひとつがメッセージ 1 件です。バッチの切れ目(縦線)は「件数」と「時間」のうち先に達した方で決まります。①では 3 秒で 10 件が埋まるのでタイムアウトを待たず、②では件数が足りないまま 5 秒で締め切られ、4 件のバッチが届きます。

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 件が重複実行されます。これを避けるのが個別制御です。

既定 — ハンドラが例外を投げる 届いたバッチ 8 件目で失敗 × 1 件も確定していない 次に届くバッチ 10 件 — うち 7 件は 2 度目の実行 個別制御 — 成功ごとに ack() 届いたバッチ 8 件目で失敗 × ack() 済みの 9 件は戻らない この 1 件だけ 次に届くバッチ 1 件 — 2 度目の実行は 0 件
図 9-3 上段が届いたバッチ、下段が次に届くバッチです。左は例外ひとつでバッチ全体が巻き戻るため、成功していた 7 件まで 2 度目の実行になります。右は成功のたびに ack() で確定させている(薄い ○)ので、下段に残るのは失敗した 1 件だけです。増えるコードは ack() の 1 行、減る再配送は 9 件ぶんです。

優先順位のルールは「先勝ち・個別優先」です。同じメッセージに対する 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 回ごとに読み取りオペレーションとして課金対象になります。

producer の send() キュー job-queue consumer queue ハンドラ バッチ配送 ack() で確定 処理成功 キューから削除 retry() → attempts +1 4 回目も失敗(max_retries 超過) DLQ job-dlq 未接続 4 日で削除 繋げば再処理できる
図 9-4 成功したメッセージは上へ抜けて一瞬で消えますが、失敗したメッセージは attempts を 1 ずつ増やしながらキューへ戻る周回に入ります。この周回を max_retries 回ぶん使い切って初めて DLQ へ移り、そこに consumer を繋がなければ 4 日で失われます。

プランと制限(2026-08 時点)

Queues は Workers Free / Paid の両プランで利用できます。課金単位は「オペレーション」で、64 KB ごとの書き込み・読み取り・削除がそれぞれ 1 オペレーションです。通常、1 メッセージの配送には書き込み 1 + 読み取り 1 + 削除 1 の計 3 オペレーションかかります。

項目Workers FreeWorkers Paid
無料枠10,000 オペレーション/日1,000,000 オペレーション/月
超過料金—(枠内のみ)$0.40 / 100 万オペレーション
メッセージ保持期間24 時間(変更不可)既定 4 日、最長 14 日まで設定可

Free プランでも本章のハンズオンは完走できますが、保持期間 24 時間固定・1 日 10,000 オペレーションという制約があるため、実運用では Workers Paid が前提と考えてください。主な技術的制限は次のとおりです。

ハンズオン 9-1|受付 API とキューを作る(producer 編)

ジョブを受け付けて即座に 202 Accepted を返し、実処理はキューへ逃がす API の「受付側」を作ります。所要時間は約 15 分です。

プランについて

Queues は 2026-08 時点で Workers Free プランでも利用できます(10,000 オペレーション/日・保持期間 24 時間固定)。継続的に使う場合は Workers Paid プラン(月 100 万オペレーション込み)を前提にしてください。

  1. Worker プロジェクトを作成し、キューを作る。テンプレートは「Hello World example」→「Worker only」→「TypeScript」を選択する
    npm create cloudflare@latest -- job-api
    cd job-api
    npx wrangler queues create job-queue
    

    キュー名は 1〜63 文字、英数字とダッシュのみ(先頭・末尾は英数字)で、作成後に変更できません。用途がわかる名前を付けましょう。

  2. wrangler.jsonc に producer バインディングを追加する。binding の値が Worker コード内の変数名になる
    {
      "queues": {
        "producers": [
          {
            "queue": "job-queue",
            "binding": "JOB_QUEUE"
          }
        ]
      }
    }
    
  3. 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>;
    
  4. ローカルで起動して動作確認する。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 がなければ保持期間経過後に削除されます)。次のハンズオンで処理側を作ります。

ハンズオン 9-2|consumer Worker で裏処理を実装する

同じ Worker に queue ハンドラを追加して、キューに溜まったジョブを裏で処理します。成功したメッセージだけを ack() で個別確定する実践形です。所要時間は約 15 分です。

  1. 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 型になります。

  2. wrangler.jsonc に consumer 設定を追加する(producers と並べて書く)
    {
      "queues": {
        "producers": [
          { "queue": "job-queue", "binding": "JOB_QUEUE" }
        ],
        "consumers": [
          {
            "queue": "job-queue",
            "max_batch_size": 10,
            "max_batch_timeout": 5
          }
        ]
      }
    }
    
  3. npx wrangler dev を再起動し、9-1 の curl を数回実行する。最大 5 秒(max_batch_timeout)待つと、dev のターミナルに processed: resize-image のログがまとめて流れる
  4. デプロイして本番でも確認する。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 が流れれば、「受け付け即応答 + 裏で処理」パターンの完成です。

ハンズオン 9-3|Dead Letter Queue で失敗メッセージを退避する

処理に失敗し続けたジョブを握りつぶさず、DLQ へ退避して後から調査できるようにします。所要時間は約 10 分です。

  1. DLQ 用のキューを作る: npx wrangler queues create job-dlq(consumer 設定に書いた名前のキューが存在しない場合は自動作成もされますが、明示的に作っておくと確実です)
  2. 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"
          }
        ]
      }
    }
    
  3. 9-2 で仕込んだ失敗スイッチを使い、わざと失敗するジョブを送る
    curl -X POST http://localhost:8787/jobs \
      -H "Content-Type: application/json" \
      -d '{"name":"broken-job","fail":true}'
    
  4. ログを観察する。attempt 1 failed から attempt 4 failed(初回 + リトライ 3 回)まで出力された後、そのメッセージは job-dlq へ書き込まれ、再配送されなくなる。npx wrangler queues info job-dlq でキューの情報を確認できる

DLQ もただのキューなので、consumer を接続すれば「失敗ジョブを Slack に通知する」「内容を修正して元のキューに戻す」といった運用が組めます。consumer のない DLQ のメッセージは 4 日で消えることを忘れないでください。

つまずきポイント

1 つのキューに consumer は 1 つしか繋げない

複数の 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() より優先されます。

まとめと次章へ

キューは「1 発の非同期処理」に向きますが、複数ステップの処理を途中経過ごと永続化してリトライしたい場合は、次章の Workflows がより適したツールです。キューで作った受付パターンはそのまま Workflows の起動にも応用できます。