cloudflare-queues

Cloudflare Queues

ステータス：本番環境対応 ✅ | 最終確認日：2025-12-27

依存関係：cloudflare-worker-base（Worker のセットアップ用）

内容：クイックスタート • 重要なルール • 主なエラー • 一般的なユースケース • リファレンスを読み込むタイミング • 制限

---

クイックスタート（10 分）

1. キューを作成する

bunx wrangler queues create my-queue
bunx wrangler queues list

2. プロデューサー（メッセージを送信）

wrangler.jsonc:

{
  "name": "my-producer",
  "main": "src/index.ts",
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",
        "queue": "my-queue"
      }
    ]
  }
}

src/index.ts:

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

const app = new Hono<{ Bindings: Bindings }>();

app.post('/send', async (c) => {
  await c.env.MY_QUEUE.send({
    userId: '123',
    action: 'process-order',
    timestamp: Date.now(),
  });

  return c.json({ status: 'queued' });
});

export default app;

3. コンシューマー（メッセージを処理）

wrangler.jsonc:

{
  "name": "my-consumer",
  "main": "src/consumer.ts",
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"
      }
    ]
  }
}

src/consumer.ts:

import type { MessageBatch } from '@cloudflare/workers-types';

export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      console.log('Processing:', message.body);
      // Your logic here
    }
    // Implicit ack: returning successfully acknowledges all messages
  },
};

デプロイ：

bunx wrangler deploy

読み込むもの：DLQ の設定を含む完全な 6 ステップのセットアップについては、references/setup-guide.md を参照してください。

---

重要なルール

必ず行うこと ✅

1. 本番環境のキューには必ずデッドレターキューを設定する
2. 冪等でない操作（DB 書き込み、API 呼び出し）には明示的な ack() を使用する
3. 送信前にメッセージサイズを検証する（128 KB 未満）
4. 複数のメッセージにはsendBatch() を使用する（より効率的です）
5. 指数バックオフを実装する
6. ワークロードに応じて適切なバッチ設定を行う
7. キューのバックログとコンシューマーのエラーを監視する
8. コンシューマーでの非同期クリーンアップにはctx.waitUntil() を使用する
9. エラーを適切に処理する - ログ記録、アラート、再試行を行う
10. 並行処理数は自動スケーリングに任せる（必要な場合を除き、max_concurrency は設定しない）

絶対にしてはいけないこと ❌

1. メッセージの順序を前提にしない - FIFO は保証されません
2. 冪等でない操作で暗黙的な ack に依存しない - 明示的な ack() を使用する
3. 128 KB を超えるメッセージを送信しない - 失敗します
4. アクティブなメッセージがあるキューを削除しない - データが失われます
5. 本番環境でDLQ の設定を省略しない
6. キューごとに 5000 msg/s を超えない - レート制限エラーになります
7. ループ内でメッセージを同期的に処理しない - Promise.all() を使用する
8. message.attempts を無視しない - バックオフのロジックに使用する
9. 特別な理由がない限り、max_concurrency=1 に設定しない
10. 明示的な確認応答を行うパターンでは、ack() を忘れない

---

主な 3 つの重大なエラー

エラー #1：メッセージが大きすぎる

問題：メッセージが 128 KB の上限を超えています

解決策：大きなデータは R2 に保存し、参照を送信します

// ❌ Wrong
await env.MY_QUEUE.send({ data: largeArray }); // >128 KB fails

// ✅ Correct
const message = { data: largeArray };
const size = new TextEncoder().encode(JSON.stringify(message)).length;

if (size > 128000) {
  const key = `messages/${crypto.randomUUID()}.json`;
  await env.MY_BUCKET.put(key, JSON.stringify(message));
  await env.MY_QUEUE.send({ type: 'large-message', r2Key: key });
} else {
  await env.MY_QUEUE.send(message);
}

エラー #2：スループット超過

問題：キューごとに 1 秒あたり 5000 件のメッセージを超えています

解決策：sendBatch() とレート制限を使用します

// ❌ Wrong
for (let i = 0; i < 10000; i++) {
  await env.MY_QUEUE.send({ id: i }); // Too fast!
}

// ✅ Correct
const messages = Array.from({ length: 10000 }, (_, i) => ({
  body: { id: i },
}));

// Send in batches of 100
for (let i = 0; i < messages.length; i += 100) {
  await env.MY_QUEUE.sendBatch(messages.slice(i, i + 100));
}

エラー #3：1 件のメッセージが失敗すると、バッチ全体が再試行される

問題：1 件のメッセージの失敗によって、すべてのメッセージが再試行されます

解決策：明示的な確認応答を使用します

// ❌ Wrong - implicit ack
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      await env.DB.prepare('INSERT INTO orders VALUES (?, ?)').bind(
        message.body.id,
        message.body.amount
      ).run();
    }
    // If any fails, ALL retry!
  },
};

// ✅ Correct - explicit ack
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        await env.DB.prepare('INSERT INTO orders VALUES (?, ?)').bind(
          message.body.id,
          message.body.amount
        ).run();

        message.ack(); // Only ack on success
      } catch (error) {
        console.error(`Failed: ${message.id}`, error);
        // Don't ack - will retry independently
      }
    }
  },
};

DLQ の設定、自動スケーリングの問題、メッセージ削除の防止、詳細な解決策を含む 10 件すべてのエラーについては、references/error-catalog.md を読み込んでください。

---

一般的なユースケース

ユースケース 1：基本的なメッセージキュー

用途：シンプルな非同期ジョブ処理（メール、通知）

基本パターン：

// Producer
await env.MY_QUEUE.send({ type: 'email', to: '[email protected]' });

// Consumer (implicit ack - for idempotent operations)
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await sendEmail(message.body.to, message.body.content);
    }
  },
};

読み込むもの：templates/queues-producer.ts + templates/queues-consumer-basic.ts

ユースケース 2：データベースへの書き込み（冪等でない操作）

用途：データベースへの書き込みで、重複を避ける必要がある場合

読み込むもの：templates/queues-consumer-explicit-ack.ts + references/consumer-api.md

ユースケース 3：指数バックオフによる再試行

用途：レート制限のある API の呼び出し、一時的な障害

読み込むもの：templates/queues-retry-with-delay.ts + references/error-catalog.md（エラー #2、#3）

ユースケース 4：デッドレターキューパターン

用途：本番システムで、恒久的に失敗したメッセージを捕捉する必要がある場合

読み込むもの：templates/queues-dlq-pattern.ts + references/setup-guide.md（ステップ 4）

ユースケース 5：高スループット処理

用途：1 秒あたり数千件のメッセージを処理する場合

基本パターン：

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100,        // Large batches
      "max_batch_timeout": 5,       // Fast processing
      "max_concurrency": null       // Auto-scale
    }]
  }
}

読み込むもの：references/best-practices.md → スループットの最適化

---

リファレンスを読み込むタイミング

次の場合は references/setup-guide.md を読み込みます：

• 完全なセットアップ手順（キュー → プロデューサー → コンシューマー → DLQ）が必要な場合
• Cloudflare Queues を初めてセットアップする場合
• 本番環境の構成例が必要な場合
• エンドツーエンドの完全な例を確認したい場合

次の場合は references/error-catalog.md を読み込みます：

• ドキュメント化されている10個のエラーのいずれかが発生した場合
• キューの問題をトラブルシューティングする場合
• メッセージが配信または処理されない場合
• 予防チェックリストが必要な場合

次の場合は references/producer-api.md を読み込みます：

• プロデューサー API の完全なリファレンスが必要な場合
• send() または sendBatch() メソッドを使用する場合
• メッセージ形式の仕様が必要な場合
• 大きなメッセージやバッチを処理する場合

次の場合は references/consumer-api.md を読み込みます：

• コンシューマー API の完全なリファレンスが必要な場合
• 明示的な ack()、retry()、または retryAll() を使用する場合
• バッチ処理のパターンが必要な場合
• 複雑なコンシューマーロジックを実装する場合

次の場合は references/best-practices.md を読み込みます：

• キューのパフォーマンスを最適化する場合
• 本番環境へのデプロイに関する指針が必要な場合
• モニタリングと可観測性を設定する場合
• セキュリティと信頼性のパターンを確認する場合

次の場合は references/wrangler-commands.md を読み込みます：

• CLI コマンドのリファレンスが必要な場合
• キューを管理する場合（作成、削除、一覧表示）
• 配信を制御する場合（一時停止、再開）
• キューの問題をデバッグする場合
• リアルタイムのモニタリングとパフォーマンス分析を行う場合

次の場合は references/typescript-types.md を読み込みます：

• TypeScript の型定義の完全な情報が必要な場合
• Queue、MessageBatch、または Message インターフェースを扱う場合
• 型安全なプロデューサーまたはコンシューマーを実装する場合
• メッセージ本文にジェネリック型を使用する場合
• メッセージ検証用の型ガードが必要な場合

次の場合は references/production-checklist.md を読み込みます：

• 本番環境へのデプロイを準備する場合
• デプロイ前の確認チェックリストが必要な場合
• 本番環境のベストプラクティスについて詳しい説明が必要な場合
• モニタリング、DLQ、またはエラー処理を設定する場合
• 負荷テストやセキュリティレビューを計画する場合

次の場合は references/pull-consumers.md を読み込みます：

• Workers 以外の環境からメッセージをコンシュームする必要がある場合
• 既存のバックエンドサービス（Node.js、Python、Go）と統合する場合
• プッシュベースのコンシュームではなく、プルベースのポーリングを実装する場合
• コンテナ化されたアプリケーションやレガシーアプリケーションを扱う場合

次の場合は references/http-publishing.md を読み込みます：

• HTTP 経由で外部システムからメッセージを発行する場合
• サードパーティサービスの Webhook と統合する場合
• Workers をデプロイせずにメッセージを送信する必要がある場合
• CI/CD パイプラインの通知を実装する場合

次の場合は references/r2-event-integration.md を読み込みます：

• R2 バケットイベントをトリガーにキューメッセージを送信する場合
• 画像やドキュメントの自動処理を実装する場合
• イベント駆動型データパイプラインを設定する場合
• R2 オブジェクトのアップロード／削除通知が必要な場合

---

エージェントとコマンド

利用可能なエージェント：

• queue-debugger - キューの問題を9段階で診断分析（体系的なトラブルシューティング）
• queue-optimizer - パフォーマンス調整とコスト最適化（バッチサイズ、並行処理数、再試行）

利用可能なコマンド：

• /queue-setup - キュー全体のセットアップを行う対話型ウィザード
• /queue-troubleshoot - 一般的な問題の簡易診断
• /queue-monitor - リアルタイムのメトリクスとステータスを表示

---

制限とクォータ

重要な制限：

• メッセージサイズ：メッセージあたり最大 128 KB
• スループット：キューあたり毎秒 5,000 メッセージ
• バッチサイズ：sendBatch() あたり最大 100 メッセージ
• コンシューマーの CPU 時間：30 秒（デフォルト）、300 秒（構成による最大値）
• 最大再試行回数：構成可能（デフォルトは3回）
• キュー名：最大63文字、小文字／数字／ハイフンのみ

制限への対処方法と最適化戦略については、references/best-practices.md を読み込んでください。

---

設定リファレンス

プロデューサー：wrangler.jsonc の queues.producers 配列に、binding と queue フィールドを指定してキューバインディングを追加します。

コンシューマー：wrangler.jsonc の queues.consumers 配列で、queue、max_batch_size（1～100）、max_batch_timeout（0～60s）、max_retries、dead_letter_queue、および必要に応じて max_concurrency（デフォルト：自動スケーリング）を設定します。

CPU 制限：処理に時間がかかる場合は、limits.cpu_ms をデフォルトの 30,000ms から増やします。

完全な設定例については references/setup-guide.md を、すぐに本番環境で使用できる設定については templates/wrangler-queues-config.jsonc を読み込んでください。

---

同梱リソースの使用

リファレンス（references/）

• setup-guide.md - 6 ステップの完全なセットアップ（キュー → プロデューサー → コンシューマー → DLQ → デプロイ → 本番環境の設定）
• error-catalog.md - 10 種類すべてのエラーと解決策 ＋ 予防チェックリスト
• producer-api.md - プロデューサー API の全容（send、sendBatch、メッセージ形式）
• consumer-api.md - コンシューマー API の全容（ack、retry、バッチ処理）
• best-practices.md - パフォーマンス、モニタリング、セキュリティ、信頼性のパターン
• wrangler-commands.md - CLI リファレンス（create、delete、list、pause、resume）
• typescript-types.md - Queue、MessageBatch、Message の TypeScript 型定義の全容
• production-checklist.md - デプロイ前の検証とベストプラクティス
• pull-consumers.md - Workers 以外の環境向けのプル型コンシューマー（HTTP ポーリング）
• http-publishing.md - 外部システムから HTTP API 経由でメッセージを公開する方法
• r2-event-integration.md - キューメッセージをトリガーする R2 イベント通知

テンプレート（templates/）

• queues-producer.ts - 単一送信とバッチ送信に対応する基本的なプロデューサー
• queues-consumer-basic.ts - 暗黙的な ack を使用するコンシューマー（冪等な操作）
• queues-consumer-explicit-ack.ts - 明示的な ack を使用するコンシューマー（非冪等な操作）
• queues-retry-with-delay.ts - 指数バックオフによる再試行パターン
• queues-dlq-pattern.ts - デッドレターキューのセットアップとコンシューマー
• wrangler-queues-config.jsonc - 本番環境向けの完全な設定

---

TypeScript 型

完全な型定義である Queue、MessageBatch<Body>、Message<Body>、QueueSendOptions を使用するには、@cloudflare/workers-types パッケージを使用してください。

インターフェース、ジェネリック、型ガード、使用例を含む完全な型リファレンスについては references/typescript-types.md を読み込んでください。

---

モニタリングとデバッグ

主なコマンド：wrangler queues info（ステータス）、wrangler tail（ログ）、wrangler queues pause-delivery/resume-delivery（制御）。

リアルタイムモニタリング、デバッグのワークフロー、パフォーマンス分析コマンドを含む完全な CLI リファレンスについては references/wrangler-commands.md を読み込んでください。

---

本番環境チェックリスト

デプロイ前の 12 項目チェックリスト：DLQ の設定、メッセージ確認戦略、サイズ検証、バッチ最適化、同時実行設定、CPU 制限、エラー処理、モニタリング、レート制限、冪等性、負荷テスト、セキュリティレビュー。

詳細な説明、コード例、デプロイのワークフローを含む完全なチェックリストについては references/production-checklist.md を読み込んでください。

---

関連スキル

• cloudflare-worker-base - Worker のセットアップと設定
• cloudflare-d1 - キューコンシューマー向けのデータベース統合
• cloudflare-r2 - 大きなメッセージペイロードの保存
• cloudflare-workflows - より複雑なオーケストレーションのニーズ

---

公式ドキュメント

• Cloudflare Queues: https://developers.cloudflare.com/queues/
• 設定: https://developers.cloudflare.com/queues/configuration/
• Wrangler コマンド: https://developers.cloudflare.com/workers/wrangler/commands/#queues
• 制限: https://developers.cloudflare.com/queues/platform/limits/
• トラブルシューティング: https://developers.cloudflare.com/queues/observability/troubleshooting/
• ベストプラクティス: https://developers.cloudflare.com/queues/configuration/best-practices/

---

質問や問題がありますか？

1. references/error-catalog.mdで、10件すべてのエラーと解決策を確認してください
2. references/setup-guide.mdで、セットアップ手順全体を確認してください
3. references/best-practices.mdで、本番環境向けのパターンを確認してください
4. 公式ドキュメントを確認してください：https://developers.cloudflare.com/queues/