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小时。具体时间如下：

• 顺序消息：重试间隔为固定时间，默认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);来模拟消费重试。发送一条消息后，查看消费结果：

触发了消费重试。