Spring Cloud Gateway 獲取請求體
一、直接在全局攔截器中獲取，偽代碼如下
private String resolveBodyFromRequest(ServerHttpRequest serverHttpRequest){
Flux<DataBuffer> body = serverHttpRequest.getBody();
AtomicReference<String> bodyRef = new AtomicReference<>();
body.subscribe(buffer -> {
CharBuffer charBuffer = StandardCharsets.UTF_8.decode(buffer.asByteBuffer());
DataBufferUtils.release(buffer);
bodyRef.set(charBuffer.toString());
});
return bodyRef.get();
}
存在的缺陷：其他攔截器無法再通過該方式獲取請求體（因為請求體已被消費），並且會拋出異常
Only one connection receive subscriber allowed.Caused by: java.lang.IllegalStateException: Only one connection receive subscriber allowed.
異常原因：實際上spring-cloud-gateway反向代理的原理是，首先讀取原請求的數據，然后構造一個新的請求，將原請求的數據封裝到新的請求中，然后再轉發出去。然而我們在他封裝之前讀取了一次request body，而request body只能讀取一次。因此就出現了上面的錯誤。
再者受版本限制
這種方法在spring-boot-starter-parent 2.0.6.RELEASE + Spring Cloud Finchley.SR2 body 中生效，
但是在spring-boot-starter-parent 2.1.0.RELEASE + Spring Cloud Greenwich.M3 body 中不生效，總是為空
二、先在全局過濾器中獲取，然后再把request重新包裝，繼續向下傳遞傳遞
@Override
public GatewayFilter apply(NameValueConfig nameValueConfig) {
return (exchange, chain) -> {
URI uri = exchange.getRequest().getURI();
URI ex = UriComponentsBuilder.fromUri(uri).build(true).toUri();
ServerHttpRequest request = exchange.getRequest().mutate().uri(ex).build();
if("POST".equalsIgnoreCase(request.getMethodValue())){//判斷是否為POST請求
Flux<DataBuffer> body = request.getBody();
AtomicReference<String> bodyRef = new AtomicReference<>();
body.subscribe(dataBuffer -> {
CharBuffer charBuffer = StandardCharsets.UTF_8.decode(dataBuffer.asByteBuffer());
DataBufferUtils.release(dataBuffer);
bodyRef.set(charBuffer.toString());
});//讀取request body到緩存
String bodyStr = bodyRef.get();//獲取request body
System.out.println(bodyStr);//這里是我們需要做的操作
DataBuffer bodyDataBuffer = stringBuffer(bodyStr);
Flux<DataBuffer> bodyFlux = Flux.just(bodyDataBuffer);
request = new ServerHttpRequestDecorator(request){
@Override
public Flux<DataBuffer> getBody() {
return bodyFlux;
}
};//封裝我們的request
}
return chain.filter(exchange.mutate().request(request).build());
};
}
protected DataBuffer stringBuffer(String value) {
byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
NettyDataBufferFactory nettyDataBufferFactory = new NettyDataBufferFactory(ByteBufAllocator.DEFAULT);
DataBuffer buffer = nettyDataBufferFactory.allocateBuffer(bytes.length);
buffer.write(bytes);
return buffer;
}
該方案的缺陷：request body獲取不完整（因為異步原因），只能獲取1024B的數據。並且請求體超過1024B，會出現響應超慢（因為我是開啟了熔斷）。
三、過濾器加路線定位器
翻查源碼發現ReadBodyPredicateFactory里面緩存了request body的信息，於是在自定義router中配置了ReadBodyPredicateFactory，然后在filter中通過cachedRequestBodyObject緩存字段獲取request body信息。
/**
* @description: 獲取POST請求的請求體
* ReadBodyPredicateFactory 發現里面緩存了request body的信息，
* 於是在自定義router中配置了ReadBodyPredicateFactory
* @modified:
*/
@EnableAutoConfiguration
@Configuration
public class RouteLocatorRequestBoby{
//自定義過濾器
@Resource
private ReqTraceFilter reqTraceFilter;
@Resource
private RibbonLoadBalancerClient ribbonLoadBalancerClient;
private static final String SERVICE = "/leap/**";
private static final String HTTP_PREFIX = "http://";
private static final String COLON = ":";
@Bean
public RouteLocator myRoutes(RouteLocatorBuilder builder) {
//通過負載均衡獲取服務實例
ServiceInstance instance = ribbonLoadBalancerClient.choose("PLATFORM-SERVICE");
//拼接路徑
StringBuilder forwardAddress = new StringBuilder(HTTP_PREFIX);
forwardAddress.append(instance.getHost())
.append(COLON)
.append(instance.getPort());
return builder.routes()
//攔截請求類型為POST Content-Type application/json application/json;charset=UTF-8
.route(r -> r
.header(HttpHeaders.CONTENT_TYPE,
MediaType.APPLICATION_JSON_VALUE + MediaType.APPLICATION_JSON_UTF8_VALUE)
.and()
.method(HttpMethod.POST)
.and()
//獲取緩存中的請求體
.readBody(Object.class, readBody -> {
return true;
})
.and()
.path(SERVICE)
//把請求體傳遞給攔截器reqTraceFilter
.filters(f -> {
f.filter(reqTraceFilter);
return f;
})
.uri(forwardAddress.toString())).build();
}
/**
* @description: 過濾器，用於獲取請求體，和處理請求體業務，列如記錄日志
* @modified:
*/
@Component
public class ReqTraceFilter implements GlobalFilter, GatewayFilter,Ordered {
private static final String CONTENT_TYPE = "Content-Type";
private static final String CONTENT_TYPE_JSON = "application/json";
//獲取請求路由詳細信息Route route = exchange.getAttribute(GATEWAY_ROUTE_BEAN)
private static final String GATEWAY_ROUTE_BEAN = "org.springframework.cloud.gateway.support.ServerWebExchangeUtils.gatewayRoute";
private static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject";
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
ServerHttpRequest request = exchange.getRequest();
//判斷過濾器是否執行
String requestUrl = RequestUtils.getCurrentRequest(request);
if (!RequestUtils.isFilter(requestUrl)) {
String bodyStr = "";
String contentType = request.getHeaders().getFirst(CONTENT_TYPE);
String method = request.getMethodValue();
//判斷是否為POST請求
if (null != contentType && HttpMethod.POST.name().equalsIgnoreCase(method) && contentType.contains(CONTENT_TYPE_JSON)) {
Object cachedBody = exchange.getAttribute(CACHE_REQUEST_BODY_OBJECT_KEY);
if(null != cachedBody){
bodyStr = cachedBody.toString();
}
}
if (HttpMethod.GET.name().equalsIgnoreCase(method)) {
bodyStr = request.getQueryParams().toString();
}
log.info("請求體內容：{}",bodyStr);
}
return chain.filter(exchange);
}
@Override
public int getOrder() {
return 5;
}
}
該方案優點：這種解決，一不會帶來重復讀取問題，二不會帶來requestbody取不全問題。三在低版本的Spring Cloud Finchley.SR2也可以運行。
缺點：不支持 multipart/form-data（異常415），這個致命。
四、通過 org.springframework.cloud.gateway.filter.factory.rewrite 包下有個 ModifyRequestBodyGatewayFilterFactory ，顧名思義，這就是修改 Request Body 的過濾器工廠類。
@Component
@Slf4j
public class ReqTraceFilter implements GlobalFilter, GatewayFilter, Ordered {
@Resource
private IPlatformFeignClient platformFeignClient;
/**
* httpheader，traceId的key名稱
*/
private static final String REQUESTID = "traceId";
private static final String CONTENT_TYPE = "Content-Type";
private static final String CONTENT_TYPE_JSON = "application/json";
private static final String GATEWAY_ROUTE_BEAN = "org.springframework.cloud.gateway.support.ServerWebExchangeUtils.gatewayRoute";
@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
ServerHttpRequest request = exchange.getRequest();
//判斷過濾器是否執行
String requestUrl = RequestUtils.getCurrentRequest(request);
if (!RequestUtils.isFilter(requestUrl)) {
String bodyStr = "";
String contentType = request.getHeaders().getFirst(CONTENT_TYPE);
String method = request.getMethodValue();
//判斷是否為POST請求
if (null != contentType && HttpMethod.POST.name().equalsIgnoreCase(method) && contentType.contains(CONTENT_TYPE_JSON)) {
ServerRequest serverRequest = new DefaultServerRequest(exchange);
List<String> list = new ArrayList<>();
// 讀取請求體
Mono<String> modifiedBody = serverRequest.bodyToMono(String.class)
.flatMap(body -> {
//記錄請求體日志
final String nId = saveRequestOperLog(exchange, body);
//記錄日志id
list.add(nId);
return Mono.just(body);
});
BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, String.class);
HttpHeaders headers = new HttpHeaders();
headers.putAll(exchange.getRequest().getHeaders());
headers.remove(HttpHeaders.CONTENT_LENGTH);
CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange, headers);
return bodyInserter.insert(outputMessage, new BodyInserterContext())
.then(Mono.defer(() -> {
ServerHttpRequestDecorator decorator = new ServerHttpRequestDecorator(
exchange.getRequest()) {
@Override
public HttpHeaders getHeaders() {
long contentLength = headers.getContentLength();
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.putAll(super.getHeaders());
httpHeaders.put(REQUESTID,list);
if (contentLength > 0) {
httpHeaders.setContentLength(contentLength);
} else {
httpHeaders.set(HttpHeaders.TRANSFER_ENCODING, "chunked");
}
return httpHeaders;
}
@Override
public Flux<DataBuffer> getBody() {
return outputMessage.getBody();
}
};
return chain.filter(exchange.mutate().request(decorator).build());
}));
}
if (HttpMethod.GET.name().equalsIgnoreCase(method)) {
bodyStr = request.getQueryParams().toString();
String nId = saveRequestOperLog(exchange, bodyStr);
ServerHttpRequest userInfo = exchange.getRequest().mutate()
.header(REQUESTID, nId).build();
return chain.filter(exchange.mutate().request(userInfo).build());
}
}
return chain.filter(exchange);
}
/**
* 保存請求日志
*
* @param exchange
* @param requestParameters
* @return
*/
private String saveRequestOperLog(ServerWebExchange exchange, String requestParameters) {
log.debug("接口請求參數：{}", requestParameters);
ServerHttpRequest request = exchange.getRequest();
String ip = Objects.requireNonNull(request.getRemoteAddress()).getAddress().getHostAddress();
SaveOperLogVO vo = new SaveOperLogVO();
vo.setIp(ip);
vo.setReqUrl(RequestUtils.getCurrentRequest(request));
vo.setReqMethod(request.getMethodValue());
vo.setRequestParameters(requestParameters);
Route route = exchange.getAttribute(GATEWAY_ROUTE_BEAN);
//是否配置路由
if (route != null) {
vo.setSubsystem(route.getId());
}
ResEntity<String> res = platformFeignClient.saveOperLog(vo);
log.debug("當前請求ID返回的數據：{}", res);
return res.getData();
}
@Override
public int getOrder() {
return 5;
}
}
該方案：完美解決以上所有問題