【ITニュース解説】How to build a job queue in Go without external dependencies

ITシステムを開発する際、時には時間のかかる処理や、メインのプログラムの流れとは別に実行したい処理が出てくることがある。例えば、ユーザーが会員登録した際に確認メールを送信したり、大量の画像をアップロードされたときに裏でサイズ変更したり、日次でレポートを生成したりするような作業だ。これらの処理をメインのプログラムが直接担当すると、ユーザーの操作が一時的に停止したり、システム全体の応答速度が低下したりする可能性がある。このような問題を解決するために、「ジョブキュー」という仕組みが用いられる。

ジョブキューとは、実行したい処理（ジョブ）を一旦リスト（キュー）に貯めておき、別の場所で専用のプログラム（ワーカー）が順番に取り出して実行する仕組みを指す。これにより、時間のかかる処理がメインのプログラムの邪魔をすることなく、システム全体のパフォーマンスを保ちながら効率的にタスクを消化できる。

一般的にジョブキューを構築する際には、RedisやRabbitMQといった専用のメッセージブローカーやマネージドサービスを利用することが多い。これらは非常に強力で、大規模なシステムや高い信頼性が求められる環境では不可欠なツールとなる。しかし、もしあなたが社内向けの小さなツールや、サイドプロジェクト、あるいは特定の機能に特化したマイクロサービスを開発していて、非同期処理が必要なだけであれば、こうした外部サービスを導入するのは「やりすぎ」となる場合がある。外部サービスの導入には、その設定、運用、保守といった追加のコストや複雑さが伴うからだ。

ここで紹介する記事は、Go言語が持つ標準機能だけで、外部の依存関係を一切持たずに、高機能なジョブキューを構築する方法を提案している。Go言語は、並行処理を簡単に記述できる「goroutine（ゴルーチン）」や、goroutine間で安全にデータをやり取りできる「channel（チャネル）」、複数の処理の同期を取るための「syncパッケージ」、そして処理のキャンセルを管理する「context（コンテキスト）」といった強力な標準ライブラリを備えている。これらを活用すれば、わずか約150行程度のコードで、外部ライブラリに頼ることなく、実用的なジョブキューを実現できる。

このジョブキューは、以下のような特徴を持つ。まず、ジョブを処理する「ワーカー」と呼ばれるプログラムの数が固定されており、これらのワーカーがバッファ（一時的な保管場所）付きのチャネルからジョブを順番に引き出して処理する。ジョブをキューに追加する際（エンキュー）には、キューが満杯の場合にはすぐにエラーを返し、呼び出し元のプログラムが処理を待って止まってしまうことはない（ノンブロッキングなエンキュー）。また、ジョブの処理中にエラーが発生した場合でも、設定された回数まで自動的にジョブを再試行する機能が備わっている。さらに、システムが終了する際に、OSからの停止指示（SIGINTやSIGTERMといったシグナル）を安全に受け取り、現在処理中のジョブが途中で中断されることなく、全て完了してからプログラムを終了する「グレースフルシャットダウン」にも対応している。ワーカーがジョブの処理に失敗した際のエラーは、専用のエラーチャネルを通じて通知されるため、ワーカーの障害がプログラム全体を停止させることも防げる。

核となる構造は、まずJobという構造体で、ジョブの識別子（ID）、実際の処理内容（Payload）、現在の再試行回数（Retries）、最大再試行回数（MaxRetry）を定義している。次に、Handlerという型で、実際にジョブを処理する関数の形式を定めている。そして、Queueというメインの構造体が、ジョブを格納するチャネル、ワーカーたちの完了を待つsync.WaitGroup、ジョブを処理するHandler関数、そしてエラーを通知するためのエラーチャネルを持っている。

キューを初期化するNew関数では、指定されたバッファサイズとワーカー数に基づいてチャネルを生成し、指定数のワーカー（goroutine）を起動する。Enqueueメソッドでジョブをキューに追加するが、ここでGoのselect文とdefaultキーワードを組み合わせることで、チャネルが満杯でジョブを受け入れられない場合に、即座にエラーを返すノンブロッキングな動作を実現している。これにより、ジョブを追加する側のプログラムは、キューの混雑状況に応じて適切な対応（例えば、後で再試行したり、ユーザーにエラーを通知したり）を自身で決められる。

workerメソッドは、各ワーカー（goroutine）が実行する処理だ。このワーカーはjobsチャネルからジョブが来るのを常に待ち、ジョブが来たらHandler関数を使ってそのジョブを処理する。もし処理中にエラーが発生し、まだ最大再試行回数に達していなければ、ワーカーはジョブの再試行回数を増やして、同じジョブをjobsチャネルに再び投入する。これにより、別のワーカー、あるいは同じワーカーがそのジョブを再度処理しようとする。もし再試行回数を超えてもジョブが失敗した場合、そのエラーはerrorsチャネルを通じて外部に通知される。ワーカーは処理を終えると、defer q.wg.Done()によってsync.WaitGroupに完了を通知する。

Stopメソッドはキューを安全に停止させるための重要な機能だ。まずjobsチャネルを閉じることで、これ以上新しいジョブが追加されないようにする。その後、q.wg.Wait()を呼び出す。これは、先に起動した全てのワーカー（goroutine）がそれぞれの処理を完了し、wg.Done()を呼び出すまで、Stopメソッドの実行を一時停止させる役割がある。これにより、キューが停止する際に、処理中のジョブが途中で失われることなく、全て完了することが保証される。ワーカーが全て終了したら、エラーチャネルも閉じられる。

この設計におけるいくつかのポイントがある。バッファ付きチャネルとノンブロッキングなエンキューは、呼び出し元がシステムの負荷状況に応じて柔軟に対応できるようにするためのものだ。また、Queue構造体がグローバルな状態を持たないため、同じプログラム内で複数の独立したジョブキューを同時に実行できる柔軟性がある。ジョブのリトライ処理はワーカー内で完結するため、別途スケジューラを用意する必要がなくシンプルだが、その分、再試行中のジョブがワーカーのスロットを占有し続けることになる。これは、ジョブの量が少ないキューであれば問題ないが、ワーカーがボトルネックになるような高負荷環境では見直しの余地がある。

システムの安全なシャットダウンも非常に重要だ。OSがプログラムの停止を要求するシグナル（例えば、Ctrl+CによるSIGINTや、システム再起動時のSIGTERM）を受け取った際、プログラムはすぐに終了するのではなく、現在実行中のジョブを最後まで処理してから終了するのが理想的である。このジョブキューでは、os/signalパッケージを使ってこれらのシグナルを捕捉し、シグナルを受け取った際にQueueのStopメソッドを呼び出す。Stopメソッド内のsync.WaitGroupが、全てのワーカーが完了するまで待機することで、実行中のジョブが途中で失われることなく、安全にシステムを終了できる。

さらに、このキューに優先度を付ける応用も可能だ。Goの標準チャネルには優先度という概念は直接ないが、これを実現する方法として、高優先度用と通常優先度用の二つのチャネルを用意するパターンが紹介されている。ワーカーは、まずselect文を使って高優先度チャネルからジョブが来ているかをdefault節付きで確認し、もしジョブがあればそれを取り出して処理する。その次に、再度select文を使って高優先度チャネルと通常優先度チャネルの両方を確認する。この二重のselect構造が重要で、最初のdefault付きselectがなければ、二つのチャネルに同時にジョブがあった場合、Goのselectはランダムにチャネルを選択するため、高優先度ジョブが後回しにされてしまう可能性があるからだ。この工夫により、高負荷時でも高優先度ジョブが確実に先に処理されるようになる。

しかし、このGo言語の標準機能で構築されたジョブキューには、外部の専用メッセージブローカーが提供するような高度な機能のいくつかがない点も理解しておく必要がある。まず、「永続性」がない。もしこのプログラムがクラッシュした場合、キューの中に残っていたジョブは失われてしまう。金融取引のような非常に重要な処理の場合には、ジョブをキューに入れる前にデータベース（PostgresやSQLiteなど）に書き込んでおくなどの対策が必要になる。次に、「分散処理」には対応していない。このキューは一つのプロセスの中でしか機能しないため、複数のサーバーにまたがってジョブを処理したい場合には、アプリケーションレベルでジョブの分散を管理する必要がある。また、「デッドレターキュー」のような機能も組み込まれていない。これは、リトライ回数を使い果たしても失敗したジョブを自動的に別のキューに移動させて、後で手動で確認したり再処理したりする仕組みだが、このキューでは単にエラーチャネルにエラーが送られるだけで、その後の処理は開発者が責任を持つことになる。さらに、より詳細な「バックプレッシャー」制御もできない。このキューはジョブを受け入れるか拒否するかの二択だが、本格的なブローカーは、ワーカーからのジョブ完了通知や、キューの深さに応じたフロー制御など、よりきめ細やかな負荷調整機能を提供する。

これらの限界を理解した上で、このGo言語で構築するシンプルなジョブキューは、単一サービス内のバックグラウンド処理、例えばメール送信、レポート生成、ファイルのアップロード処理などには非常に強力なツールとなる。外部サービスを運用する手間やコストが一切かからず、数千のジョブを毎秒処理できる能力を持っている。耐久性やプロセス間の連携が必要な場合には、改めて本格的なメッセージブローカーの導入を検討すべきだが、まずはGoの標準機能だけで構築できるこのシンプルなジョブキューから始めることが推奨されている。バッファ付きチャネルとワーカープール、安全なシャットダウンのためのWaitGroup、そしてエラーを通知するためのチャネルがあれば、RedisやRabbitMQといった外部依存なしに、多くのユースケースに対応できる強固な基盤を築けるのだ。本当に外部ブローカーが必要だと、実際の運用データで示せるようになるまで、まずはこのシンプルなアプローチを試してみるべきである。