ApsaraMQ for RocketMQ:Message retry

Yeniden deneme zamanlaması

Arka arkaya gelen iki tüketim girişimi arasındaki aralık üç bileşenden oluşur:

Girişimler arası aralık = tüketim süresi + yeniden deneme aralığı + Ready durumunda geçen süre

Örneğin, bir mesajın Ready durumunda 5 saniye beklediğini ve başarısız olmadan önce işlenmesinin 6 saniye sürdüğünü varsayalım. 10 saniyelik bir yeniden deneme aralığı ile:

1. 0 sn -- Mesaj Ready durumuna girer.

2. 5 sn -- Tüketici mesajı çeker (Inflight).

3. 11 sn -- 6 saniye sonra tüketim başarısız olur. Mesaj WaitingRetry durumuna girer.

4. 21 sn -- 10 saniyelik yeniden deneme aralığından sonra mesaj Ready durumuna döner.

5. 26 sn -- Tüketici, sonraki girişim için mesajı tekrar çeker.

Toplam aralık: 6 + 10 + 5 = 21 saniye.

Sıralı mesajlar

Sırasız mesajlar

MaxReconsumeTimes değeri 16'yı aşarsa, önceden tanımlanmış zamanlama ilk 16 yeniden deneme için uygulanır. Sonraki tüm yeniden denemeler sabit 2 saatlik bir aralık kullanır.

Yeniden denemeyi etkinleştirme

Kümelenmiş tüketim modunda tüketim başarısız olduğunda yeniden denemeyi tetiklemek için MessageListener arayüzünü şu yaklaşımlardan biriyle uygulayın:

• **Action.ReconsumeLater değerini döndürün** (önerilir)

• null değerini döndürün

• Bir istisna fırlatın

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        // If the consumption logic throws an exception, the message is retried.
        doConsumeMessage(message);
        // Method 1: Return Action.ReconsumeLater and retry the message.
        return Action.ReconsumeLater;
        // Method 2: Return null and retry the message.
        return null;
        // Method 3: Throw an exception and retry the message.
        throw new RuntimeException("Consumer Message exception");
    }
}

Yeniden denemeyi devre dışı bırakma

Yeniden iletimi önlemek için tüketim mantığınızda tüm istisnaları yakalayın ve Action.CommitMessage değerini döndürün:

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        try {
            doConsumeMessage(message);
        } catch (Throwable e) {
            // Catch all exceptions and return CommitMessage to skip retry.
            return Action.CommitMessage;
        }
        // Consumption succeeded.
        return Action.CommitMessage;
    }
}

Yeniden deneme aralığını ve maksimum yeniden deneme sayısını özelleştirme

Tüketiciyi başlatmadan önce tüketici özelliklerinde MaxReconsumeTimes ve SuspendTimeMillis değerlerini ayarlayın. Özel yeniden deneme aralıkları yalnızca sıralı mesajlar için geçerlidir. Sırasız mesajlar her zaman önceden tanımlanmış zamanlamayı takip eder.

Properties properties = new Properties();
// Set the maximum number of retries to 20.
properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");
// Set the retry interval to 3,000 milliseconds (ordered messages only).
properties.put(PropertyKeyConst.SuspendTimeMillis, "3000");
Consumer consumer = ONSFactory.createConsumer(properties);

Yeniden deneme sayısını sorgulama

Bir tüketici bir mesaj aldıktan sonra, mesajın kaç kez yeniden denendiğini kontrol etmek için message.getReconsumeTimes() yöntemini çağırın:

public class MessageListenerImpl implements MessageListener {

    @Override
    public Action consume(Message message, ConsumeContext context) {
        // Get the current retry count.
        System.out.println(message.getReconsumeTimes());
        return Action.CommitMessage;
    }
}