在 Spring Cloud Stream 中，提供了通用的消费异常处理机制，可以拦截到消费者消费消息时发生的异常，进行自定义的处理逻辑。
4.1 复制项目
将消费重试小节的 [sca-stream-rocketmq-consumer-retry](，复制出 sca-stream-rocketmq-consumer-error-handler。
注意修改：<artifactId>sc-stream-rocketmq-consumer-error-handler</artifactId>.
4.2 Demo01Consumer
修改 [Demo01Consumer] 类，增加消费异常处理方法。完整代码如下：
package com.erbadagang.springcloudalibaba.stream.rocketmq.consumer.listener;
import com.erbadagang.springcloudalibaba.stream.rocketmq.consumer.message.Demo01Message;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Component;
@Component
public class Demo01Consumer {
private Logger logger = LoggerFactory.getLogger(getClass());
@StreamListener(MySink.ERBADAGANG_INPUT)// 对应 ERBADAGANG-TOPIC-01.erbadagang-consumer-group-ERBADAGANG-TOPIC-01
public void onMessage(@Payload Demo01Message message) {
logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
// <1> 注意，此处抛出一个 RuntimeException 异常，模拟消费失败
throw new RuntimeException("我就是故意抛出一个异常");
}
@ServiceActivator(inputChannel = "ERBADAGANG-TOPIC-01.erbadagang-consumer-group-ERBADAGANG-TOPIC-01.errors")
public void handleError(ErrorMessage errorMessage) {
logger.error("[handleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
logger.error("[handleError][originalMessage：{}]", errorMessage.getOriginalMessage());
logger.error("[handleError][headers：{}]", errorMessage.getHeaders());
}
@StreamListener(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) // errorChannel
public void globalHandleError(ErrorMessage errorMessage) {
logger.error("[globalHandleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
logger.error("[globalHandleError][originalMessage：{}]", errorMessage.getOriginalMessage());
logger.error("[globalHandleError][headers：{}]", errorMessage.getHeaders());
}
}
① 在 Spring Integration 的设定中，若 #onMessage(@Payload Demo01Message message) 方法消费消息发生异常时，会发送错误消息（ErrorMessage）到对应的错误 Channel（<destination>.<group>.errors）中。同时，所有错误 Channel 都桥接到了 Spring Integration 定义的全局错误 Channel(errorChannel)。
因此，我们有两种方式来实现异常处理：
- 局部的异常处理：通过订阅指定错误 Channel
- 全局的异常处理：通过订阅全局错误 Channel
② 在 #handleError(ErrorMessage errorMessage) 方法上，我们声明了 @ServiceActivator 注解，订阅指定错误 Channel的错误消息，实现 #onMessage(@Payload Demo01Message message) 方法的局部异常处理。如下图所示：
③ 在 #globalHandleError(ErrorMessage errorMessage) 方法上，我们声明了 @StreamListener 注解，订阅全局错误 Channel的错误消息，实现全局异常处理。
④ 在全局和局部异常处理都定义的情况下，错误消息仅会被符合条件的局部错误异常处理。如果没有符合条件的，错误消息才会被全局异常处理。
4.3 简单测试
① 执行 ConsumerApplication，启动消费者的实例。
② 执行 ProducerApplication，启动生产者的实例。
之后，请求 http://127.0.0.1:18080/demo01/send 接口，发送一条消息。IDEA 控制台输出日志如下：
// onMessage 方法
2020-08-06 16:47:41.195 INFO 6692 --- [MessageThread_1] c.e.s.s.r.c.listener.Demo01Consumer : [onMessage][线程编号:71 消息内容：Demo01Message{id=742970346}]
// handleError 方法
2020-08-06 16:47:41.195 ERROR 6692 --- [MessageThread_1] c.e.s.s.r.c.listener.Demo01Consumer : [handleError][payload：RuntimeException: 我就是故意抛出一个异常]
2020-08-06 16:47:41.195 ERROR 6692 --- [MessageThread_1] c.e.s.s.r.c.listener.Demo01Consumer : [handleError][originalMessage：GenericMessage [payload=byte[16], headers={rocketmq_QUEUE_ID=1, rocketmq_TOPIC=ERBADAGANG-TOPIC-01, rocketmq_FLAG=0, rocketmq_RECONSUME_TIMES=0, rocketmq_MESSAGE_ID=C0A82B7C418018B4AAC21D5A5C1D0000, rocketmq_SYS_FLAG=0, id=3c82e982-5fac-f08f-ffe7-e3956d5c6cb5, CLUSTER=DefaultCluster, rocketmq_BORN_HOST=103.3.96.229, contentType=application/json, rocketmq_BORN_TIMESTAMP=1596703661085, timestamp=1596703661195}]]
2020-08-06 16:47:41.195 ERROR 6692 --- [MessageThread_1] c.e.s.s.r.c.listener.Demo01Consumer : [handleError][headers：{id=9b060a74-d81c-eb9f-c761-2cec21958df8, timestamp=1596703661195}]
不过要注意，如果异常处理方法成功，没有重新抛出异常，会认定为该消息被消费成功，所以就不会进行消费重试。
底线
本文源代码使用 Apache License 2.0开源许可协议，这里是本文源码Gitee地址，可通过命令git clone+地址下载代码到本地，也可直接点击链接通过浏览器方式查看源代码。
