Spring Cloud Stream異常處理過程解析
應(yīng)用處理
當(dāng)消費(fèi)者在處理接收到的消息時(shí)，有可能會(huì)由于某些原因而拋出異常。若希望對(duì)拋出來的異常進(jìn)行處理的話，就需要采取一些異常處理手段，異常處理的方式可分為三種：應(yīng)用層面的處理、系統(tǒng)層面的處理以及通過RetryTemplate進(jìn)行處理。
本小節(jié)先來介紹較為常用的應(yīng)用層面的異常處理方式，該方式又細(xì)分為局部處理和全局處理。
局部處理
Stream相關(guān)的配置內(nèi)容如下：
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.190.129:9876
bindings:
input:
destination: stream-test-topic
group: binder-group
所謂局部處理就是針對(duì)指定的channel進(jìn)行處理，需要定義一個(gè)處理異常的方法，并在該方法上添加@ServiceActivator注解，該注解有一個(gè)inputChannel屬性，用于指定對(duì)哪個(gè)channel進(jìn)行處理，格式為{destination}.{group}.errors。具體代碼如下：
package com.zj.node.usercenter.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Service;
/**
* 消費(fèi)者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(Sink.INPUT)
public void receive1(String messageBody) {
log.info("消費(fèi)消息，messageBody = {}", messageBody);
throw new IllegalArgumentException("參數(shù)錯(cuò)誤");
}
/**
* 處理局部異常的方法
*
* @param errorMessage 異常消息對(duì)象
*/
@ServiceActivator(
// 通過特定的格式指定處理哪個(gè)channel的異常
inputChannel = "stream-test-topic.binder-group.errors"
)
public void handleError(ErrorMessage errorMessage) {
// 獲取異常對(duì)象
Throwable errorMessagePayload = errorMessage.getPayload();
log.error("發(fā)生異常", errorMessagePayload);
// 獲取消息體
Message<?> originalMessage = errorMessage.getOriginalMessage();
if (originalMessage != null) {
log.error("消息體: {}", originalMessage.getPayload());
} else {
log.error("消息體為空");
}
}
}
全局處理
全局處理則是可以處理所有channel拋出來的異常，所有的channel拋出異常后會(huì)生成一個(gè)ErrorMessage對(duì)象，即錯(cuò)誤消息。錯(cuò)誤消息會(huì)被放到一個(gè)專門的channel里，這個(gè)channel就是errorChannel。所以通過監(jiān)聽errorChannel就可以實(shí)現(xiàn)全局異常的處理。具體代碼如下：
@StreamListener(Sink.INPUT)
public void receive1(String messageBody) {
log.info("消費(fèi)消息，messageBody = {}", messageBody);
throw new IllegalArgumentException("參數(shù)錯(cuò)誤");
}
/**
* 處理全局異常的方法
*
* @param errorMessage 異常消息對(duì)象
*/
@StreamListener("errorChannel")
public void handleError(ErrorMessage errorMessage) {
log.error("發(fā)生異常. errorMessage = {}", errorMessage);
}
系統(tǒng)處理
系統(tǒng)處理方式，因消息中間件的不同而異。如果應(yīng)用層面沒有配置錯(cuò)誤處理，那么error將會(huì)被傳播給binder，而binder則會(huì)將error回傳給消息中間件。消息中間件可以選擇：
- 丟棄消息：錯(cuò)誤消息將被丟棄。雖然在某些情況下可以接受，但這種方式一般不適用于生產(chǎn)
- requeue（重新排隊(duì)，從而重新處理）
- 將失敗的消息發(fā)送給DLQ（死信隊(duì)列）
DLQ
目前RabbitMQ對(duì)DLQ的支持比較好，這里以RabbitMQ為例，只需要添加DLQ相關(guān)的配置：
spring:
cloud:
stream:
bindings:
input:
destination: stream-test-topic
group: binder-group
rabbit:
bindings:
input:
consumer:
# 自動(dòng)將失敗的消息發(fā)送給DLQ
auto-bind-dlq: true
消息消費(fèi)失敗后，就會(huì)放入死信隊(duì)列。在控制臺(tái)操作一下，即可將死信放回消息隊(duì)列，這樣，客戶端就可以重新處理。
如果想獲取原始錯(cuò)誤的異常堆棧，可添加如下配置：
spring:
cloud:
stream:
rabbit:
bindings:
input:
consumer:
republish-to-dlq: true
requeue
Rabbit及Kafka的binder依賴RetryTemplate實(shí)現(xiàn)消息重試，從而提升消息處理的成功率。然而，如果設(shè)置了spring.cloud.stream.bindings.input.consumer.max-attempts=1 ，那么RetryTemplate則不會(huì)再重試。此時(shí)可以通過requeue方式來處理異常。
需要添加如下配置：
# 默認(rèn)是3，設(shè)為1則禁用重試 spring.cloud.stream.bindings.<input channel名稱>.consumer.max-attempts=1 # 表示是否要requeue被拒絕的消息(即：requeue處理失敗的消息) spring.cloud.stream.rabbit.bindings.input.consumer.requeue-rejected=true
這樣，失敗的消息將會(huì)被重新提交到同一個(gè)handler進(jìn)行處理，直到handler拋出 AmqpRejectAndDontRequeueException 異常為止。
RetryTemplate
RetryTemplate主要用于實(shí)現(xiàn)消息重試，也是錯(cuò)誤處理的一種手段。有兩種配置方式，一種是通過配置文件進(jìn)行配置，如下示例：
spring:
cloud:
stream:
bindings:
<input channel名稱>:
consumer:
# 最多嘗試處理幾次，默認(rèn)3
maxAttempts: 3
# 重試時(shí)初始避退間隔，單位毫秒，默認(rèn)1000
backOffInitialInterval: 1000
# 重試時(shí)最大避退間隔，單位毫秒，默認(rèn)10000
backOffMaxInterval: 10000
# 避退乘數(shù)，默認(rèn)2.0
backOffMultiplier: 2.0
# 當(dāng)listen拋出retryableExceptions未列出的異常時(shí)，是否要重試
defaultRetryable: true
# 異常是否允許重試的map映射
retryableExceptions:
java.lang.RuntimeException: true
java.lang.IllegalStateException: false
另一種則是通過代碼配置，在多數(shù)場景下，使用配置文件定制重試行為都是可以滿足需求的，但配置文件里支持的配置項(xiàng)可能無法滿足一些復(fù)雜需求。此時(shí)可使用代碼方式配置RetryTemplate，如下示例：
@Configuration
class RetryConfiguration {
@StreamRetryTemplate
public RetryTemplate sinkConsumerRetryTemplate() {
RetryTemplate retryTemplate = new RetryTemplate();
retryTemplate.setRetryPolicy(retryPolicy());
retryTemplate.setBackOffPolicy(backOffPolicy());
return retryTemplate;
}
private ExceptionClassifierRetryPolicy retryPolicy() {
BinaryExceptionClassifier keepRetryingClassifier = new BinaryExceptionClassifier(
Collections.singletonList(IllegalAccessException.class
));
keepRetryingClassifier.setTraverseCauses(true);
SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy(3);
AlwaysRetryPolicy alwaysRetryPolicy = new AlwaysRetryPolicy();
ExceptionClassifierRetryPolicy retryPolicy = new ExceptionClassifierRetryPolicy();
retryPolicy.setExceptionClassifier(
classifiable -> keepRetryingClassifier.classify(classifiable) ?
alwaysRetryPolicy : simpleRetryPolicy);
return retryPolicy;
}
private FixedBackOffPolicy backOffPolicy() {
final FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
backOffPolicy.setBackOffPeriod(2);
return backOffPolicy;
}
}
最后還需要添加一段配置：
spring.cloud.stream.bindings.<input channel名稱>.consumer.retry-template-name=myRetryTemplate
注：Spring Cloud Stream 2.2才支持設(shè)置retry-template-name
以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
- SpringCloud Stream使用解析
- SpringCloud Stream消息驅(qū)動(dòng)實(shí)例詳解
- spring-cloud-stream結(jié)合kafka使用詳解
- Spring Cloud Stream微服務(wù)消息框架原理及實(shí)例解析
- Spring Cloud Stream分區(qū)分組原理圖解
- Spring Cloud Stream如何實(shí)現(xiàn)服務(wù)之間的通訊
- 詳解Spring Cloud Stream使用延遲消息實(shí)現(xiàn)定時(shí)任務(wù)(RabbitMQ)
- Spring Cloud Stream簡單用法
相關(guān)文章
這篇文章主要介紹了Java注釋代碼執(zhí)行方法解析,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-05-05
大家好，本篇文章主要講的是java面向?qū)ο罄^承與多態(tài)介紹，感興趣的同學(xué)趕快來看一看吧，對(duì)你有幫助的話記得收藏一下2022-01-01
今天小編就為大家分享一篇關(guān)于Java分布式鎖的概念與實(shí)現(xiàn)方式詳解，小編覺得內(nèi)容挺不錯(cuò)的，現(xiàn)在分享給大家，具有很好的參考價(jià)值，需要的朋友一起跟隨小編來看看吧2021-08-08
這篇文章主要介紹了Java中的接口和抽象類用法,結(jié)合實(shí)例形式較為詳細(xì)的分析了Java中關(guān)于接口和抽象類的概念、定義、用法與相關(guān)注意事項(xiàng),需要的朋友可以參考下2015-12-12
Jenkins如何使用DockerFile自動(dòng)部署Java項(xiàng)目
這篇文章主要介紹了Jenkins如何使用DockerFile自動(dòng)部署Java項(xiàng)目,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-08-08
SpringBoot+MinIO實(shí)現(xiàn)對(duì)象存儲(chǔ)方式
這篇文章主要介紹了SpringBoot+MinIO實(shí)現(xiàn)對(duì)象存儲(chǔ)方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-08-08
log4j如何根據(jù)變量動(dòng)態(tài)生成文件名
這篇文章主要介紹了log4j如何根據(jù)變量動(dòng)態(tài)生成文件名方式，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-12-12
SpringBoot集成WebSocket的兩種方式(JDK內(nèi)置版和Spring封裝版)
這篇文章主要介紹了SpringBoot集成WebSocket的兩種方式,這兩種方式為JDK內(nèi)置版和Spring封裝版,本文結(jié)合示例代碼給大家介紹的非常詳細(xì)，需要的朋友可以參考下2023-06-06
Spring在注入bean的時(shí)候會(huì)做循環(huán)依賴檢查，下面這篇文章主要給大家介紹了關(guān)于Spring中循環(huán)依賴的相關(guān)資料，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面來一起看看吧2018-05-05
Spring Boot 配置 IDEA和DevTools 熱部署的方法
這篇文章主要介紹了Spring Boot 配置 IDEA和DevTools 熱部署的方法,需要的朋友可以參考下2018-02-02