Cloudflare のメッセージキューサービス Queues を使ってみた。

#Cloudflare

#Cloudflare Workers

小西諒

2022.12.23

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

Cloudflare Queues のオープンベータ版が公開されていたので触ってみました。

Cloudflare Queues とは

Cloudflare が提供するグローバルメッセージキューサービスで、メッセージの配信とバッチ処理を提供します。

Cloudflare Queues は Cloudflare Workers と統合されており、後述するように Workers からメッセージの送信と取得を操作します。

例えば複数の Workers 間でデータのやり取りが必要な場合、負荷に応じたバッファリングやバッチ処理、それによる処理の確実性の向上、またサービスコンポーネントを分離する役割などが期待できます。

エグレスやスループットに対しては料金が発生せず、オペレーション（キューの読み書き）の合計数に応じて課金されます。

Cloudflare Queues を触ってみる

ざっくり利用の流れです。

Queues の有効化と作成
Wrangler のインストール
Worker プロジェクトの作成
データプロデューサー側の Worker の作成
データコンシューマー側の Worker の作成

1. Queues の有効化

Cloudflare ダッシュボードから新規キューを作成します。

※まだ Queues 有効化前であれば、 [Enable] のようなボタンをクリック

Cloudflare Queue

[sample-queue] という名前で作成してみました。いったん [inactive] ステータスのままで大丈夫です。

Cloudflare Queue

2. Wrangler のインストール

Wrangler は Workers をローカルから開発/デプロイする CLI ツールです。

Queue を利用するには Wrangler CLI のバージョン 2.2.1 以降がインストールされている必要があります。

▼未インストールの方

$ npm install -g wrangler
もしくは
$ yarn global add wrangler

▼既にインストール済みの方

$ wrangler --version
2.2.1

上記でバージョンが 2.2.1 未満であれば…

$ npm update -g wrangler

アカウント認証のためにログインします。

$ wrangler login

ブラウザが開き、認証が完了したら下記のような画面になります。

Cloudflare Queue

3. Worker プロジェクトの作成

サンプルのプロジェクトを作成します。.git を作成するかなど色々聞かれますが、全部 y (Yes) で問題ないです。

$ wrangler init sample-worker-for-queues
✨ Initialized git repository at sample-worker-for-queues
No package.json found. Would you like to create one? (y/n)
✨ Created sample-worker-for-queues/package.json
Would you like to use TypeScript? (y/n)
✨ Created sample-worker-for-queues/tsconfig.json
Would you like to create a Worker at sample-worker-for-queues/src/index.ts?
  None
❯ Fetch handler
  Scheduled handler

Workers から Queues を操作するため、Cloudflare においてバイディングとよばれる設定が必要です。これは特定のキューを Worker に接続して呼び出せるようにするための操作です。

先ほどの init で作成されたディレクトリ sample-worker-for-queues の直下に wrangler.toml があるはずですので、そちらに下記を追記します。

[[queues.producers]]
  queue = "sample-queue(先ほど作成したキューの名前)" 
  binding = "SAMPLE_QUEUE"

[[queues.consumers]]
  queue = "sample-queue"

4. データプロデューサー側の Worker の作成

プロデューサーとは、キューにメッセージを発行するクライアントを指します。

今回は試しにタイムスタンプをキューに送信するサンプルを作成してみます。

src/index.ts を書き換えます。

export interface Env {
    SAMPLE_QUEUE: Queue,
}

export default {
    async fetch(
        request: Request,
        env: Env,
        ctx: ExecutionContext
    ): Promise<Response> {
        const timestamp_now = Date.now().toString()
        await env.SAMPLE_QUEUE.send({timestamp: timestamp_now});
        return new Response(timestamp_now);
    },
};

バインディング上で [キュー名].send() を呼び出すことで、該当のキューにメッセージを書き込めます。

なお、1つのキューは複数のプロデューサーを持つことができます。1つのキューに書き込めるプロデューサー（= Worker）の総数に制限はありません。

逆に、複数のキューを単一の Worker にバインドすることも可能とのことです。

またメッセージの中身については、シリアライズ可能なオブジェクトなら何でも発行できるらしく（Any serializable object can be published to a queue）、上記の例だとシンプルな JSON を渡しています。

Cloudflare に Workers を公開します（※現状 wrangler dev では Queues は動きません）。

$ wrangler publish

ブラウザで Workers の URL を開いて、タイムスタンプが表示されていたらOKです。

なお、デバッグには wrangler tail が便利です↓ https://developers.cloudflare.com/workers/wrangler/commands/#tail

5. データコンシューマー側の Worker の作成

先ほどはキューにメッセージを書き込む処理でしたので、次に書き込まれたメッセージを取得するコンシューマーのコードを追加します。

export interface Env {
	SAMPLE_QUEUE: Queue,
}

export default {
	async fetch(
		request: Request,
		env: Env,
		ctx: ExecutionContext
	): Promise {
		const timestamp_now = Date.now().toString()
		await env.SAMPLE_QUEUE.send({sample_timestamp: timestamp_now});
		return new Response(timestamp_now);
	},
	async queue(
		batch: MessageBatch,
		env: Env,
		ctx: ExecutionContext
	): Promise {
		let messages = JSON.stringify(batch.messages)
		console.log("Received", messages)
	}
};

先ほどのプロデューサーと異なり、各キューにつきアクティブなコンシューマーは1つまでになっており、これにより「少なくとも1回の配信」の要件を実現し、重複メッセージのリスクを最小化しているとのことです。

逆に、1つのコンシューマーからは複数のキューを使用できます。

上記の Worker を実行して $ wrangler tail でログを見るとこんな感じ↓（各メッセージは自動でIDとタイムスタンプを持つので、明示的に timestamp を持たせた今回の例は少し分かりづらかった、すみません…）

(log) Received [{"body":{"sample_timestamp":"1671731104378"},"timestamp":"2022-12-22T17:45:04.466Z","id":"e10ffb69c9456d970ef53f4a9a2ddcc9"},{"body":{"sample_timestamp":"1671731104700"},"timestamp":"2022-12-22T17:45:04.729Z","id":"eccc41d2d3780e57e75eb8ec1d548dfa"}]

以上、簡単にメッセージの配信と取得を行ってみました。

Queues のオプション設定

wrangler.toml のコンシューマーからバッチ処理の設定が可能です。

[[queues.consumers]]
  queue = "sample-queue"
  max_batch_size = 10
  max_batch_timeout = 30
  max_retries = 10
  dead_letter_queue = "sample-queue-dlq"

下記はいずれもオプショナルな設定です。

max_batch_size … 各バッチで許可されるメッセージの最大数（最大100）。
max_batch_timeout … バッチが一杯になるまでの最大待ち時間の秒数（最大30秒）。
max_retries … 何度まで再試行するかの回数（最大100）。
dead_letter_queue … 後述

デッドレターキューの処理

Queues でもデッドレターキューの処理が可能です。wrangler.toml からオプションとして設定し、処理が少なくとも max_retries 回失敗した場合にメッセージを送信する別の Queue の名前を指定できます。指定された名前の Queue がまだ存在しない場合は自動的に作成されます。このオプションが定義されていない場合、処理に何度も失敗するメッセージは最終的に廃棄されます。