【ITニュース解説】Turning PostgreSQL into a Robust Queue for Go Applications

システム開発において、アプリケーションが実行する処理の中には、ユーザーからのリクエストに応答するのと同時に行う必要がなく、バックグラウンドで時間をかけて実行しても問題ないものがある。たとえば、新規ユーザーが登録された後に確認メールを送ったり、アップロードされた画像を適切なサイズに変換したりする処理などがこれにあたる。このような処理は「非同期タスク」と呼ばれ、これらを効率的に管理する仕組みが「キュー（Queue）」だ。多くの開発では、RabbitMQやRedisといった専用のキューシステムが使われるが、今回の記事では、すでに多くのシステムで利用されているリレーショナルデータベース「PostgreSQL」を、このキューとして活用する方法を解説する。

PostgreSQLをキューとして使う最大の利点は、システム構成をシンプルに保てることにある。新たなミドルウェアを追加導入する手間やコストを省き、既存のデータベースインフラをそのまま利用できるため、セットアップが非常に簡単だ。さらに、PostgreSQLが持つ強力な「トランザクション」機能と「ACID特性」（原子性、一貫性、独立性、永続性）によって、タスクの処理は非常に高い信頼性を持つ。つまり、途中でシステムがクラッシュしたり、ネットワークに障害が発生したりしても、タスクのデータが失われたり、同じタスクが複数回処理されてしまったりするリスクを低く抑えられる。これは、専用のキューシステムが必ずしも自動で提供するとは限らない重要な特徴だ。PostgreSQLは、秒間何万件ものタスクを処理するような超高速なスループットには向かないが、毎分数千件程度のタスクを処理する中程度のワークロードであれば、十分にその信頼性と性能を発揮する。

PostgreSQLをキューとして利用するための準備は、まずPostgreSQLデータベースのインストールと設定から始まる。記事ではバージョン14以降の使用を推奨しており、これは新しいバージョンに性能改善や便利な機能が追加されているためだ。キュー用のデータベースとして、task_queueのような名前で新しいデータベースを作成する。Go言語でPostgreSQLデータベースを操作するには、「github.com/lib/pq」という専用のドライバーが必要になるので、Goプロジェクトにこれをインストールする。データベースへの接続は、ユーザー名、パスワード、ホスト名、ポート番号、データベース名などの情報を含む「接続文字列」を使って行う。複数のアプリケーションや処理が同時にデータベースにアクセスする場合、Goのdatabase/sqlパッケージが提供する「コネクションプール」という機能を使うことで、効率的に接続を管理できる。

次に、キューの核となる「タスクテーブル」の設計を行う。タスクの情報を保存するために、例えばtasksという名前のテーブルを作成する。このテーブルには、各タスクを一意に識別するためのid（自動で連番が振られる）、タスクの具体的な内容をJSON形式で保存するpayload（JSONB型はJSONデータを効率的に保存・検索できる）、タスクの現在の状態を示すstatus（pending：処理待ち、processing：処理中、failed：失敗、done：完了など）、タスクが作成された日時created_at、これまでにタスク処理を試みた回数attempts、最後に処理が試みられた日時last_attempt_atといったカラムを含める。特にstatusカラムには、処理待ちのタスクを素早く見つけ出すために「インデックス」を設定しておくのが効率的だ。インデックスは本の索引のようなもので、特定の条件に合うデータをデータベースが高速に探し出すのに役立つ。

Go言語のプログラムからタスクをキューに追加する（この操作を「エンキュー」と呼ぶ）には、database/sqlパッケージを利用して、tasksテーブルに新しい行を挿入する。例えば、メール送信のタスクであれば、宛先や件名などの情報をJSON形式に変換し、それをpayloadカラムに格納する。statusカラムはデフォルトでpending（処理待ち）とする。この挿入操作は、もし複数の関連するデータベース操作とまとめて行いたい場合は、トランザクションの中で実行することで、一連の処理が完全に成功するか、完全に失敗するかのどちらかになることを保証できる。

キューからタスクを取り出して処理する（この操作を「デキュー」と呼ぶ）部分は、少し高度なテクニックが必要になる。複数のGoワーカー（タスクを処理するプログラムのインスタンス）が同時に稼働している場合、同じタスクを二重に処理してしまう「競合状態」を避ける必要があるからだ。ここでPostgreSQLの強力な機能である「SELECT ... FOR UPDATE SKIP LOCKED」というSQLクエリが活躍する。このクエリは、pending状態のタスクの中から、最も古く作成されたもの（ORDER BY created_at ASC）を一つだけ選び出す。そして、その選ばれたタスクの行を他のワーカーが取得できないように「ロック」する。さらにSKIP LOCKEDという指定があるため、もしそのタスクがすでに別のワーカーによってロックされていて利用できない場合でも、そのタスクをスキップして、次に利用可能なタスクを探しに行くことができる。これにより、複数のワーカーが互いに邪魔し合うことなく、それぞれ異なるタスクを効率的に処理できるようになる。タスクを取得したら、そのタスクのstatusをprocessingに更新し、実際の処理（例えばメール送信）を実行する。処理が問題なく完了したら、最後にstatusをdoneに更新する。

タスクの処理は、常に成功するとは限らない。ネットワークエラーや外部サービスの障害など、様々な理由で失敗する可能性があるため、「リトライロジック」（再試行の仕組み）の実装も重要だ。タスク処理中にエラーが発生した場合、すぐにそのタスクを諦めるのではなく、数回再試行する機会を与える。具体的には、処理中にエラーが発生したら、tasksテーブルのattemptsカラムの値を1増やし、statusを再びpendingに戻す。これにより、後で別のワーカーがそのタスクを再度取得し、処理を試みることができる。ただし、無限にリトライし続けるのはシステム負荷を高めるだけなので、attemptsカラムの値が一定の回数（例えば3回）を超えた場合は、statusをfailedに設定し、それ以上再試行しないようにする。

より大規模なワークロードに対応するためには、スケーリングとパフォーマンスの最適化を考慮する必要がある。前述のSKIP LOCKEDのおかげで、Goワーカーの数を増やして並行して実行するだけで、タスク処理のスループットを簡単に向上させることができる。しかし、ワーカーが常にデータベースに新しいタスクがないか問い合わせる「ポーリング」方式は、データベースとワーカーの両方に無駄な負荷をかける可能性がある。これを効率化するのが、PostgreSQLの「LISTEN/NOTIFY」機能だ。新しいタスクがキューに追加されたときにデータベースからNOTIFY（通知）を送り、GoワーカーはLISTEN（待機）してその通知を受け取る。通知を受け取ったワーカーはすぐにタスクの処理を開始できるため、無駄なポーリングが減り、CPUなどのリソースを節約できる。

その他にも、パフォーマンスを最適化するための様々な手段がある。SQLクエリの実行計画を分析するEXPLAIN ANALYZEコマンドを使って、どの部分の処理に時間がかかっているか（ボトルネック）を特定したり、VACUUM ANALYZEコマンドでデータベースの統計情報を更新し、クエリの効率を高めたりする。非常に多くのタスクを扱う場合は、tasksテーブルをstatusや作成日時で「パーティショニング」し、データ量を分割して管理することで、検索や更新の速度を向上させることが有効だ。Goのデータベース接続では、MaxOpenConnectionsなどの設定でコネクションプールを適切に調整することも、パフォーマンス維持には重要だ。定期的にベンチマークテストを実行して、アプリケーションの負荷状況を把握し、処理が完了したタスクを別のアーカイブテーブルに移動する「アーカイブ処理」を導入して、tasksテーブルのサイズを適切に保つことも、長期的なパフォーマンス維持には不可欠となる。

このように、PostgreSQLとGo言語を組み合わせることで、専用のキューシステムに匹敵する信頼性と柔軟性を持ちながら、既存のデータベースインフラ上で効率的な非同期タスクキューを構築できる。システムエンジニアを目指す初心者にとって、このようなシンプルな構成で信頼性の高い非同期処理を実現するアーキテクチャは、非常に実践的で役立つ知識となるだろう。