Spring Cloud Gateway-獲取body踩坑實踐
問題1:無法獲取body內容
問題原因分析
在使用過程中碰到過濾器中獲取的內容一直都是空的,嘗試了網上的各種解析body內容的方法,但是得到結果都是一樣,死活獲取不到body數據,一度很崩潰。后來進行了各種嘗試,最終發現使用不同的spring boot版本和spring cloud版本,對結果影響很大。
最佳實踐
方案1:降低版本
springboot版本:2.0.5-RELEASE
springcloud版本:Finchley.RELEASE
使用以上的版本會報以下的錯誤:
java.lang.IllegalStateException: Only one connection receive subscriber allowed.
原因在於spring boot在2.0.5版本如果使用了WebFlux就自動配置HiddenHttpMethodFilter過濾器。查看源碼發現,這個過濾器的作用是,針對當前的瀏覽器一般只支持GET和POST表單提交方法,如果想使用其他HTTP方法(如:PUT、DELETE、PATCH),就只能通過一個隱藏的屬性如(_method=PUT)來表示,那么HiddenHttpMethodFilter的作用是將POST請求的_method參數里面的value替換掉http請求的方法。但是這就導致已經讀取了一次body,導致后面的過濾器無法讀取body。解決方案就是可以自己重寫HiddenHttpMethodFilter來覆蓋原來的實現,實際上gateway本身就不應該做這種事情,原始請求是怎樣的,轉發給下游的請求就應該是怎樣的。
@Bean
public HiddenHttpMethodFilter hiddenHttpMethodFilter() {
return new HiddenHttpMethodFilter() {
@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
return chain.filter(exchange);
}
};
}
這個方案也是gateway官方開發者目前所提出的解決方案。
方案2:不降低版本,緩存body內容
springboot版本:2.1.5-RELEASE
springcloud版本:Greenwich.SR1
在較高版本中,上面的方法已經行不通了,可以自定義一個高優先級的過濾器先獲取body內容並緩存起來,解決body只能讀取一次的問題。具體解決方案見問題2。
問題2:body只能讀取一次
這個問題網上主要的解決思路就是獲取body之后,重新封裝request,然后把封裝后的request傳遞下去。思路很清晰,但是實現的方式卻千奇百怪。在使用的過程中碰到了各種千奇百怪的問題,比如說第一次請求正常,第二次請求報400錯誤,這樣交替出現。最終定位原因就是我自定義的全局過濾器把request重新包裝導致的,去掉就好了。鑒於踩得坑比較多,下面給出在實現過程中筆者認為的最佳實踐。
核心代碼
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/**
* 這個過濾器解決body不能重復讀的問題
* 實際上這里沒必要把body的內容放到attribute中去,因為從attribute取出body內容還是需要強轉成
* Flux<DataBuffer>,然后轉換成String,和直接讀取body沒有什么區別
*/
@Component
public class CacheBodyGlobalFilter implements Ordered, GlobalFilter {
// public static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject";
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
if (exchange.getRequest().getHeaders().getContentType() == null) {
return chain.filter(exchange);
} else {
return DataBufferUtils.join(exchange.getRequest().getBody())
.flatMap(dataBuffer -> {
DataBufferUtils.retain(dataBuffer);
Flux<DataBuffer> cachedFlux = Flux
.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));
ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(
exchange.getRequest()) {
@Override
public Flux<DataBuffer> getBody() {
return cachedFlux;
}
};
// exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, cachedFlux);
return chain.filter(exchange.mutate().request(mutatedRequest).build());
});
}
}
@Override
public int getOrder() {
return Ordered.HIGHEST_PRECEDENCE;
}
}
CacheBodyGlobalFilter這個全局過濾器的目的就是把原有的request請求中的body內容讀出來,並且使用ServerHttpRequestDecorator這個請求裝飾器對request進行包裝,重寫getBody方法,並把包裝后的請求放到過濾器鏈中傳遞下去。這樣后面的過濾器中再使用exchange.getRequest().getBody()來獲取body時,實際上就是調用的重載后的getBody方法,獲取的最先已經緩存了的body數據。這樣就能夠實現body的多次讀取了。
值得一提的是,這個過濾器的order設置的是Ordered.HIGHEST_PRECEDENCE,即最高優先級的過濾器。優先級設置這么高的原因是某些系統內置的過濾器可能也會去讀body,這樣就會導致我們自定義過濾器中獲取body的時候報body只能讀取一次這樣的錯誤如下:
java.lang.IllegalStateException: Only one connection receive subscriber allowed.
at reactor.ipc.netty.channel.FluxReceive.startReceiver(FluxReceive.java:279)
at reactor.ipc.netty.channel.FluxReceive.lambda$subscribe$2(FluxReceive.java:129)
at
所以需要先解決body只能讀取一次的問題,把CacheBodyGlobalFilter的優先級設到最高。
import io.netty.buffer.ByteBufAllocator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
/**
* @author mjw
* @date 2020/3/24
*/
@Component
@Slf4j
public class AuthGlobalFilter implements GlobalFilter, Ordered
{
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain)
{
String bodyContent = RequestUtil.resolveBodyFromRequest(exchange.getRequest());
// TODO 身份認證相關邏輯
return chain.filter(exchange.mutate().build());
}
@Override
public int getOrder()
{
return -100;
}
}
這個類是自定義的身份認證的全局過濾器,這里需要說一下的就是讀取body之后如何解析。由於spring cloud gateway使用的是webFlux,因此獲取的body內容是Flux
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import reactor.core.publisher.Flux;
import java.nio.charset.StandardCharsets;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
* @author mjw
* @date 2020/3/30
*/
public class RequestUtil
{
/**
* 讀取body內容
* @param serverHttpRequest
* @return
*/
public static String resolveBodyFromRequest(ServerHttpRequest serverHttpRequest){
//獲取請求體
Flux<DataBuffer> body = serverHttpRequest.getBody();
StringBuilder sb = new StringBuilder();
body.subscribe(buffer -> {
byte[] bytes = new byte[buffer.readableByteCount()];
buffer.read(bytes);
// DataBufferUtils.release(buffer);
String bodyString = new String(bytes, StandardCharsets.UTF_8);
sb.append(bodyString);
});
return formatStr(sb.toString());
}
/**
* 去掉空格,換行和制表符
* @param str
* @return
*/
private static String formatStr(String str){
if (str != null && str.length() > 0) {
Pattern p = Pattern.compile("\\s*|\t|\r|\n");
Matcher m = p.matcher(str);
return m.replaceAll("");
}
return str;
}
}
實際上在網上查找資料的過程中發現,解析body內容網上普遍提到兩種方式,一種就是上文中的方式,讀取字節方式拼接字符串,另一種方式如下:
private String getBodyContent(ServerWebExchange exchange){
Flux<DataBuffer> body = exchange.getRequest().getBody();
AtomicReference<String> bodyRef = new AtomicReference<>();
// 緩存讀取的request body信息
body.subscribe(dataBuffer -> {
CharBuffer charBuffer = StandardCharsets.UTF_8.decode(dataBuffer.asByteBuffer());
DataBufferUtils.release(dataBuffer);
bodyRef.set(charBuffer.toString());
});
//獲取request body
return bodyRef.get();
}
但是網上有網友說這種方式最多能獲取1024字節的數據,數據過長會被截斷,導致數據丟失。這里筆者沒有親自驗證過,只是把這種方式提供在這里供大家參考。
另外需要注意的是在我們創建ByteBuf對象后,它的引用計數是1,當你每次調用DataBufferUtils.release之后會釋放引用計數對象時,它的引用計數減1,如果引用計數為0,這個引用計數對象會被釋放(deallocate),並返回對象池。當嘗試訪問引用計數為0的引用計數對象會拋出IllegalReferenceCountException異常如下:
io.netty.util.IllegalReferenceCountException: refCnt: 0
at io.netty.buffer.AbstractByteBuf.ensureAccessible(AbstractByteBuf.java:1423) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
at io.netty.buffer.UnpooledHeapByteBuf.capacity(UnpooledHeapByteBuf.java:102) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
at io.netty.buffer.ReadOnlyByteBuf.capacity(ReadOnlyByteBuf.java:408) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
at io.netty.buffer.AbstractByteBuf.setIndex(AbstractByteBuf.java:126) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
at io.netty.buffer.ReadOnlyByteBuf.<init>(ReadOnlyByteBuf.java:50) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
at io.netty.buffer.ReadOnlyByteBuf.duplicate(ReadOnlyByteBuf.java:278) ~[netty-all-4.1.0.Final.jar:4.1.0.Final]
因此這里為了能夠在多個自定義過濾器中使用相同的方法來獲取body數據,就不進行release了。