20、RocketMQ 实战 - RocketMQ5.x消费重试

消费重试指的是,消费者在消费某条消息失败后,Apache RocketMQ 服务端会根据重试策略重新消费该消息,超过一次定数后若还未消费成功,则该消息将不再继续重试,直接被发送到死信队列中。Apache RocketMQ 的消费重试主要解决的是业务处理逻辑失败导致的消费完整性问题,是一种为业务兜底的策略,不应该被用做业务流程控制。建议以下消费失败场景使用重试机制:

  • 业务处理失败,且失败原因跟当前的消息内容相关,比如该消息对应的事务状态还未获取到,预期一段时间后可执行成功。
  • 消费失败的原因不会导致连续性,即当前消息消费失败是一个小概率事件,不是常态化的失败,后面的消息大概率会消费成功。此时可以对当前消息进行重试,避免进程阻塞。
消息重试的触发条件
  • 消费失败,包括消费者返回消息失败状态标识或抛出非预期异常。
  • 消息处理超时,包括在PushConsumer中排队超时。

PushConsumer消费重试策略

PushConsumer消费消息时,消息的几个主要状态如下:

 

  • Ready:已就绪状态。消息在Apache RocketMQ服务端已就绪,可以被消费者消费。
  • Inflight:处理中状态。消息被消费者客户端获取,处于消费中还未返回消费结果的状态。
  • WaitingRetry:待重试状态,PushConsumer独有的状态。当消费者消息处理失败或消费超时,会触发消费重试逻辑判断。如果当前重试次数未达到最大次数,则该消息变为待重试状态,经过重试间隔后,消息将重新变为已就绪状态可被重新消费。多次重试之间,可通过重试间隔进行延长,防止无效高频的失败。
  • Commit:提交状态。消费成功的状态,消费者返回成功响应即可结束消息的状态机。
  • DLQ:死信状态。消费逻辑的最终兜底机制,若消息一直处理失败并不断进行重试,直到超过最大重试次数还未成功,此时消息不会再重试,会被投递至死信队列。您可以通过消费死信队列的消息进行业务恢复。

两次消费间的间隔时间实际由消费耗时及重试间隔控制。PushConsumer的最大重试次数由消费者分组创建时的元数据控制,默认16次,例如,最大重试次数为3次,则该消息最多可被投递4次,1次为原始消息,3次为重试投递次数。

重试间隔时间
  • 无序消息(非顺序消息):重试间隔为阶梯时间,若重试次数超过16次,后面每次重试间隔都为2小时。具体时间如下:
第几次重试 与上次重试的间隔时间 第几次重试 与上次重试的间隔时间
1 10秒 9 7分钟
2 30秒 10 8分钟
3 1分钟 11 9分钟
4 2分钟 12 10分钟
5 3分钟 13 20分钟
6 4分钟 14 30分钟
7 5分钟 15 1小时
8 6分钟 16 2小时
  • 顺序消息:重试间隔为固定时间,默认3000毫秒。

PushConsumer触发消息重试只需要返回消费失败的状态码即可,当出现非预期的异常时,也会被SDK捕获:

provider.newPushConsumerBuilder()
                .setClientConfiguration(configuration)
                // 设置消费者分组。
                .setConsumerGroup("my-consumer")
                // 设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMaxCacheMessageSizeInBytes(5 * 1024 * 1024)
                // 设置消费监听器。
                .setMessageListener(messageView -> {
                   
                    log.info("消费时间:{},消息内容为:{}", LocalDateTime.now(), uncompress(StandardCharsets.UTF_8.decode(messageView.getBody()).toString()));
                    return ConsumeResult.FAILURE;
                }).build();

返回消费失败的状态,发送两条消息,发现一直在重试,同时发现重试间隔时间越来越长:

 

provider.newPushConsumerBuilder()
                .setClientConfiguration(configuration)
                // 设置消费者分组。
                .setConsumerGroup("my-consumer")
                // 设置预绑定的订阅关系。
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMaxCacheMessageSizeInBytes(5 * 1024 * 1024)
                // 设置消费监听器。
                .setMessageListener(messageView -> {
                      String body = StandardCharsets.UTF_8.decode(messageView.getBody()).toString();
                    log.info("消费时间:{},消息内容为:{}", LocalDateTime.now(), uncompress(body));
                    int i = Integer.parseInt(body);
                    log.info("数字:{}", i);
                }).build();

这里Integer.parseInt(body);模拟出现消费异常。当消息体不全是数字时会报错。首先在Rocket dashboard中发送一条消息内容为123的消息,一条消息内容为qwer的消息。

 

发现消息内容为qwer的消息在重试消费。

SimpleConsumer消费重试策略

SimpleConsumer消费消息时,消息的几个主要状态如下:

 

  • Ready:已就绪状态。消息在Apache RocketMQ服务端已就绪,可以被消费者消费。
  • Inflight:处理中状态。消息被消费者客户端获取,处于消费中还未返回消费结果的状态。
  • Commit:提交状态。消费成功的状态,消费者返回成功响应即可结束消息的状态机。
  • DLQ:死信状态。消费逻辑的最终兜底机制,若消息一直处理失败并不断进行重试,直到超过最大重试次数还未成功,此时消息不会再重试,会被投递至死信队列。您可以通过消费死信队列的消息进行业务恢复。

和PushConsumer消费重试策略不同的是,SimpleConsumer消费者的重试间隔是预分配的,每次获取消息消费者会在调用API时设置一个不可见时间参数 InvisibleDuration,即消息的最大处理时长。若消息消费失败触发重试,不需要设置下一次重试的时间间隔,直接复用不可见时间参数的取值。由于不可见时间为预分配的,可能和实际业务中的消息处理时间差别较大,您可以通过API接口修改不可见时间。例如,您预设消息处理耗时最多20 ms,但实际业务中20 ms内消息处理不完,您可以修改消息不可见时间,延长消息处理时间,避免消息触发重试机制。

simpleConsumer.receive(10, Duration.ofSeconds(30));

消息不可见时间在调用SimpleConsumer的receive方法时设置,这里设置30秒。

修改消息不可见时间需要满足以下条件:

  • 消息处理未超时
  • 消息处理未提交消费状态

如下图所示,消息不可见时间修改后立即生效,即从调用API时刻开始,重新计算消息不可见时间。

 

SimpleConsumer的最大重试次数由消费者分组创建时的元数据控制,默认16次,消息重试间隔=不可见时间-消息实际处理时长。SimpleConsumer 的消费重试间隔通过消息的不可见时间控制。例如,消息不可见时间为30 ms,实际消息处理用了10 ms就返回失败响应,则距下次消息重试还需要20 ms,此时的消息重试间隔即为20 ms;若直到30 ms消息还未处理完成且未返回结果,则消息超时,立即重试,此时重试间隔即为0 ms。

 List<MessageView> messageViewList =  simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        log.info("消费消息:{}", StandardCharsets.UTF_8.decode(messageView.getBody()).toString());
        try {
            //消费处理完成后,需要主动调用ACK提交消费结果。
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });

要想消费重试,不调用simpleConsumer.ack提交消费结果即可。注释 simpleConsumer.ack(messageView);来模拟消费重试。发送一条消息后,查看消费结果:

 

触发了消费重试。