ApsaraMQ for RocketMQ:Pengulangan pesan

Waktu retry

Interval antara dua upaya konsumsi berturut-turut terdiri dari tiga komponen:

Interval antar upaya = durasi pemrosesan + interval retry + waktu dalam status Ready

Sebagai contoh, asumsikan sebuah pesan menunggu 5 detik dalam status Ready dan membutuhkan 6 detik untuk diproses sebelum gagal. Dengan interval retry 10 detik:

1. 0 dtk — Pesan memasuki status Ready.

2. 5 dtk — Konsumen menarik pesan (Inflight).

3. 11 dtk — Konsumsi gagal setelah 6 detik. Pesan memasuki WaitingRetry.

4. 21 dtk — Setelah interval retry 10 detik, pesan kembali ke status Ready.

5. 26 dtk — Konsumen menarik pesan lagi untuk upaya berikutnya.

Total interval: 6 + 10 + 5 = 21 detik.

Pesan terurut

Pesan tidak terurut

Jika MaxReconsumeTimes melebihi 16, jadwal bawaan berlaku untuk 16 retry pertama. Semua retry berikutnya menggunakan interval tetap 2 jam.

Aktifkan retry

Untuk memicu retry saat konsumsi gagal dalam clustering consumption mode, implementasikan antarmuka MessageListener dengan salah satu pendekatan berikut:

• Kembalikan Action.ReconsumeLater (disarankan)

• Kembalikan null

• Lempar exception

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        // Jika logika konsumsi melempar exception, pesan akan di-retry.
        doConsumeMessage(message);
        // Metode 1: Kembalikan Action.ReconsumeLater untuk retry pesan.
        return Action.ReconsumeLater;
        // Metode 2: Kembalikan null untuk retry pesan.
        return null;
        // Metode 3: Lempar exception untuk retry pesan.
        throw new RuntimeException("Consumer Message exception");
    }
}

Nonaktifkan retry

Untuk mencegah pengiriman ulang, tangkap semua exception dalam logika konsumsi Anda dan kembalikan Action.CommitMessage:

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            doConsumeMessage(message);
        } catch (Throwable e) {
            // Tangkap semua exception dan kembalikan CommitMessage untuk melewati retry.
            return Action.CommitMessage;
        }
        // Konsumsi berhasil.
        return Action.CommitMessage;
    }
}

Kustomisasi interval retry dan maksimum retry

Atur MaxReconsumeTimes dan SuspendTimeMillis dalam properti konsumen sebelum memulai konsumen. Interval retry kustom hanya berlaku untuk pesan terurut. Pesan tidak terurut selalu mengikuti jadwal bawaan.

Properties properties = new Properties();
// Atur jumlah maksimum retry menjadi 20.
properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");
// Atur interval retry menjadi 3.000 milidetik (hanya untuk pesan terurut).
properties.put(PropertyKeyConst.SuspendTimeMillis, "3000");
Consumer consumer = ONSFactory.createConsumer(properties);

Periksa jumlah retry

Setelah konsumen menerima pesan, panggil message.getReconsumeTimes() untuk memeriksa berapa kali pesan tersebut telah di-retry:

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        // Dapatkan jumlah retry saat ini.
        System.out.println(message.getReconsumeTimes());
        return Action.CommitMessage;
    }
}