【ITニュース解説】How to Build a Job Queue in Go Without External Dependencies

システム開発では、ユーザーからのリクエストに直接関係しないけれど、後で実行する必要がある処理が頻繁に発生する。例えば、新しいユーザーが登録された後にウェルカムメールを送ったり、アップロードされた画像をバックグラウンドで処理したり、毎日のレポートを生成したりといった作業だ。このような処理は、ウェブサイトの表示など、ユーザーがすぐに結果を期待する主要な処理とは切り離して、裏側で実行されることが望ましい。これを「バックグラウンドジョブ処理」と呼ぶ。

従来、このようなバックグラウンド処理を行う際には、RedisやRabbitMQといった「外部のメッセージキュー」と呼ばれる専用のシステムを使うのが一般的だった。これらは、処理すべきタスクを一時的に保管し、別のプログラム（ワーカー）がそれを取り出して実行するという仕組みを提供してくれる。しかし、全てのアプリケーションにとって、これらの外部システムを導入・運用することは、過剰な手間やコストになる場合がある。Go言語を使っている場合であれば、Go自体が持つ強力な「並行処理」の機能だけで、多くのバックグラウンド処理をシンプルに実現できる。本記事では、外部のシステムに頼らず、Goだけでジョブキューを構築する方法について解説する。

なぜ外部キューを使わない選択肢があるのか、その理由から説明しよう。外部キューは、例えばサーバーが再起動してもタスクが失われないようにしたい場合、複数のサーバーでワーカーを分散して動かしたい場合、あるいはタスクが確実に一度だけ実行されたことを確認したい場合に、非常に優れた解決策となる。しかし、もしあなたのアプリケーションが単一のサーバーで動作しており、万が一プログラムが予期せず終了しても、キューに入っていた未処理のタスクが失われても問題ない、または再試行できる性質のものであれば、Goの「チャネル」という機能を使った内部的なジョブキューで十分事足りるケースが多いのだ。

このGoのチャネルベースのジョブキューの最大のメリットは、「追加で何も設定する必要がない」という点だ。専用のサーバーを立てたり、データベースを設定したりする手間がなく、アプリケーションの実行ファイル一つで全てが完結するため、デプロイ（配置・公開）も非常にシンプルになる。また、タスクをキューに入れる際も、ネットワークを介した通信が発生しないため、処理が高速だ。Go言語の型システムがそのまま利用できるため、ジョブの型安全性が保たれるのも大きな利点だ。ただし、注意すべき欠点もある。このキューはプログラムが実行されているメモリ上に存在するため、もしアプリケーションがクラッシュしたり、再起動したりすると、まだ処理されていないタスクは全て失われてしまう。しかし、メール通知の送信や一時的なユーザーリクエスト処理など、多くのバックグラウンドワークロードでは、この「失われる可能性」は許容できる範囲のトレードオフとなる。

Goでジョブキューを構築するために必要な要素は主に三つある。一つ目は「ジョブ」、つまり実行したい具体的な処理だ。二つ目は、このジョブを順番に受け渡すための「チャネル」。そして三つ目は、チャネルからジョブを受け取って実際に処理を実行する「ワーカー」だ。

コードでは、まずJob型がfunc(ctx context.Context) errorという関数型として定義されている。これは、実行したい処理を特定のインターフェースを実装した構造体として定義するのではなく、直接関数として記述できるため、非常にシンプルで柔軟だ。ctx context.Contextは、Goの標準機能で、処理のキャンセルやタイムアウトを伝えるためのものだ。これにより、ワーカーがジョブを実行中にアプリケーションが停止する命令を受け取った場合などに、安全に処理を中断できるようになる。

次に、ジョブキュー全体を管理するQueueという構造体がある。この構造体は、jobsという名前のチャネル（chan Job型）と、起動するワーカーの数を示すworkers、そして起動したワーカーの全ゴルーチンが終了するのを待つためのsync.WaitGroupというGoの並行処理の補助機能を持っている。jobsチャネルは、make(chan Job, bufferSize)のように作成され、bufferSize分のジョブを一時的に蓄えることができる「バッファ付きチャネル」として機能する。これにより、ジョブをキューに入れる側が、ワーカーの処理速度に関わらず、ある程度の数のジョブをチャネルにためておくことが可能になる。

New関数は、指定されたワーカー数とバッファサイズでQueueの新しいインスタンスを初期化する。そして、Startメソッドが呼ばれると、指定されたワーカー数分の「ゴルーチン」を起動する。ゴルーチンはGoにおける軽量な並行処理の単位で、多数の処理を同時に実行するのに非常に適している。それぞれのゴルーチンはworkerメソッドを実行し、これが実際にジョブを処理するワーカーとして動作する。

workerメソッドの中では、無限ループでジョブチャネルq.jobsから新しいジョブが送られてくるのを待ち続ける。Goのselect文を使って、ジョブがチャネルに送られてくるか、またはコンテキストがキャンセルされるか（ctx.Done()）、どちらかのイベントを監視する。ジョブが送られてくればそれを実行し、もしエラーが発生した場合はログに出力する。もしチャネルが閉じられたり、コンテキストがキャンセルされたりすれば、ワーカーは安全にループを終了し、処理を終える。defer q.wg.Done()は、ワーカーのゴルーチンが終了する直前にWaitGroupのカウントを減らし、Queue.Stopメソッドが全てのワーカーの終了を正確に待てるようにするために使われる。

ジョブをキューに追加するのはEnqueueメソッドの役割だ。ここでもselect文が使われるが、特徴的なのは、チャネルにジョブを追加できた場合はtrueを返し、もしチャネルがすでに満杯でジョブを追加できない場合はdefaultケースに入り、ジョブを追加しようとブロックせずに即座にfalseを返す点だ。これにより、ジョブを追加する側が、キューが満杯だった場合にどう対処するか（例えば、ユーザーに「サーバーが混雑しています」と伝えるHTTP 429エラーを返すなど）を柔軟に決定できる。

最後に、Stopメソッドは、まずジョブチャネルq.jobsをcloseし、その後q.wg.Wait()を呼び出して、起動中の全てのワーカーゴルーチンがそれぞれの処理を終えるのを待つ。これにより、キューに残っていたジョブが安全に処理され、ワーカーがきちんと終了することが保証される。

このジョブキューを実際のHTTPサーバーに組み込む例を見てみよう。ユーザーが/registerというウェブアドレスにアクセスし、メールアドレスを送信すると、そのメールアドレスにウェルカムメールを送るという処理をバックグラウンドで行うシナリオだ。HTTPリクエストを受け取ったハンドラー関数内でq.Enqueueを呼び出し、メール送信処理を定義した関数をジョブとしてキューに追加する。もしキューが満杯でEnqueueがfalseを返した場合、ユーザーには「サーバーが混雑しています」というメッセージと共にHTTP 429エラーを返すといった処理が可能になる。

ここで特に重要なのは、アプリケーションを停止する際の「グレースフルシャットダウン」という、穏やかに停止させる手順だ。正しくアプリケーションを停止するには、まず新しいHTTPリクエストの受け付けを停止する。これは、HTTPサーバーのShutdownメソッドを使うことで実現できる。その後に、ジョブキューのStopメソッドを呼び出し、キュー内の未処理ジョブが全て完了するのを待つ。この順序を逆にすると、キューを閉じた後に新しいジョブがHTTPサーバー経由で追加されてしまい、閉じられたチャネルにデータを送ろうとしてプログラムが異常終了（パニック）する可能性があるため、この停止順序は非常に重要となる。

実際の運用では、バックグラウンドジョブが常に成功するとは限らない。ネットワークの一時的な問題や、外部サービスの不具合などで、一時的に失敗することがある。このような場合に対応するために「リトライロジック」（再試行の仕組み）を組み込むことを検討する。記事では、ワーカーの内部にリトライロジックを組み込むのではなく、ジョブをキューに入れる際に、リトライ機能を備えたラッパー関数で元のジョブを包み込む方法が提案されている。withRetry関数は、指定された回数だけジョブの実行を試み、失敗した場合は、時間間隔を徐々に延ばしながら再試行する（リニアバックオフ）。この際、待機中にシステムからのキャンセル命令があれば即座に中断し、安全にシャットダウンできるよう、time.Afterとctx.Done()をselect文で組み合わせるという工夫が施されている。

このシンプルなチャネルベースのジョブキューは非常に強力だが、限界も理解しておく必要がある。最も大きな制約は、すでに述べた「永続性がない」点だ。もしプロセスがクラッシュしたり、デプロイのために再起動したりすると、キューに入っている未処理のジョブは全て失われる。もしこれが許容できない場合は、Redisやデータベースのような永続的なストレージにジョブを保存する仕組みが必要となる。

また、このキューには「内部の可視性」がほとんどない。現在どれくらいのジョブがキューに残っているのか、どのジョブが処理中で、どのジョブが失敗したのかといった詳細な情報を、標準では把握できない。必要であれば、別途カウンターを追加してメトリクスとして公開するなどの実装が必要になるだろう。

さらに、「優先順位付け」の機能もない。単一のチャネルは基本的にFIFO（先入れ先出し）でジョブを処理するため、緊急性の高いジョブを優先的に処理したいといった要件には対応できない。このような場合は、複数のチャネルを優先度別に用意し、高い優先度のチャネルから先にジョブを取り出すようなディスパッチャーを実装する必要がある。

最後に、キューの「メモリはバッファサイズによって制限される」という点も重要だ。キューが常に満杯の状態であれば、ワーカーの処理速度が遅すぎるか、または処理能力を超えるトラフィックが来ていることを示唆している。この場合は、ワーカー数を増やすか、より大規模な分散型ソリューションへの移行を検討する必要がある。

まとめると、Goのチャネルを使ったジョブキューは、多くの非同期処理のニーズに対して、極めてシンプルな実装で応えることができる。運用上のオーバーヘッドがゼロであり、短いコードで安全なシャットダウンやリトライ機能を容易に追加できる。永続性、分散ワーカー、高度なジョブ監視が必要になるまでは、あえてRedisのような外部の複雑なソリューションに飛びつく必要はない。まずは「最もシンプルな動くものから始める」という考え方が、Goの並行処理機能を使う上で非常に有効であり、バッファ付きチャネルといくつかのゴルーチンは、そのシンプルさを体現していると言えるだろう。