ApsaraMQ for RocketMQ:Message retry

Chronologie des nouvelles tentatives

L'intervalle entre deux tentatives de consommation consécutives comprend trois composantes :

Intervalle entre les tentatives = durée de consommation + intervalle de nouvelle tentative + temps passé dans l'état Ready

Par exemple, supposons qu'un message attende 5 secondes dans l'état Ready et prenne 6 secondes pour être traité avant d'échouer. Avec un intervalle de nouvelle tentative de 10 secondes :

1. 0 s : Le message passe à l'état Ready.

2. 5 s : Le consommateur extrait le message (Inflight).

3. 11 s : La consommation échoue après 6 secondes. Le message passe à l'état WaitingRetry.

4. 21 s : Après l'intervalle de nouvelle tentative de 10 secondes, le message revient à l'état Ready.

5. 26 s : Le consommateur extrait à nouveau le message pour la tentative suivante.

Intervalle total : 6 + 10 + 5 = 21 secondes.

Messages ordonnés

Messages non ordonnés

Si MaxReconsumeTimes dépasse 16, le planning prédéfini s'applique aux 16 premières tentatives. Toutes les tentatives suivantes utilisent un intervalle fixe de 2 heures.

Activer la nouvelle tentative

Pour déclencher une nouvelle tentative en cas d'échec de la consommation en mode de consommation en cluster, implémentez l'interface MessageListener en utilisant l'une des approches suivantes :

• **Renvoyer Action.ReconsumeLater** (recommandé)

• Renvoyer null

• Lever une exception

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");
    }
}

Désactiver la nouvelle tentative

Pour empêcher la redistribution, interceptez toutes les exceptions dans votre logique de consommation et renvoyez Action.CommitMessage :

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;
    }
}

Personnaliser l'intervalle de nouvelle tentative et le nombre maximal de tentatives

Définissez MaxReconsumeTimes et SuspendTimeMillis dans les propriétés du consommateur avant de démarrer ce dernier. Les intervalles de nouvelle tentative personnalisés s'appliquent uniquement aux messages ordonnés. Les messages non ordonnés suivent toujours le planning prédéfini.

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);

Consulter le nombre de nouvelles tentatives

Une fois qu'un consommateur a reçu un message, appelez message.getReconsumeTimes() pour vérifier le nombre de fois où le message a fait l'objet d'une nouvelle tentative :

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;
    }
}