cloudflare-queues

Cloudflare Queues

ステータス: 本番環境対応済み ✅ 最終更新日: 2025-10-21 依存関係: cloudflare-worker-base（Worker のセットアップ用） 最新バージョン: [email protected]、@cloudflare/[email protected]

---

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

1. キューを作成する

# Create a new queue
npx wrangler queues create my-queue

# Output:
# ✅ Successfully created queue my-queue

# List all queues
npx wrangler queues list

# Get queue info
npx wrangler queues info my-queue

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

wrangler.jsonc:

{
  "name": "my-producer",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "queues": {
    "producers": [
      {
        "binding": "MY_QUEUE",           // Available as env.MY_QUEUE
        "queue": "my-queue"               // Queue name from step 1
      }
    ]
  }
}

src/index.ts（プロデューサー）:

import { Hono } from 'hono';

type Bindings = {
  MY_QUEUE: Queue;
};

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

// Send single message
app.post('/send', async (c) => {
  const body = await c.req.json();

  await c.env.MY_QUEUE.send({
    userId: body.userId,
    action: 'process-order',
    timestamp: Date.now(),
  });

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

// Send batch of messages
app.post('/send-batch', async (c) => {
  const items = await c.req.json();

  await c.env.MY_QUEUE.sendBatch(
    items.map((item) => ({
      body: { userId: item.userId, action: item.action },
    }))
  );

  return c.json({ status: 'queued', count: items.length });
});

export default app;

3. コンシューマーをセットアップする（メッセージを処理）

コンシューマー Worker を作成する:

# In a new directory
npm create cloudflare@latest my-consumer -- --type hello-world --ts
cd my-consumer

wrangler.jsonc:

{
  "name": "my-consumer",
  "main": "src/index.ts",
  "compatibility_date": "2025-10-11",
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",              // Queue to consume from
        "max_batch_size": 10,             // Process up to 10 messages at once
        "max_batch_timeout": 5            // Or wait max 5 seconds
      }
    ]
  }
}

src/index.ts（コンシューマー）:

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    console.log(`Processing batch of ${batch.messages.length} messages`);

    for (const message of batch.messages) {
      console.log('Message:', message.id, message.body, `Attempt: ${message.attempts}`);

      // Your processing logic here
      await processMessage(message.body);
    }

    // Implicit acknowledgement: if this function returns without error,
    // all messages are automatically acknowledged
  },
};

async function processMessage(body: any) {
  // Process the message
  console.log('Processing:', body);
}

4. デプロイしてテストする

# Deploy producer
cd my-producer
npm run deploy

# Deploy consumer
cd my-consumer
npm run deploy

# Test by sending a message
curl -X POST https://my-producer.<your-subdomain>.workers.dev/send \
  -H "Content-Type: application/json" \
  -d '{"userId": "123", "action": "welcome-email"}'

# Watch consumer logs
npx wrangler tail my-consumer

---

プロデューサー API の全容

send() - 単一メッセージを送信

interface QueueSendOptions {
  delaySeconds?: number;  // Delay delivery (0-43200 seconds / 12 hours)
}

await env.MY_QUEUE.send(body: any, options?: QueueSendOptions);

例:

// Simple send
await env.MY_QUEUE.send({ userId: '123', action: 'send-email' });

// Send with delay (10 minutes)
await env.MY_QUEUE.send(
  { userId: '123', action: 'reminder' },
  { delaySeconds: 600 }
);

// Send structured data
await env.MY_QUEUE.send({
  type: 'order-confirmation',
  orderId: 'ORD-123',
  email: '[email protected]',
  items: [{ sku: 'ITEM-1', quantity: 2 }],
  total: 49.99,
  timestamp: Date.now(),
});

重要:

• メッセージの本文は JSON にシリアライズ可能である必要があります（構造化クローンアルゴリズム）
• メッセージの最大サイズ: 128 KB（約 100 バイトの内部メタデータを含む）
• 128 KB を超えるメッセージは失敗します。分割するか、R2 に保存して参照を送信してください

---

sendBatch() - 複数のメッセージを送信

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;  // Default delay for all messages
}

await env.MY_QUEUE.sendBatch(
  messages: Iterable<MessageSendRequest>,
  options?: QueueSendBatchOptions
);

例:

// Send batch of messages
await env.MY_QUEUE.sendBatch([
  { body: { userId: '1', action: 'email' } },
  { body: { userId: '2', action: 'email' } },
  { body: { userId: '3', action: 'email' } },
]);

// Send batch with individual delays
await env.MY_QUEUE.sendBatch([
  { body: { task: 'task1' }, delaySeconds: 60 },   // 1 min
  { body: { task: 'task2' }, delaySeconds: 300 },  // 5 min
  { body: { task: 'task3' }, delaySeconds: 600 },  // 10 min
]);

// Send batch with default delay
await env.MY_QUEUE.sendBatch(
  [
    { body: { task: 'task1' } },
    { body: { task: 'task2' } },
  ],
  { delaySeconds: 3600 } // All delayed by 1 hour
);

// Dynamic batch from array
const tasks = await getTasks();
await env.MY_QUEUE.sendBatch(
  tasks.map((task) => ({
    body: {
      taskId: task.id,
      userId: task.userId,
      priority: task.priority,
    },
  }))
);

制限:

• 1 バッチあたり最大 100 メッセージ
• バッチ全体の最大サイズは 256 KB
• 各メッセージの個別の上限は引き続き 128 KB です

---

コンシューマー API の全容

キューハンドラー関数

export default {
  async queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void> {
    // Process messages
  },
};

パラメーター:

• batch - メッセージを含む MessageBatch オブジェクト
• env - 環境バインディング（KV、D1、R2 など）
• ctx - waitUntil()、passThroughOnException() 用の実行コンテキスト

---

MessageBatch インターフェース

interface MessageBatch<Body = unknown> {
  readonly queue: string;              // Queue name
  readonly messages: Message<Body>[];  // Array of messages
  ackAll(): void;                      // Acknowledge all messages
  retryAll(options?: QueueRetryOptions): void;  // Retry all messages
}

プロパティ:

• queue - このバッチの送信元キューの名前

  • 1 つのコンシューマーが複数のキューを処理する場合に便利です

• messages - Message オブジェクトの配列

  • 順序はベストエフォートであり、保証されません
  • 処理順序を前提にしないでください

メソッド:

• ackAll() - すべてのメッセージを配信成功としてマークします

  • ハンドラーがエラーをスローしても、これらのメッセージは再試行されません
  • すべてのメッセージを安全に処理できた場合に使用します

• retryAll(options?) - すべてのメッセージを再試行対象としてマークします

  • メッセージは即時（または遅延後に）再キューイングされます
  • max_retries の上限回数にカウントされます

---

Message インターフェース

interface Message<Body = unknown> {
  readonly id: string;          // Unique message ID
  readonly timestamp: Date;     // When message was sent
  readonly body: Body;          // Message content
  readonly attempts: number;    // Retry count (starts at 1)
  ack(): void;                  // Acknowledge this message
  retry(options?: QueueRetryOptions): void;  // Retry this message
}

プロパティ:

• id - システムが生成する一意の ID（UUID）
• timestamp - メッセージがキューに送信された日時を示す Date オブジェクト
• body - メッセージの内容（JSON にシリアライズ可能な任意の型）
• attempts - コンシューマーがこのメッセージを処理した回数
  • 初回の配信時は 1 から始まります
  • 再試行のたびに増加します
  • 指数バックオフに使用できます: delaySeconds: 60 * message.attempts

メソッド:

• ack() - メッセージを配信成功としてマークします

  • 後でハンドラーが失敗しても、メッセージは再試行されません
  • べき等ではない操作（DB 書き込み、API 呼び出し）では重要です

• retry(options?) - メッセージを再試行対象としてマークします

  • 即時、または delaySeconds 後に再キューイングされます
  • max_retries（デフォルトは 3）にカウントされます

---

QueueRetryOptions

interface QueueRetryOptions {
  delaySeconds?: number;  // Delay retry (0-43200 seconds / 12 hours)
}

例:

// Retry immediately
message.retry();

// Retry after 5 minutes
message.retry({ delaySeconds: 300 });

// Exponential backoff based on attempts
message.retry({
  delaySeconds: Math.min(60 * Math.pow(2, message.attempts - 1), 3600),
});

---

コンシューマーパターン

1. 基本的なコンシューマー（暗黙的な確認応答）

最適な用途: 再試行しても安全な、べき等な操作

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Process message
      await sendEmail(message.body.email, message.body.content);
    }

    // Implicit ack: returning successfully acknowledges ALL messages
  },
};

動作:

• 関数が正常に戻ると、すべてのメッセージが確認応答されます
• 関数がエラーをスローすると、すべてのメッセージが再試行されます
• シンプルですが、一部の処理が失敗した場合に重複処理が発生する可能性があります

---

2. 明示的な確認応答（べき等ではない操作）

最適な用途: データベースへの書き込み、API 呼び出し、金融取引

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

        // Explicitly acknowledge success
        message.ack();
      } catch (error) {
        console.error(`Failed to process ${message.id}:`, error);

        // Don't ack - will retry (or let it fail)
        // Optionally: message.retry() for explicit retry
      }
    }
  },
};

明示的な確認応答が必要な理由:

• バッチ内の 1 つのメッセージが失敗しても、書き込みの重複を防ぎます
• 正常に処理されたメッセージのみが確認応答されます
• 失敗したメッセージは個別に再試行されます

---

3. 指数バックオフによる再試行

最適な用途: レート制限のある API、一時的な障害

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      try {
        // Call rate-limited API
        await fetch('https://api.example.com/process', {
          method: 'POST',
          body: JSON.stringify(message.body),
        });

        message.ack();
      } catch (error) {
        if (error.status === 429) {
          // Rate limited - retry with exponential backoff
          const delaySeconds = Math.min(
            60 * Math.pow(2, message.attempts - 1), // 60s, 120s, 240s, ...
            3600 // Max 1 hour
          );

          console.log(`Rate limited. Retrying in ${delaySeconds}s (attempt ${message.attempts})`);
          message.retry({ delaySeconds });
        } else {
          // Other error - retry immediately
          message.retry();
        }
      }
    }
  },
};

---

4. デッドレターキュー（DLQ）パターン

最適な用途： 恒久的に失敗したメッセージの処理

DLQ のセットアップ：

# Create DLQ
npx wrangler queues create my-dlq

# Configure consumer with DLQ

wrangler.jsonc：

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 10,
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // Failed messages go here
      }
    ]
  }
}

DLQ コンシューマー：

// Consumer for dead letter queue
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    for (const message of batch.messages) {
      // Log failed message
      console.error('PERMANENTLY FAILED MESSAGE:', {
        id: message.id,
        attempts: message.attempts,
        body: message.body,
        timestamp: message.timestamp,
      });

      // Store in database for manual review
      await env.DB.prepare(
        'INSERT INTO failed_messages (id, body, attempts, failed_at) VALUES (?, ?, ?, ?)'
      ).bind(
        message.id,
        JSON.stringify(message.body),
        message.attempts,
        new Date().toISOString()
      ).run();

      // Optionally: send alert to ops team
      await sendAlert({
        type: 'queue-dlq',
        messageId: message.id,
        queue: batch.queue,
      });

      // Acknowledge to remove from DLQ
      message.ack();
    }
  },
};

仕組み：

1. メインキューでメッセージが失敗します
2. max_retries（デフォルトは 3）まで再試行します
3. 最大再試行回数に達すると、DLQ に送信されます
4. DLQ コンシューマーが失敗したメッセージを処理します
5. DLQ がない場合、メッセージは完全に削除されます

---

5. 複数のキュー、単一のコンシューマー

最適な用途： 処理ロジックの一元化

export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Switch based on queue name
    switch (batch.queue) {
      case 'high-priority-queue':
        await processHighPriority(batch.messages, env);
        break;

      case 'low-priority-queue':
        await processLowPriority(batch.messages, env);
        break;

      case 'email-queue':
        await processEmails(batch.messages, env);
        break;

      default:
        console.warn(`Unknown queue: ${batch.queue}`);
        // Log to DLQ or monitoring
    }
  },
};

async function processHighPriority(messages: Message[], env: Env) {
  for (const message of messages) {
    // Process with urgency
    await fastProcess(message.body);
    message.ack();
  }
}

async function processLowPriority(messages: Message[], env: Env) {
  for (const message of messages) {
    // Can take longer
    await slowProcess(message.body);
    message.ack();
  }
}

wrangler.jsonc：

{
  "queues": {
    "consumers": [
      { "queue": "high-priority-queue" },
      { "queue": "low-priority-queue" },
      { "queue": "email-queue" }
    ]
  }
}

---

コンシューマーの設定

バッチ設定

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_batch_size": 100,        // 1-100 messages (default: 10)
        "max_batch_timeout": 30       // 0-60 seconds (default: 5)
      }
    ]
  }
}

バッチ処理の仕組み：

• 次のいずれかの条件を満たすと、コンシューマーが呼び出されます。
  • max_batch_size 件のメッセージが蓄積されたとき
  • max_batch_timeout 秒が経過したとき（先に発生した方）

例：

• max_batch_size: 100、max_batch_timeout: 10
• 3 秒で 100 件のメッセージが到着した場合 → バッチがすぐに配信されます
• 10 秒で 50 件のメッセージしか到着しなかった場合 → 50 件のバッチが配信されます

調整のガイドライン：

• 高ボリューム、低レイテンシー → max_batch_size: 100、max_batch_timeout: 1
• 低ボリューム、バッチ書き込み → max_batch_size: 50、max_batch_timeout: 30
• コストの最適化 → バッチサイズを大きくすると、呼び出し回数が減ります

---

再試行の設定

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 5,             // 0-100 (default: 3)
        "retry_delay": 300            // Seconds (default: 0)
      }
    ]
  }
}

max_retries：

• 失敗したメッセージを再試行する回数です
• 最大再試行回数に達した後：
  • DLQ がある場合 → メッセージは DLQ に送信されます
  • DLQ がない場合 → メッセージは完全に削除されます

retry_delay：

• 再試行されるすべてのメッセージに適用されるデフォルトの遅延時間（秒）です
• message.retry({ delaySeconds }) で上書きできます
• 最大遅延時間：43200 秒（12 時間）

---

同時実行の設定

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_concurrency": 10         // 1-250 (default: auto-scale)
      }
    ]
  }
}

同時実行の仕組み：

• キューはバックログに応じてコンシューマーを自動的にスケールします
• デフォルト：最大 250 件の同時呼び出しまでスケールします
• max_concurrency を設定すると、スケーリングが制限されます

max_concurrency を設定する場合：

• ✅ 上流 API にレート制限がある
• ✅ データベース接続数に制限がある
• ✅ コストを管理したい
• ❌ ほとんどの場合：最高のパフォーマンスを得るため、設定しないでください

自動スケーリングのトリガー：

• バックログの増加（メッセージの蓄積）
• 高いエラー率
• 処理速度とメッセージ受信速度の差

---

デッドレターキュー

{
  "queues": {
    "consumers": [
      {
        "queue": "my-queue",
        "max_retries": 3,
        "dead_letter_queue": "my-dlq"  // Name of DLQ
      }
    ]
  }
}

重要：

• DLQ は別途作成する必要があります：npx wrangler queues create my-dlq
• DLQ がない場合、失敗したメッセージは完全に削除されます
• DLQ 内のメッセージは、コンシューマーがいない場合でも 4 日間保持されます
• 本番環境のキューでは、必ず DLQ を設定してください

---

Wrangler コマンド

キューの作成

# Create queue with defaults
npx wrangler queues create my-queue

# Create with custom retention (4 days default, max 14 days)
npx wrangler queues create my-queue --message-retention-period-secs 1209600

# Create with delivery delay
npx wrangler queues create my-queue --delivery-delay-secs 60

---

キューの一覧表示

npx wrangler queues list

---

キュー情報の取得

npx wrangler queues info my-queue

# Output shows:
# - Queue name
# - Message retention period
# - Delivery delay
# - Consumers
# - Backlog size

---

キューの更新

# Update message retention (60 seconds to 14 days)
npx wrangler queues update my-queue --message-retention-period-secs 604800

# Update delivery delay (0 to 12 hours)
npx wrangler queues update my-queue --delivery-delay-secs 3600

---

キューの削除

npx wrangler queues delete my-queue

# Warning: This deletes ALL messages in the queue!
# Use with caution in production

---

コンシューマーの管理

# Add consumer to queue
npx wrangler queues consumer add my-queue my-consumer-worker \
  --batch-size 50 \
  --batch-timeout 10 \
  --message-retries 5 \
  --max-concurrency 20 \
  --retry-delay-secs 300

# Remove consumer from queue
npx wrangler queues consumer remove my-queue my-consumer-worker

---

キューの消去

# Delete ALL messages in queue (use with caution!)
npx wrangler queues purge my-queue

---

配信の一時停止／再開

# Pause message delivery to consumers
npx wrangler queues pause-delivery my-queue

# Resume message delivery
npx wrangler queues resume-delivery my-queue

使用例：

• コンシューマー Worker のメンテナンス
• 処理の一時停止
• メッセージを失わずに問題をデバッグ

---

制限とクォータ

機能	制限
アカウントあたりのキュー数	10,000
メッセージサイズ	128 KB（約 100 バイトのメタデータを含む）
メッセージの再試行回数	最大 100
バッチサイズ	1～100 件のメッセージ
バッチタイムアウト	0～60 秒
sendBatch あたりのメッセージ数	100 件（または合計 256 KB）
キューのスループット	キューあたり毎秒 5,000 件のメッセージ
メッセージの保持期間	4 日（デフォルト）、14 日（最大）
キューのバックログサイズ	キューあたり 25 GB
同時実行コンシューマー数	250（プッシュ方式、自動スケーリング）
コンシューマーの実行時間	15 分（実時間）
コンシューマーの CPU 時間	30 秒（デフォルト）、5 分（最大）
可視性タイムアウト	12 時間（プル方式のコンシューマー）
メッセージの遅延時間	12 時間（最大）
API レート制限	1200 リクエスト / 5 分

---

料金

Workers Paid プランが必要です（$5/月）

オペレーション料金：

• 月間最初の1,000,000オペレーション：無料
• それ以降：100万オペレーションあたり$0.40

オペレーションとしてカウントされるもの：

• 書き込み、読み取り、削除される64 KBごとのチャンク
• 64 KBを超えるメッセージは複数のオペレーションとしてカウントされます：
  • 65 KBのメッセージ＝2オペレーション
  • 127 KBのメッセージ＝2オペレーション
  • 128 KBのメッセージ＝2オペレーション

一般的なメッセージのライフサイクル：

• 書き込み1回＋読み取り1回＋削除1回＝3オペレーション

再試行：

• 再試行1回ごとに読み取りオペレーションが追加されます
• メッセージを3回再試行した場合＝書き込み1回＋読み取り4回＋削除1回＝6オペレーション

デッドレターキュー：

• DLQへの書き込みには書き込みオペレーションが追加されます

料金例：

• 月間100万メッセージ（再試行なし）：((1M × 3) - 1M) / 1M × $0.40 = $0.80
• 月間1,000万メッセージ：((10M × 3) - 1M) / 1M × $0.40 = $11.60
• 月間1億メッセージ：((100M × 3) - 1M) / 1M × $0.40 = $119.60

---

TypeScript の型

// Queue binding (producer)
interface Queue<Body = any> {
  send(body: Body, options?: QueueSendOptions): Promise<void>;
  sendBatch(
    messages: Iterable<MessageSendRequest<Body>>,
    options?: QueueSendBatchOptions
  ): Promise<void>;
}

interface QueueSendOptions {
  delaySeconds?: number;
}

interface MessageSendRequest<Body = any> {
  body: Body;
  delaySeconds?: number;
}

interface QueueSendBatchOptions {
  delaySeconds?: number;
}

// Consumer handler
export default {
  queue(
    batch: MessageBatch,
    env: Env,
    ctx: ExecutionContext
  ): Promise<void>;
}

interface MessageBatch<Body = unknown> {
  readonly queue: string;
  readonly messages: Message<Body>[];
  ackAll(): void;
  retryAll(options?: QueueRetryOptions): void;
}

interface Message<Body = unknown> {
  readonly id: string;
  readonly timestamp: Date;
  readonly body: Body;
  readonly attempts: number;
  ack(): void;
  retry(options?: QueueRetryOptions): void;
}

interface QueueRetryOptions {
  delaySeconds?: number;
}

---

エラー処理

よくあるエラー

1. メッセージが大きすぎます

// ❌ Bad: Message >128 KB
await env.MY_QUEUE.send({
  data: largeArray, // >128 KB
});

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

if (size > 128000) {
  // Store in R2, send reference
  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. スループットの上限超過

// ❌ Bad: Exceeding 5000 msg/s per queue
for (let i = 0; i < 10000; i++) {
  await env.MY_QUEUE.send({ id: i }); // Too fast!
}

// ✅ Good: Use sendBatch
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));
}

// ✅ Even better: Rate limit with delay
for (let i = 0; i < messages.length; i += 100) {
  await env.MY_QUEUE.sendBatch(messages.slice(i, i + 100));
  if (i + 100 < messages.length) {
    await new Promise(resolve => setTimeout(resolve, 100)); // 100ms delay
  }
}

---

3. コンシューマーのタイムアウト

// ❌ Bad: Long processing without CPU limit increase
export default {
  async queue(batch: MessageBatch): Promise<void> {
    for (const message of batch.messages) {
      await processForMinutes(message.body); // CPU timeout!
    }
  },
};

// ✅ Good: Increase CPU limit in wrangler.jsonc

wrangler.jsonc：

{
  "limits": {
    "cpu_ms": 300000  // 5 minutes (max allowed)
  }
}

---

4. バックログの増加

// Issue: Consumer too slow, backlog growing

// ✅ Solution 1: Increase batch size
{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "max_batch_size": 100  // Process more per invocation
    }]
  }
}

// ✅ Solution 2: Let concurrency auto-scale (don't set max_concurrency)

// ✅ Solution 3: Optimize consumer code
export default {
  async queue(batch: MessageBatch, env: Env): Promise<void> {
    // Process in parallel
    await Promise.all(
      batch.messages.map(async (message) => {
        await process(message.body);
        message.ack();
      })
    );
  },
};

---

必ず行うこと ✅

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. キューごとに5,000 msg/sを超えない - レート制限エラーが発生します
7. ループ内でメッセージを同期的に処理しない - Promise.all() を使用する
8. message.attempts を無視しない - バックオフロジックに使用する
9. 非常に特別な理由がない限り、max_concurrency=1 に設定しない
10. 明示的な確認応答パターンでは、ack() を忘れない

---

トラブルシューティング

問題：メッセージがコンシューマーに配信されない

考えられる原因：

1. コンシューマーがデプロイされていない
2. wrangler.jsonc のキュー名が間違っている
3. 配信が一時停止されている
4. コンシューマーでエラーが発生している

解決方法：

# Check queue info
npx wrangler queues info my-queue

# Check if delivery paused
npx wrangler queues resume-delivery my-queue

# Check consumer logs
npx wrangler tail my-consumer

---

問題：1つのメッセージが失敗すると、バッチ全体が再試行される

原因： 冪等でない操作で暗黙的な確認応答を使用している

解決方法： 明示的な ack() を使用する

// ✅ Explicit ack
for (const message of batch.messages) {
  try {
    await dbWrite(message.body);
    message.ack(); // Only ack on success
  } catch (error) {
    console.error(`Failed: ${message.id}`);
    // Don't ack - will retry
  }
}

---

問題：処理されずにメッセージが削除される

原因： デッドレターキューが設定されていない

解決方法：

# Create DLQ
npx wrangler queues create my-dlq

# Add to consumer config

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      "dead_letter_queue": "my-dlq"
    }]
  }
}

---

問題：コンシューマーが自動スケーリングされない

考えられる原因：

1. max_concurrency が1に設定されている
2. コンシューマーがエラーを返している（処理していない）
3. バッチ処理が速すぎる（バックログがない）

解決方法：

{
  "queues": {
    "consumers": [{
      "queue": "my-queue",
      // Don't set max_concurrency - let it auto-scale
      "max_batch_size": 50  // Increase batch size instead
    }]
  }
}

---

本番環境へのデプロイ前チェックリスト

本番環境にデプロイする前に：

• デッドレターキューが作成され、設定されている
• 冪等でない操作に明示的な ack() を使用している
• メッセージサイズを検証している（<128 KB）
• ワークロードに合わせてバッチ設定が最適化されている
• 再試行設定（max_retries、retry_delay）が設定されている
• コンシューマーの並行数設定が適切である
• エラー処理とログ記録が実装されている
• 監視とアラートが設定されている
• DLQコンシューマーがデプロイされている
• レート制限のあるAPI向けに指数バックオフを設定している
• 必要に応じてCPU制限（limits.cpu_ms）を引き上げている
• メッセージの保持期間が適切である
• 負荷テストを実施している

---

Related Documentation

• Cloudflare Queues Docs
• How Queues Works
• JavaScript APIs
• Batching & Retries
• Consumer Concurrency
• Dead Letter Queues
• Wrangler Commands
• Limits
• Pricing

---

Last Updated: 2025-10-21 Version: 1.0.0 Maintainer: Jeremy Dawes | [email protected]