ApsaraMQ for RocketMQ:コンシューマー側のべき等性

プロデューサーのリトライ

プロデューサーがメッセージを送信し、ApsaraMQ for RocketMQ のブローカーがそれを永続化します。しかし、一時的なネットワーク障害やプロデューサーのクラッシュにより、ブローカーがプロデューサーへの応答に失敗することがあります。プロデューサーはこれを送信失敗とみなし、再送信（リトライ）します。その結果、コンシューマーは内容が同じだがメッセージ ID が異なる 2 つのメッセージを受信します。

ブローカーによる再配信

コンシューマーがメッセージを受信・処理したものの、一時的なネットワーク障害によりブローカーへの応答（ACK）に失敗します。ブローカーはメッセージが正常に消費されたかどうかを確認できないため、「少なくとも 1 回」の保証を守るためにネットワークが回復後に再度メッセージを配信します。この場合、コンシューマーは内容もメッセージ ID も同じ 2 つのメッセージを受信します。

負荷分散

ネットワークジッター、ブローカーの再起動、コンシューマーアプリケーションの再起動などのイベントにより負荷分散がトリガーされます。リバランス中に、すでに配信済みのメッセージを再度受信することがあります。

ステップ 1：プロデューサー側でメッセージキーを設定する

メッセージ送信時に、一意なビジネス識別子をメッセージキーとして設定します。

Message message = new Message();
message.setKey("ORDERID_100");
SendResult sendResult = producer.send(message);

ORDERID_100 は、実際の注文 ID やトランザクション ID などの一意なビジネス識別子に置き換えてください。

ステップ 2：コンシューマー側でメッセージキーを取得する

コンシューマーのコールバック内でメッセージキーを取得し、べき等処理を実施します。

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey()
        // ビジネスを一意に識別するメッセージキーに基づいて、べき等処理を実行します。
    }
});

ステップ 3：重複排除ストアを使用してべき等性を保証する

メッセージキーだけでは重複処理を防げません。アプリケーション側で、そのキーがすでに処理済みかどうかをチェックする必要があります。一般的なパターンとして、一意制約付きのカラムを持つリレーショナルデータベースを使用する方法があります。

1. 処理前に、メッセージキーを一意制約付きの重複排除テーブルに挿入します。

2. 挿入が成功した場合は、メッセージを処理します。

3. プライマリキーまたは一意制約違反により挿入に失敗した場合は、そのメッセージはすでに処理済みです。スキップします。

重複排除テーブルの例 (MySQL)：

CREATE TABLE message_dedup (
    message_key VARCHAR(255) NOT NULL,
    created_at  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (message_key)
);

べき等コンシューマーのロジック例：

consumer.subscribe("ons_test", "*", new MessageListener() {
    public Action consume(Message message, ConsumeContext context) {
        String key = message.getKey();

        try {
            // メッセージキーを挿入します。すでに処理済みの場合は失敗します。
            insertMessageKey(key);
        } catch (DuplicateKeyException e) {
            // すでに処理済みです。スキップします。
            return Action.CommitMessage;
        }

        // ビジネスロジックを処理します。
        processOrder(key);

        return Action.CommitMessage;
    }
});

insertMessageKey および processOrder は、実際のデータベースおよびビジネスロジックのメソッドに置き換えてください。

リレーショナルデータベースがボトルネックとなるような高スループットのシナリオでは、代わりに Redis の SETNX (SET if Not eXists) を使用したアプローチを採用してください。