はじめに

Cloudflare Queues は、メッセージを非同期処理のためにキューへ入れられる、柔軟なメッセージングキューです。このガイドでは、最初のキュー、そのキューへメッセージを公開する Worker、そのキューからメッセージを消費するコンシューマー Worker を作成します。

前提条件

Queues を使うには、次が必要です。

Cloudflare アカウント ↗ に登録します。
Node.js ↗ をインストールします。

Node.js のバージョンマネージャー

権限の問題を避け、Node.js のバージョンを切り替えられるよう、Volta ↗ や nvm ↗ などの Node バージョンマネージャーを使います。このガイドの後半で説明する Wrangler には、Node バージョン 16.17.0 以降が必要です。

1. Worker プロジェクトを作成する

キューには、プロデューサー Worker からアクセスします。キューへメッセージを公開するには、プロデューサー Worker を少なくとも 1 つ作成します。R2 Bucket Event Notifications を使う場合は、プロデューサー Worker は不要です。

プロデューサー Worker を作成するには、次を実行します。

npm create cloudflare@latest -- producer-worker

yarn create cloudflare producer-worker

pnpm create cloudflare@latest producer-worker

セットアップでは、次のオプションを選びます。

What would you like to start with? では、Hello World example を選びます。
Which template would you like to use? では、Worker only を選びます。
Which language do you want to use? では、TypeScript を選びます。
Do you want to use git for version control? では、Yes を選びます。
Do you want to deploy your application? では、No を選びます（デプロイ前にいくつか変更します）。

新しいディレクトリが作成され、src/index.ts の Worker スクリプトと wrangler.jsonc 設定ファイルが含まれます。Worker を作成したあと、アクセス用の Queue を作成します。

作成したディレクトリへ移動します。

cd producer-worker

2. キューを作成する

キューを使うには、メッセージの公開先および消費元となるキューを少なくとも 1 つ作成します。

キューを作成するには、次を実行します。

npx wrangler queues create <MY-QUEUE-NAME>

扱うメッセージの種類が分かる、説明的な名前を選びます。例: debug-logs、user-clickstream-data、password-reset-prod。

キュー名の長さは 1〜63 文字です。ハイフン (-) 以外の特殊文字は使えません。先頭と末尾は英字または数字である必要があります。

キュー名は設定後に変更できません。キューを作成したあと、プロデューサー Worker からアクセスできるように設定します。

3. プロデューサー Worker を設定する

Worker 内のコードからキューを使えるようにするには、バインドを作成してキューを Worker に接続します。バインドを使うと、Worker は Queues など Cloudflare 開発者プラットフォーム上のリソースへアクセスできます。

バインドを作成するには、生成された wrangler.jsonc を開き、次を追加します。

{
	"queues": {
		"producers": [
			{
				"queue": "MY-QUEUE-NAME",
				"binding": "MY_QUEUE"
			}
		]
	}
}

[[queues.producers]]
queue = "MY-QUEUE-NAME"
binding = "MY_QUEUE"

MY-QUEUE-NAME を手順 2 で作成したキュー名に置き換えます。次に、MY_QUEUE を使いたい binding 名に置き換えます。バインド名は有効な JavaScript 変数名である必要があります。Worker 内でこのキューを参照する変数になります。

プロデューサー Worker を書く

プロデューサー Worker がキューへ公開するメッセージを作るように設定します。このプロデューサー Worker は次を行います。

ブラウザーから受け取ったリクエストを取ります。
リクエストを JSON 形式に変換します。
リクエストをキューへ直接書き込みます。

Worker プロジェクトのディレクトリで src フォルダーを開き、index.ts に次を追加します。

export default {
  async fetch(request, env, ctx): Promise<Response> {
    const log = {
      url: request.url,
      method: request.method,
      headers: Object.fromEntries(request.headers),
    };
    await env.<MY_QUEUE>.send(log);
    return new Response("Success!");
  },
} satisfies ExportedHandler<Env>;

MY_QUEUE を、wrangler.jsonc で設定したバインド名に置き換えます。

あわせて、index.ts の Env インターフェイスにキューを追加します。

export interface Env {
   <MY_QUEUE>: Queue;
}

この書き込みが失敗すると、Worker はエラーを返します（例外を送出します）。成功すると、ブラウザーへ HTTP 200 ステータスコードとともに Success を返します。

本番アプリケーションでは、例外を捕捉して直接扱うために try...catch ↗ を使うのが一般的です（カスタムエラーを返す、リトライするなど）。

プロデューサー Worker を公開する

Wrangler ファイルと index.ts の設定が終わったら、プロデューサー Worker を公開できます。公開するには、次を実行します。

npx wrangler deploy

デフォルトでは *.workers.dev URL 付きで、次のような出力が表示されます。

Uploaded <YOUR-WORKER-NAME> (0.76 sec)
Published <YOUR-WORKER-NAME> (0.29 sec)
  https://<YOUR-WORKER-NAME>.<YOUR-ACCOUNT>.workers.dev

*.workers.dev サブドメインをコピーし、新しいブラウザーのタブに貼り付けます。ページを数回更新して、キューへのリクエスト公開を始めます。リクエストをキューへ書き込むたびに、ブラウザーは Success レスポンスを返します。

キューと、そのキューへメッセージを公開するプロデューサー Worker ができました。次に、公開されたメッセージを消費するコンシューマー Worker を作成します。コンシューマー Worker がないと、メッセージは期限切れまでキューに残ります。デフォルトの保持期間は 4 日です。

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

コンシューマー Worker は、キューからメッセージを受け取ります。受け取ったメッセージは、ログコンソールやストレージオブジェクトなど、別の出力先へ書けます。

このガイドではコンシューマー Worker を作成し、wrangler tail でメッセージをログ出力して確認します。コンシューマー Worker は、プロデューサー Worker を作成した同じ Worker プロジェクトに作ります。

コンシューマー Worker を作成するには、index.ts を開き、既存の fetch ハンドラーに次の queue ハンドラーを追加します。

export default {
  async fetch(request, env, ctx): Promise<Response> {
    const log = {
      url: request.url,
      method: request.method,
      headers: Object.fromEntries(request.headers),
    };
    await env.<MY_QUEUE>.send(log);
    return new Response("Success!");
  },
  async queue(batch, env, ctx): Promise<void> {
    for (const message of batch.messages) {
      console.log("consumed from our queue:", JSON.stringify(message.body));
    }
  },
} satisfies ExportedHandler<Env>;

MY_QUEUE を、wrangler.jsonc で設定したバインド名に置き換えます。

キューへメッセージが公開されるたびに、コンシューマー Worker の queue ハンドラー（async queue）が呼び出され、1 件以上のメッセージが渡されます。

この例では、コンシューマー Worker はキューの JSON 形式メッセージを文字列に変換してログに出します。実際のアプリケーションでは、オブジェクトストレージ（R2 など）への書き込み、データベース（D1 など）への書き込み、外部 API（メール API など）を呼ぶ前の追加処理、既存クラウドプロバイダーのデータウェアハウスへの書き込みなどを設定できます。

コンシューマーハンドラー内で非同期タスクを行うときは、関数のレスポンスが確実に扱われるよう waitUntil() を使います。このメソッドのスコープ内では、ほかの非同期メソッドは使えません。

コンシューマー Worker をキューに接続する

コンシューマー Worker の設定が終わったら、キューへ接続します。

1 つのキューに接続できるコンシューマー Worker は 1 つだけです。同じキューに複数のコンシューマーを接続しようとすると、その Worker を公開するときにエラーになります。

キューをコンシューマー Worker に接続するには、Wrangler ファイルを開き、末尾に次を追加します。

{
	"queues": {
		"consumers": [
			{
				"queue": "<MY-QUEUE-NAME>",
				// Required: this should match the name of the queue you created in step 3.
				// If you misspell the name, you will receive an error when attempting to publish your Worker.
				"max_batch_size": 10, // optional: defaults to 10
				"max_batch_timeout": 5 // optional: defaults to 5 seconds
			}
		]
	}
}

[[queues.consumers]]
queue = "<MY-QUEUE-NAME>"
max_batch_size = 10
max_batch_timeout = 5

MY-QUEUE-NAME を手順 2 で作成したキューに置き換えます。

このコンシューマー Worker では、max_batch_size と max_batch_timeout でメッセージを自動バッチします。コンシューマー Worker は、10 件のバッチ、または 5 秒ごと、いずれか早い方でメッセージを受け取ります。

max_batch_size（デフォルトは 10）は、コンシューマー Worker の呼び出し回数を減らします。メッセージごとではなく、キューに 10 件入ったあとで呼び出されます。

max_batch_timeout（デフォルトは 5 秒）は待ち時間を短くします。プロデューサー Worker が 10 件まで送らず、コンシューマー Worker が呼び出されない場合でも、キューで待っているメッセージを受け取るために 5 秒ごとに呼び出されます。

コンシューマー Worker を公開する

Wrangler ファイルと index.ts の設定が終わったら、次を実行してコンシューマー Worker を公開します。

npx wrangler deploy

5. キューからメッセージを読み取る

コンシューマー Worker を設定したあと、キューからメッセージを読めます。

wrangler tail を実行して、コンシューマーが受け取るメッセージのログを待ちます。

npx wrangler tail

wrangler tail を動かしたまま、手順 3 で開いた Worker URL を開きます。

ブラウザーウィンドウに Success メッセージが表示されます。

Success が表示されたら、URL を数回更新してメッセージを生成し、キューへ送ります。

wrangler tail が動いていると、更新で発生したリクエストをコンシューマー Worker がログに出し始めます。

更新が 10 回未満の場合、バッチタイムアウトが 10 秒に設定されているため、メッセージが表示されるまで数秒かかることがあります。10 秒後には、ターミナルにメッセージが届きます。

更新時にエラーになる場合は、手順 2 で作成したキュー名と、Wrangler ファイルで参照しているキューが同じか確認します。プロデューサー Worker が Success を返し、エラーを返していないことも確認します。

このガイドを完了すると、キュー、そのキューへメッセージを公開するプロデューサー Worker、そのキューからメッセージを消費するコンシューマー Worker ができています。