ApsaraMQ for RocketMQ:Message retry

Tempo de nova tentativa

O intervalo entre duas tentativas consecutivas de consumo possui três componentes:

Intervalo entre tentativas = duração do consumo + intervalo de nova tentativa + tempo no estado Ready

Por exemplo, suponha que uma mensagem aguarde 5 segundos no estado Ready e leve 6 segundos para ser processada antes de falhar. Com um intervalo de nova tentativa de 10 segundos:

1. 0 s — A mensagem entra no estado Ready.

2. 5 s — O consumidor puxa a mensagem (Inflight).

3. 11 s — O consumo falha após 6 segundos. A mensagem entra em WaitingRetry.

4. 21 s — Após o intervalo de nova tentativa de 10 segundos, a mensagem retorna ao estado Ready.

5. 26 s — O consumidor puxa a mensagem novamente para a próxima tentativa.

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

Mensagens ordenadas

Mensagens não ordenadas

Se MaxReconsumeTimes exceder 16, o cronograma predefinido se aplica às primeiras 16 tentativas. Todas as tentativas subsequentes usam um intervalo fixo de 2 horas.

Ative nova tentativa

Para acionar uma nova tentativa quando o consumo falhar no modo de consumo por cluster, implemente a interface MessageListener com uma das seguintes abordagens:

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

• Retornar null

• Lançar uma exceção

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

Desativar nova tentativa

Para evitar o reenvio, capture todas as exceções na lógica de consumo e retorne 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 o intervalo de nova tentativa e o máximo de tentativas

Defina MaxReconsumeTimes e SuspendTimeMillis nas propriedades do consumidor antes de iniciá-lo. Os intervalos personalizados de nova tentativa aplicam-se apenas a mensagens ordenadas. Mensagens não ordenadas sempre seguem o cronograma 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 a contagem de novas tentativas

Após receber uma mensagem, chame message.getReconsumeTimes() para verificar quantas vezes ela foi submetida a nova tentativa:

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