ApsaraMQ for RocketMQ:メッセージリトライ

リトライのタイミング

2つの連続する消費試行間の間隔には、次の3つのコンポーネントがあります。

試行間の間隔 = 消費時間 + 再試行間隔 + Ready 状態での時間

例えば、メッセージが Ready 状態で5秒間待機し、失敗するまでに処理に6秒かかると仮定します。10秒の再試行間隔の場合：

1. 0 秒 -- メッセージは Ready 状態に入ります。

2. 5 秒 -- コンシューマーがメッセージをプルします (Inflight)。

3. 11 秒 -- 6秒後に消費に失敗します。メッセージは WaitingRetry に入ります。

4. 21 秒 -- 10秒の再試行間隔の後、メッセージは Ready に戻ります。

5. 26 秒 -- コンシューマーは次の試行のためにメッセージを再度プルします。

合計間隔：6 + 10 + 5 = 21 秒。

順序付きメッセージ

順序なしメッセージ

MaxReconsumeTimes が 16 を超える場合、最初の 16 回のリトライには事前定義されたスケジュールが適用されます。その後のすべてのリトライは、固定の 2 時間間隔を使用します。

リトライの有効化

クラスター消費モードで消費に失敗したときにリトライをトリガーするには、次のいずれかのアプローチで MessageListener インターフェイスを実装します。

• Action.ReconsumeLater を返す (推奨)

• null を返す

• 例外をスローする

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        // 消費ロジックが例外をスローした場合、メッセージはリトライされます。
        doConsumeMessage(message);
        // 方法1: Action.ReconsumeLater を返し、メッセージをリトライします。
        return Action.ReconsumeLater;
        // 方法2: null を返し、メッセージをリトライします。
        return null;
        // 方法3: 例外をスローし、メッセージをリトライします。
        throw new RuntimeException("Consumer Message exception");
    }
}

リトライの無効化

再配信を防ぐには、消費ロジック内のすべての例外をキャッチし、Action.CommitMessage を返します。

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            doConsumeMessage(message);
        } catch (Throwable e) {
            // すべての例外をキャッチし、CommitMessage を返してリトライをスキップします。
            return Action.CommitMessage;
        }
        // 消費に成功しました。
        return Action.CommitMessage;
    }
}

リトライ間隔と最大リトライ回数のカスタマイズ

コンシューマーを開始する前に、コンシューマープロパティで MaxReconsumeTimes と SuspendTimeMillis を設定します。カスタムリトライ間隔は順序付きメッセージにのみ適用されます。順序なしメッセージは常に事前定義されたスケジュールに従います。

Properties properties = new Properties();
// 最大リトライ回数を20に設定します。
properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");
// 再試行間隔を3,000ミリ秒に設定します (順序付きメッセージのみ)。
properties.put(PropertyKeyConst.SuspendTimeMillis, "3000");
Consumer consumer = ONSFactory.createConsumer(properties);

リトライ回数の照会

コンシューマーがメッセージを受信した後、message.getReconsumeTimes() を呼び出して、メッセージが何回リトライされたかを確認します。

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        // 現在のリトライ回数を取得します。
        System.out.println(message.getReconsumeTimes());
        return Action.CommitMessage;
    }
}