Posted inDevelopment

RabbitMQにおけるメッセージ重複の処理：Node.jsによる冪等なコンシューマー（Idempotent Consumer）の実装

Development tutorial - IT technology blog
Development tutorial - IT technology blog

古典的な課題：なぜ顧客は二重に課金されるのか？

決済システムを運用していると想像してみてください。顧客が確定ボタンを押しましたが、ネットワークの遅延（network jitter）により、Consumerが処理を完了したもののRabbitMQにACKを返す前に接続が切断されてしまいました。RabbitMQは接続断を検知し、そのメッセージを別のConsumerに再送します。その結果、顧客は二重に課金されてしまいます。これはコードのバグではなく、分散システム特有の性質です。

RabbitMQは「at-least-once delivery」（少なくとも1回は配信されること）を保証しますが、「exactly-once」は保証しません。メッセージの重複は避けられないものです。これを解決するには、Consumerを冪等なコンシューマー（Idempotent Consumer）にする必要があります。簡単に言えば、同じメッセージを1回受け取っても100回受け取っても、システムの状態が変わらないようにすることです。

環境構築

実践するために、Node.jsと、RabbitMQおよびRedisを素早く起動するためのDockerが必要です。Redisは読み書きが非常に高速なため、数ミリ秒で重複チェックを行う message_id の保存先として適しています。

# RabbitMQとRedisを起動
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management
docker run -d --name redis -p 6379:6379 redis:alpine

プロジェクトを初期化し、必要なライブラリをインストールします：

mkdir rabbitmq-idempotency && cd rabbitmq-idempotency
npm init -y
npm install amqplib ioredis uuid

戦略：先にチェックし、後で処理する

プロセスは非常に明確です。送信される各メッセージには一意の messageId を付与します。Consumerがメッセージを受信すると、Redisを検索します。IDが既に存在する場合は無視します。存在しない場合は、処理を実行し、そのIDをRedisに保存します。

1. Producer：識別子付きのメッセージ送信

ペイロードをそのまま送信してはいけません。管理しやすいように、メタデータを含むオブジェクトでデータをラップしましょう。

const amqp = require('amqplib');
const { v4: uuidv4 } = require('uuid');

async function sendOrder() {
    const conn = await amqp.connect('amqp://localhost');
    const channel = await conn.createChannel();
    const queue = 'order_queue';

    const message = {
        id: uuidv4(), // 各トランザクションの一意の識別子
        data: { orderId: 'ORD-999', amount: 500000 }
    };

    await channel.assertQueue(queue, { durable: true });
    channel.sendToQueue(queue, Buffer.from(JSON.stringify(message)), {
        persistent: true
    });

    console.log(`[x] 注文を送信しました: ${message.id}`);
    setTimeout(() => conn.close(), 500);
}

sendOrder();

2. Consumer：SETNXによる重複防止メカニズム

ここでは、Redisの SETNX（Set if Not Exists）コマンドを使用します。これは原子的な操作（atomic operation）です。これにより、たとえ10個のインスタンスが同時に同じメッセージを受信したとしても、ただ一つのインスタンスだけが処理権限を取得できることが保証されます。

const amqp = require('amqplib');
const Redis = require('ioredis');
const redis = new Redis();

async function consume() {
    const conn = await amqp.connect('amqp://localhost');
    const channel = await conn.createChannel();
    const queue = 'order_queue';

    await channel.assertQueue(queue, { durable: true });
    channel.prefetch(1);

    channel.consume(queue, async (msg) => {
        if (!msg) return;
        
        const { id, data } = JSON.parse(msg.content.toString());
        
        // メモリの肥大化を防ぐため、24時間のTTL付きでRedisへの書き込みを試行
        const isNew = await redis.set(`msg:${id}`, 'processing', 'NX', 'EX', 86400);

        if (isNew) {
            try {
                console.log(`[v] メッセージを処理中: ${id}`);
                await processOrder(data); // ビジネスロジックのシミュレーション
                channel.ack(msg);
            } catch (err) {
                console.error("処理エラー:", err);
                await redis.del(`msg:${id}`); // リトライ可能にするためにキーを削除
                channel.nack(msg, false, true);
            }
        } else {
            console.warn(`[!] 重複を検知: ${id}。スキップします...`);
            channel.ack(msg); // キューから削除するためにACKを返す必要がある
        }
    });
}

async function processOrder(data) {
    return new Promise(res => setTimeout(res, 1000));
}

consume();

実際、私はかつて1日100万件以上のメッセージを処理するシステムを運用していました。そこから得た教訓は、TTL (Time To Live) が極めて重要であるということです。TTLを設定しないと、数ヶ月後にはRedisが膨大になり、リソースの無駄遣いや検索速度の低下を招きます。

監視と測定

コードを書いて終わりではありません。以下の指標に注目する必要があります：

  • Redis Hit/Miss Rate: 重複率が急増（例：5%以上）した場合、ネットワークに深刻な問題が発生している可能性があります。
  • Consumer Lag: RabbitMQ UI for 監視します。Unacknowledged Messagesが増加している場合、処理ロジックがメッセージの流入速度に追いついていないことを意味します。
  • Dead Letter Queue (DLQ): 何度もエラーになるメッセージのために、常に「墓場」を用意しておきましょう。1つのエラーメッセージが処理フロー全体を停滞させないようにします。

分散システムを構築するということは、不確実性を受け入れることを学ぶことです。重複を完全に防ごうとするのではなく、すでに行った処理を認識し、拒否できるほど賢いConsumerを設計しましょう。

Share:
Tags: