ApsaraMQ for RocketMQ:Message retry

Temporización de reintentos

El intervalo entre dos intentos consecutivos de consumo tiene tres componentes:

Intervalo entre intentos = duración del consumo + intervalo de reintento + tiempo en estado Ready

Por ejemplo, suponga que un mensaje espera 5 segundos en el estado Ready y tarda 6 segundos en procesarse antes de fallar. Con un intervalo de reintento de 10 segundos:

1. 0 s: El mensaje entra en el estado Ready.

2. 5 s: El consumidor extrae el mensaje (Inflight).

3. 11 s: El consumo falla después de 6 segundos. El mensaje entra en WaitingRetry.

4. 21 s: Tras el intervalo de reintento de 10 segundos, el mensaje vuelve a Ready.

5. 26 s: El consumidor extrae el mensaje nuevamente para el siguiente intento.

Intervalo total: 6 + 10 + 5 = 21 segundos.

Mensajes ordenados

Mensajes no ordenados

Si MaxReconsumeTimes supera 16, el programa predefinido se aplica a los primeros 16 reintentos. Todos los reintentos posteriores utilizan un intervalo fijo de 2 horas.

Habilitar el reintento

Para activar un reintento cuando falla el consumo en el modo de consumo por clúster, implemente la interfaz MessageListener con uno de estos enfoques:

• **Devolver Action.ReconsumeLater** (recomendado)

• Devolver null

• Lanzar una excepció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");
    }
}

Deshabilitar el reintento

Para evitar la reentrega, capture todas las excepciones en su lógica de consumo y devuelva 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;
    }
}

Personalizar el intervalo de reintento y los reintentos máximos

Establezca MaxReconsumeTimes y SuspendTimeMillis en las propiedades del consumidor antes de iniciar el consumidor. Los intervalos de reintento personalizados se aplican únicamente a mensajes ordenados. Los mensajes no ordenados siempre siguen el programa predefinido.

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

Consultar el recuento de reintentos

Después de que un consumidor reciba un mensaje, llame a message.getReconsumeTimes() para comprobar cuántas veces se ha reintentado el mensaje:

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