Message Broker不要のリアルタイム処理：PostgreSQLのLISTEN/NOTIFYとWebSocketを活用する

ポーリングの悪夢とリアルタイム通信の課題

午前2時、PagerDutyのアラートがけたたましく鳴り響きます。ダッシュボードを開くと、8 vCPUの本番サーバーのCPU使用率が98%に達していました。原因は明白でした。15,000のクライアントが新規注文を確認するために、2秒ごとにポーリングリクエストを送り続けていたのです。さらに悪いことに、隣のRabbitMQクラスタはメモリ枯渇でOOMKilled（強制終了）を起こしていました。インフラ全体が悲鳴を上げている理由はただ一つ、「注文通知を即座にプッシュ配信する」ためだけだったのです。

データベースの更新をリアルタイムにUIへ反映する際、バックエンドエンジニアは通常、次の3つのアプローチのいずれかを選択します。

• Short PollingまたはLong Polling：クライアントが2〜3秒ごとにサーバーへHTTP GETリクエストを送信する方式。実装は最も簡単ですが、無駄なリソースを大量に消費します。
• 専用Message Broker（RabbitMQ、Kafka、Redis Pub/Sub）：中間のブローカークラスタでキュー管理とメッセージ配信を行う方式。DBの変更時にバックエンドがブローカーへイベントを送信し、ワーカーがそれを取り出してWebSocket経由で配信します。
• PostgreSQLのLISTEN/NOTIFYとWebSocketの組み合わせ：Postgresに組み込まれているPub/Subエンジンを活用する方式。新規データ発生時にDB自身が内部チャネルへシグナルを発行します。バックエンドはLISTEN接続を1本維持してペイロードを受け取り、WebSocket経由でクライアントへ直接ブロードキャストします。

各アーキテクチャのメリット・デメリットを徹底比較

高負荷システムの最適化を重ねてきた経験から、各ソリューションの実質的なコストとトレードオフをまとめました。

1. 従来のポーリング

• メリット：実装が極めて容易。フロントエンド側でsetIntervalを使ってfetch()を呼び出すだけで完結します。
• デメリット：帯域とリソースの浪費が深刻。実際にはリクエストの95%以上が空のデータを返します。セール時などに数百万件のリクエストが殺到すると、データベースのコネクションプール（50〜100接続程度）が容易に枯渇し、システム全体のダウンにつながります。

2. 専用Message Broker（RabbitMQ、Redis）

• メリット：優れたスケーラビリティ。毎秒数十万件のメッセージを処理可能で、柔軟なルーティングや本格的なバックプレッシャー機構に対応しています。
• デメリット：運用コストの増大。クラスタの構築、アラート設定、追加インフラの維持管理が必要です。何より危険なのがDual-write（二重書き込み）問題です。データベースのコミット直後にアプリケーションサーバーがクラッシュすると、RabbitMQへのパブリッシュが失敗し、DBにはデータが存在するのにユーザーには一切通知が届かないという不整合が発生します。

3. PostgreSQLのLISTEN/NOTIFY

• メリット：
  • インフラのオーバーヘッドがゼロ：追加のコンテナや仲介サーバークラスタが一切不要。普段CRUD操作を行っているPostgreSQLそのものがブローカーとして機能します。
  • トランザクション整合性の担保：NOTIFYシグナルはトランザクションが正常にCOMMITされた場合にのみ送信されます。ROLLBACKされた場合は通知も自動的に破棄されるため、不正なゴミ通知が飛ぶ心配がありません。
  • 極めて低いリソース消費：リスナー接続はノンブロッキングな永続接続（persistent connection）を使用し、バックエンド側のメモリ消費量はわずか2〜5MB程度に収まります。
• デメリット：
  • ペイロードの上限が8,000バイトに制限されていること。IDや少数のメタデータを含むJSONには十分ですが、巨大なオブジェクト丸ごとを送ることはできません。
  • キューの永続化（persistence）がないこと。PostgresがNOTIFYを発行したまさにその瞬間にNode.jsサーバーが再起動中だった場合、そのメッセージは消失します。
  • コネクションプーラーの落とし穴：セッションレベルの接続が必要です。PgBouncerをtransaction poolingモードで使用している場合、LISTENコマンドは機能しません。解決策として、リスナー接続だけはPostgreSQLへ直接向けるか、専用にsessionモードのプールを確保する必要があります。

PostgreSQL LISTEN/NOTIFYを採用すべきケース

数百万CCUを抱える取引所や、毎日テラバイト規模のログを処理するシステムを構築しているのであれば、KafkaやRabbitMQへの投資は不可欠です。しかし、一般的な中小型アプリケーション、B2B SaaS、管理画面のダッシュボード、社内向けの注文通知ベル、あるいはレポート出力の進捗表示（秒間500〜1,000イベント未満）といった用途では、RedisやRabbitMQクラスタをわざわざ立ち上げることは単に単一障害点（SPOF）を増やすだけに過ぎません。

PostgresのLISTEN/NOTIFYは、シンプルで無駄のないアーキテクチャが求められる場面で最大の真価を発揮します。データが確実にディスクへ書き込まれたことを100%保証した上でユーザーへ通知したい、かつクラウドコストを最小限に抑えたい場合に最適です。

通知システムやマスタデータの初期セットアップを行う際、設定ファイルのデータ変換が必要になることがよくあります。データベースへのインポート用にCSVをJSONへ手早く変換したい場合、ブラウザ上で完結して外部サーバーにデータを送信しないtoolcraft.app/ja/tools/data/csv-to-jsonを活用すると、セキュリティ面でも安心でおすすめです。

実践ハンズオン：PostgreSQL + Node.js + WebSocket

それでは実際の実装に入りましょう。全体の流れはシンプルな4つのステップで構成されています。

ステップ1：PostgreSQLでトリガーと関数を作成する

ordersテーブルを作成し、データをJSON形式にまとめてpg_notifyを呼び出す関数を定義し、AFTER INSERTイベントで発火するように設定します。

-- 1. 注文テーブルを作成
CREATE TABLE IF NOT EXISTS orders (
    id SERIAL PRIMARY KEY,
    customer_name VARCHAR(100) NOT NULL,
    total_amount NUMERIC(12, 2) NOT NULL,
    status VARCHAR(50) DEFAULT 'PENDING',
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- 2. NOTIFYを発行するトリガー関数を作成
CREATE OR REPLACE FUNCTION notify_new_order()
RETURNS TRIGGER AS $$
DECLARE
    notification_payload JSON;
BEGIN
    -- 最小限のデータにパッキング（8000バイト上限の超過を防止）
    notification_payload = json_build_object(
        'id', NEW.id,
        'customer_name', NEW.customer_name,
        'total_amount', NEW.total_amount,
        'status', NEW.status,
        'created_at', NEW.created_at
    );

    -- 'new_order_channel' チャネルへペイロードを送信
    PERFORM pg_notify('new_order_channel', notification_payload::text);
    
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 3. ordersテーブルにトリガーを紐付け
DROP TRIGGER IF EXISTS trg_orders_notify ON orders;
CREATE TRIGGER trg_orders_notify
AFTER INSERT ON orders
FOR EACH ROW
EXECUTE FUNCTION notify_new_order();

ステップ2：LISTENを受信してWebSocket配信するNode.jsバックエンドの構築

バックエンド層では、定番ライブラリであるpgとwsをインストールします。

npm install pg ws

WebSocketサーバーを初期化し、LISTENコマンドを維持するためのPostgreSQL専用接続を別途確立します。

// server.js
const { Client } = require('pg');
const { WebSocketServer } = require('ws');

// 1. ポート8080でWebSocketサーバーを初期化
const wss = new WebSocketServer({ port: 8080 });
wss.on('connection', (ws) => {
    console.log('[WebSocket] クライアントが接続しました。');
    ws.on('close', () => console.log('[WebSocket] クライアントの接続が切断されました。'));
});

function broadcast(data) {
    wss.clients.forEach((client) => {
        if (client.readyState === 1) { // 1 = OPEN
            client.send(JSON.stringify(data));
        }
    });
}

// 2. PostgreSQL専用接続を確立（CRUD用のコネクションプールとの共用を回避）
const pgClient = new Client({
    connectionString: process.env.DATABASE_URL || 'postgresql://postgres:password@localhost:5432/production_db'
});

async function startListening() {
    try {
        await pgClient.connect();
        console.log('[PostgreSQL] データベースに正常に接続しました。');

        // チャネルからの通知受信を登録
        await pgClient.query('LISTEN new_order_channel');
        console.log('[PostgreSQL] チャネル「new_order_channel」でNOTIFYを待機中...');

        // Postgresから通知を受信した際のイベント処理
        pgClient.on('notification', (msg) => {
            try {
                const payload = JSON.parse(msg.payload);
                broadcast({
                    event: 'ORDER_CREATED',
                    data: payload
                });
            } catch (err) { 
                console.error('JSONペイロードのパースエラー:', err);
            }
        });

        pgClient.on('error', (err) => {
            console.error('[PostgreSQL Error]', err);
            process.exit(1); // プロセスマネージャー（Docker/PM2等）による再起動と接続復旧に委ねる
        });
    } catch (error) {
        console.error('PostgreSQLへの接続に失敗しました:', error);
        setTimeout(startListening, 5000);
    }
}

startListening();

ステップ3：フロントエンドでリアルタイム通知を受信する

Webクライアント側では、サーバーへのWebSocket接続を開き、メッセージ受信時に即座にUIを更新します。

// client.js
const socket = new WebSocket('ws://localhost:8080');

socket.onopen = () => {
    console.log('WebSocketサーバーに接続しました。');
};

socket.onmessage = (event) => {
    const response = JSON.parse(event.data);
    
    if (response.event === 'ORDER_CREATED') {
        const order = response.data;
        console.log('新規注文:', order);
        // 実際のUIではalertの代わりにToast通知などに置き換えてください
        alert(`新規注文 #${order.id}（顧客: ${order.customer_name}、金額: $${order.total_amount}）が作成されました！`);
    }
};

socket.onerror = (err) => {
    console.error('WebSocket接続エラー:', err);
};

ステップ4：動作確認

psqlコンソールを開き、テスト用のレコードを挿入して検証してみましょう。

INSERT INTO orders (customer_name, total_amount) 
VALUES ('Nguyen Van A', 450.00);

トランザクションがコミットされた瞬間にトリガーがpg_notifyを発行します。ポーリングリクエストを一切送信することなく、10ミリ秒未満でブラウザ上に通知が表示されます。Redisのために追加のメモリを消費することも一切なく、データベースのCPU使用率も5%未満に維持され、PagerDutyのアラートに怯えることなく夜もぐっすり眠れるようになります。