Gateway读取Request Body
我们使用SpringCloud Gateway做微服务网关的时候，经常需要在过滤器Filter中读取到Post请求中的Body内容进行日志记录、签名验证、权限验证等操作。我们知道，Request的Body是只能读取一次的，如果直接通过在Filter中读取，而不封装回去回导致后面的服务无法读取数据。
SpringCloud Gateway内部提供了一个断言工厂类ReadBodyPredicateFactory，这个类实现了读取Request的Body内容并放入缓存，我们可以通过从缓存中获取body内容来实现我们的目的。
分析ReadBodyPredicateFactory
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
public AsyncPredicate<ServerWebExchange> applyAsync(ReadBodyPredicateFactory.Config config) { return (exchange) -> { Class inClass = config.getInClass(); Object cachedBody = exchange.getAttribute( "cachedRequestBodyObject" ); if (cachedBody != null ) { try { boolean test = config.predicate.test(cachedBody); exchange.getAttributes().put( "read_body_predicate_test_attribute" , test); return Mono.just(test); } catch (ClassCastException var7) { if (LOGGER.isDebugEnabled()) { LOGGER.debug( "Predicate test failed because class in predicate does not match the cached body object" , var7); } return Mono.just( false ); } } else { return DataBufferUtils.join(exchange.getRequest().getBody()).flatMap((dataBuffer) -> { DataBufferUtils.retain(dataBuffer); final Flux<DataBuffer> cachedFlux = Flux.defer(() -> { return Flux.just(dataBuffer.slice( 0 , dataBuffer.readableByteCount())); }); ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(exchange.getRequest()) { public Flux<DataBuffer> getBody() { return cachedFlux; } }; return ServerRequest.create(exchange.mutate().request(mutatedRequest).build(), messageReaders).bodyToMono(inClass).doOnNext((objectValue) -> { exchange.getAttributes().put( "cachedRequestBodyObject" , objectValue); exchange.getAttributes().put( "cachedRequestBody" , cachedFlux); }).map((objectValue) -> { return config.predicate.test(objectValue); }); }); } }; } |
通过查看ReadBodyPredicateFactory内部实现，我们可以看到，该工厂类将request body内容读取后存放在 exchange的cachedRequestBodyObject中。
那么我们可以通过代码：
|
1 |
exchange.getAttribute([cachedRequestBodyObject]); //将body内容取出来 |
知道如何取body内容后，我们只需将该工厂类注册到yml配置文件中的predicates，然后从Filter中获取即可。
配置ReadBodyPredicateFactory
查看ReadBodyPredicateFactory关于配置的代码：
|
1 2 3 4 5 |
public <T> ReadBodyPredicateFactory.Config setPredicate(Class<T> inClass, Predicate<T> predicate) { this .setInClass(inClass); this .predicate = predicate; return this ; } |
配置该工厂类需要两个参数：
inClass ：接收body内容的对象Class，我们用字符串接收，配置String即可。 Predicate ：Predicate的接口实现类，我们自定义一个Predicate的实现类即可。自定义Predicate实现，并注册Bean。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
/** * 用于readBody断言，可配置到yml * @return */ @Bean public Predicate bodyPredicate(){ return new Predicate() { @Override public boolean test(Object o) { return true ; } }; } |
两个参数都有了，直接在yml中配置：
|
1 2 3 4 5 6 |
predicates: - Path=/card/api/** - name: ReadBodyPredicateFactory #使用ReadBodyPredicateFactory断言，将body读入缓存 args: inClass: '#{T(String)}' predicate: '#{@bodyPredicate}' #注入实现predicate接口类 |
编写自定义GatewayFilterFactory
编写自己的过滤器工厂类，读取缓存的body内容，并支持在配置文件中配置。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 |
public class ReadBodyGatewayFilterFactory extends AbstractGatewayFilterFactory<ReadBodyGatewayFilterFactory.Config> { private Logger logger = LoggerFactory.getLogger(ReadBodyGatewayFilterFactory. class ); private static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject" ; public ReadBodyGatewayFilterFactory() { super (Config. class ); } @Override public GatewayFilter apply(Config config) { return ((exchange, chain) -> { //利用ReadBodyPredicateFactory断言，会将body读入exchange的cachedRequestBodyObject中 Object requestBody = exchange.getAttribute(CACHE_REQUEST_BODY_OBJECT_KEY); logger.info( "request body is:{}" , requestBody); return chain.filter(exchange); }); } @Override public List<String> shortcutFieldOrder() { return Arrays.asList( "withParams" ); //将参数放入 } public static class Config { private boolean withParams; //接收配置的参数值，可以随便写 public boolean isWithParams() { return withParams; } public void setWithParams( boolean withParams) { this .withParams = withParams; } } } |
将ReadBodyGatewayFilterFactory工程类在容器中注入。
|
1 2 3 4 5 6 7 8 |
/** * 注入ReadBody过滤器 * @return */ @Bean public ReadBodyGatewayFilterFactory readBodyGatewayFilterFactory() { return new ReadBodyGatewayFilterFactory(); } |
到此，我们的Filter类也可以在yml配置文件中直接配置使用了。
完整的yml配置
|
1 2 3 4 5 6 7 8 9 10 11 |
- id: body_route #读取post中的body路由 order: 5 uri: lb://API-CARD filters: - ReadBody=true #使用自定义的过滤器工厂类，读取request body内容 predicates: - Path=/card/api/** - name: ReadBodyPredicateFactory #使用ReadBodyPredicateFactory断言，将body读入缓存 args: inClass: '#{T(String)}' predicate: '#{@bodyPredicate}' #注入实现predicate接口类 |
OK，以上是通过ReadBodyPredicateFactory这个类读取到request body内容。
另外springcloud gateway内部还提供了ModifyRequestBodyGatewayFilterFactory类用于修改body内容，既然能修改，自然也能获取body，大家可自行去研究。
Gateway自定义filter获取body的数据为空
最近在使用SpringCloud Gateway进行网关的开发，我使用的版本是：SpringBoot的2.3.4.RELEASE+SpringCloud的Hoxton.SR8，在自定义过滤器时需要获取ServerHttpRequest中body的数据，发现一直无法获取到数据，经过各种百度、谷歌，再加上自己的实践，终于找到解决方案：
首先创建一个全局过滤器把body中的数据缓存起来
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
package com.cloudpath.gateway.portal.filter; import lombok.extern.slf4j.Slf4j; 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; /** * @author mazhen * @className CacheBodyGlobalFilter * @Description 把body中的数据缓存起来 * @date 2020/10/28 18:02 */ @Slf4j @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
所以，必须把CacheBodyGlobalFilter的优先级设到最高。
在自定义的过滤器中尝试获取body中的数据
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
package com.cloudpath.iam.gateway.customerfilter; import com.cloudpath.iam.gateway.utils.FilterRequestResponseUtil; import lombok.extern.slf4j.Slf4j; import org.springframework.cloud.gateway.filter.GatewayFilter; import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.http.server.reactive.ServerHttpRequest; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux; import java.util.Arrays; import java.util.List; /** * @author by mazhen * @Classname TestGatewayFilterFactory * @Description 自定义过滤器获取body中的数据 * @Date 2020/10/27 14:38 */ @Component @Slf4j public class TestGatewayFilterFactory extends AbstractGatewayFilterFactory<TestGatewayFilterFactory.Config> { @Override public List<String> shortcutFieldOrder() { return Arrays.asList( "enabled" ); } public TestGatewayFilterFactory() { super (Config. class ); log.info( "Loaded TestGatewayFilterFactory" ); } @Override public GatewayFilter apply(Config config) { return (exchange, chain) -> { if (!config.isEnabled()) { return chain.filter(exchange); } if ( null != exchange) { ServerHttpRequest httpRequest = exchange.getRequest(); try { Flux<DataBuffer> dataBufferFlux = httpRequest.getBody(); //获取body中的数据 String body = FilterRequestResponseUtil.resolveBodyFromRequest(dataBufferFlux); log.info( "body:{}" ,body); } catch (Exception e) { log.error( "异常:" ,e); return chain.filter(exchange); } } return chain.filter(exchange); }; } public static class Config { /** * 控制是否开启统计 */ private boolean enabled; public Config() { } public boolean isEnabled() { return enabled; } public void setEnabled( boolean enabled) { this .enabled = enabled; } } } |
解析body的工具类
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
package com.cloudpath.iam.gateway.utils; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferUtils; import reactor.core.publisher.Flux; import java.nio.CharBuffer; import java.nio.charset.StandardCharsets; import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Matcher; import java.util.regex.Pattern; /** * @author mazhen * @className FilterHeadersUtil * @Description 过滤器请求/响应工具类 * @date 2020/10/29 9:31 */ public final class FilterRequestResponseUtil { /** * spring cloud gateway 获取post请求的body体 * @param body * @return */ public static String resolveBodyFromRequest( Flux<DataBuffer> body){ 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(); } /** * 读取body内容 * @param body * @return */ public static String resolveBodyFromRequest2( Flux<DataBuffer> body){ 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测试数据pile( "\\s*|\t|\r|\n" ); Matcher m = p.matcher(str); return m.replaceAll( "" ); } return str; } } |
解析body的内容，网上普遍是上面的两种方式,亲测resolveBodyFromRequest方法解析body中的数据，没有1024字节的限制。
ps：我传的参数有1万多字节。。。。。。。
大家可以按需所选。
以上为个人经验，希望能给大家一个参考，也希望大家多多支持。
原文链接：https://blog.csdn.net/cjbfzxz/article/details/106653242
查看更多关于SpringCloud Gateway读取Request Body方式的详细内容...