RocketMQ 与 Spring Cloud Stream整合(三、消费重试)

RocketMQ 提供消费重试的机制。在消息消费失败的时候,RocketMQ 会通过消费重试机制,重新投递该消息给 Consumer ,让 Consumer 有机会重新消费消息,实现消费成功。

当然,RocketMQ 并不会无限重新投递消息给 Consumer 重新消费,而是在默认情况下,达到 16 次重试次数时,Consumer 还是消费失败时,该消息就会进入到死信队列

死信队列用于处理无法被正常消费的消息。当一条消息初次消费失败,消息队列会自动进行消息重试;达到最大重试次数后,若消费依然失败,则表明消费者在正常情况下无法正确地消费该消息,此时,消息队列不会立刻将消息丢弃,而是将其发送到该消费者对应的特殊队列中。

RocketMQ 将这种正常情况下无法被消费的消息称为死信消息(Dead-Letter Message),将存储死信消息的特殊队列称为死信队列(Dead-Letter Queue)。在 RocketMQ 中,可以通过使用 console 控制台对死信队列中的消息进行重发来使得消费者实例再次进行消费。

每条消息的失败重试,是有一定的间隔时间。实际上,消费重试是基于上篇文章「 定时消息」 来实现,第一次重试消费按照延迟级别为 3 开始。所以,默认为 16 次重试消费,也非常好理解,毕竟延迟级别最高为 18 呀。

不过要注意,只有集群消费模式下,才有消息重试。

下面,我们来搭建一个 RocketMQ 消息重试的使用示例。考虑方便,我们直接复用 快速入门小节的项目,使用 [sca-stream-rocketmq-producer]发送消息,从 sca-stream-rocketmq-consumer 复制出 sca-stream-rocketmq-consumer-retry模拟消费失败后的重试

3.1 复制项目

sca-stream-rocketmq-consumer 复制出 sca-stream-rocketmq-consumer-retry。注意修改:<artifactId>sc-stream-rocketmq-consumer-retry</artifactId>。

3.2 配置文件

修改 [application.yml]配置文件,增加消费重试相关的两个配置项 delay-level-when-next-consumemax-attempts。最终配置如下:

spring:
  application:
    name: erbadagang-consumer-application
  cloud:
    # Spring Cloud Stream 配置项,对应 BindingServiceProperties 类
    stream:
      # Binding 配置项,对应 BindingProperties Map
      bindings:
        erbadagang-input:
          destination: ERBADAGANG-TOPIC-01 # 目的地。这里使用 RocketMQ Topic
          content-type: application/json # 内容格式。这里使用 JSON
          group: erbadagang-consumer-group-ERBADAGANG-TOPIC-01 # 消费者分组,命名规则:组名+topic名
          # Consumer 配置项,对应 ConsumerProperties 类
          consumer:
            max-attempts: 1

        trek-input:
          destination: TREK-TOPIC-01 # 目的地。这里使用 RocketMQ Topic
          content-type: application/json # 内容格式。这里使用 JSON
          group: trek-consumer-group-TREK-TOPIC-01 # 消费者分组,命名规则:组名+topic名
      # Spring Cloud Stream RocketMQ 配置项
      rocketmq:
        # RocketMQ Binder 配置项,对应 RocketMQBinderConfigurationProperties 类
        binder:
          name-server: 101.133.227.13:9876 # RocketMQ Namesrv 地址
        # RocketMQ 自定义 Binding 配置项,对应 RocketMQBindingProperties Map
        bindings:
          erbadagang-input:
            # RocketMQ Consumer 配置项,对应 RocketMQConsumerProperties 类
            consumer:
              enabled: true # 是否开启消费,默认为 true
              broadcasting: false # 是否使用广播消费,默认为 false 使用集群消费
              delay-level-when-next-consume: 0 # 异步消费消息模式下消费失败重试策略,默认为 0

server:
  port: ${random.int[10000,19999]} # 随机端口,方便启动多个消费者

① 对于 delay-level-when-next-consume 配置项,一共有三种选择:

-1:不重复,直接放入死信队列
0:RocketMQ Broker 控制重试策略
>0:RocketMQ Consumer 控制重试策略

可能对 Broker 和 Consumer 控制重试策略有点懵,每天消息首次消费失败时,Consumer 会发回给 Broker,并告诉 Broker 按照什么延迟级别开始,不断重新投递给 Consumer 直到消费成功或者到达最大延迟级别。

举个例子,如果这里我们设置了delay-level-when-next-consume配置项为 18,则 2 小时后 Broker 会投递该消息给 Consumer 进行重新消费。

一般情况下,我们设置 delay-level-when-next-consume 配置项为 0 即可,使用 Broker 控制重试策略即可。默认配置下,Broker 会使用延迟级别从 3 开始,10 秒后 Broker 会投递该消息给 Consumer 进行重新消费。

② 对于 max-attempts 配置项,每次拉取到消息到本地时,如果消费重试,本地重试的最大总次数(包括第一次)。这个是 Spring Cloud Stream 提供的通用消费重试功能,是 Consumer 级别的,而 RocketMQ 提供的独有消费重试功能,是 Broker 级别的。

因为 Spring Cloud Stream 提供的重试间隔,是通过 sleep 实现,会占掉当前线程,影响 Consumer 的消费速度,所以这里并不推荐使用,因此设置 max-attempts 配置项为 1,禁用 Spring Cloud Stream 提供的重试功能,使用 RocketMQ 提供的重试功能。

友情提示:如果无法保证消费重试不会带来副作用,也就是说无法保证消费的幂等性,建议关闭消费重试功能,即设置delay-level-when-next-consume配置项为 -1,max-attempts 配置项为 1。

3.3 Demo01Consumer

修改 [Demo01Consumer] 类,在消费消息时抛出异常,从而模拟消费错误。代码如下:

package com.erbadagang.springcloudalibaba.stream.rocketmq.consumer.listener;

import com.erbadagang.springcloudalibaba.stream.rocketmq.consumer.message.Demo01Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

@Component
public class Demo01Consumer {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @StreamListener(MySink.ERBADAGANG_INPUT)
    public void onMessage(@Payload Demo01Message message) {
        logger.info("[onMessage][线程编号:{} 消息内容:{}]", Thread.currentThread().getId(), message);
        // <1> 注意,此处抛出一个 RuntimeException 异常,模拟消费失败
        throw new RuntimeException("我就是故意抛出一个异常");
    }

3.4 简单测试

① 执行retry这个新项目的 ConsumerApplication,启动消费者的实例。
② 执行 ProducerApplication,启动生产者的实例。

之后,请求 http://127.0.0.1:18080/demo01/send 接口,发送一条消息。IDEA 控制台输出日志如下:

// Demo01Consumer 第一次消费失败,抛出 RuntimeException 异常
2020-08-06 15:16:44.845  INFO 2056 --- [MessageThread_1] c.e.s.s.r.c.listener.Demo01Consumer      : [onMessage][线程编号:78 消息内容:Demo01Message{id=174496181}]
2020-08-06 15:16:44.861 ERROR 2056 --- [MessageThread_1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Exception thrown while invoking Demo01Consumer#onMessage[1 args]; nested exception is java.lang.RuntimeException: 我就是故意抛出一个异常, failedMessage=GenericMessage [payload=byte[16], headers={rocketmq_QUEUE_ID=1, rocketmq_TOPIC=ERBADAGANG-TOPIC-01, rocketmq_FLAG=0, rocketmq_RECONSUME_TIMES=0, rocketmq_MESSAGE_ID=C0A82B7C1F0818B4AAC21D0719CF0001, rocketmq_SYS_FLAG=0, id=9d459a6f-da60-33a1-e46f-c90ec349634f, CLUSTER=DefaultCluster, rocketmq_BORN_HOST=103.3.96.229, contentType=application/json, rocketmq_BORN_TIMESTAMP=1596698204623, timestamp=1596698204814}]
Caused by: java.lang.RuntimeException: 我就是故意抛出一个异常
    at com.erbadagang.springcloudalibaba.stream.rocketmq.consumer.listener.Demo01Consumer.onMessage(Demo01Consumer.java:19)

// Demo01Consumer 第一次重试消费失败,抛出 RuntimeException 异常。间隔了 10 秒,对应延迟级别 3 。
2020-08-06 15:16:55.005  INFO 2056 --- [MessageThread_1] c.e.s.s.r.c.listener.Demo01Consumer      : [onMessage][线程编号:78 消息内容:Demo01Message{id=174496181}]
2020-08-06 15:16:55.005 ERROR 2056 --- [MessageThread_1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessagingException: Exception thrown while invoking Demo01Consumer#onMessage[1 args]; nested exception is java.lang.RuntimeException: 我就是故意抛出一个异常, failedMessage=GenericMessage [payload=byte[16], headers={rocketmq_QUEUE_ID=0, rocketmq_TOPIC=ERBADAGANG-TOPIC-01, rocketmq_FLAG=0, rocketmq_RECONSUME_TIMES=1, rocketmq_MESSAGE_ID=C0A82B7C1F0818B4AAC21D0719CF0001, rocketmq_SYS_FLAG=0, id=f2c078ec-784f-38f7-7c46-f5584597a66b, CLUSTER=DefaultCluster, rocketmq_BORN_HOST=103.3.96.229, contentType=application/json, rocketmq_BORN_TIMESTAMP=1596698204623, timestamp=1596698215005}]
......

从日志中,我们可以看到,消息因为消费失败后,又重试消费了多次。

底线


本文源代码使用 Apache License 2.0开源许可协议,这里是本文源码Gitee地址,可通过命令git clone+地址下载代码到本地,也可直接点击链接通过浏览器方式查看源代码。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,324评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,303评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,192评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,555评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,569评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,566评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,927评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,583评论 0 257
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,827评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,590评论 2 320
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,669评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,365评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,941评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,928评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,159评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,880评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,399评论 2 342