Spring Cloud Gateway(读取、修改 Request Body)的操作
Spring Cloud Gateway(以下简称 SCG)做为网关服务,是其他各服务对外中转站,通过 SCG 进行请求转发。
在请求到达真正的微服务之前,我们可以在这里做一些预处理,比如:来源合法性检测,权限校验,反爬虫之类…
因为业务需要,我们的服务的请求参数都是经过加密的。
之前是在各个微服务的拦截器里对来解密验证的,现在既然有了网关,自然而然想把这一步骤放到网关层来统一解决。
如果是使用普通的 Web 编程中(比如用 Zuul),这本就是一个 pre filter 的事儿,把之前 Interceptor 中代码搬过来稍微改改就 OK 了。
不过因为使用的 SCG,它基于 Spring 5 的 WebFlux,即 Reactor 编程,要读取 Request Body 中的请求参数就没那么容易了。
本篇内容涉及 WebFlux 的响应式编程及 SCG 自定义全局过滤器,如果对这两者不了解的话,可以先看看相关的内容。
两个大坑
我们先建一个 Filter 来看看
public class ValidateFilter implements GlobalFilter, Ordered { @Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { ServerHttpRequest request = exchange.getRequest(); HttpHeaders headers = request.getHeaders(); MultiValueMap<String, HttpCookie> cookies = request.getCookies(); MultiValueMap<String, String> queryParams = request.getQueryParams(); Flux<DataBuffer> body = request.getBody(); return null; } @Override public int getOrder() { return 0; } }
从上边的返回值可以看出,如果是取 Header、Cookie、Query Params 都易如反掌,如果你需要校验的数据在这三者之中的话,就没必要往下看了。
说回 Body,这里是一个Flux<DataBuffer>,即一个包含 0-N 个DataBuffer类型元素的异步序列。
首先不考虑 Request Body 只能读取一次问题(这个问题可以用缓存解决),我们先来把这个 Flux 转化成我们可以处理的字符串,第一反应想到的有两个办法:
block() 异步变同步
subscribe() 订阅并触发序列
BUT,理想很丰满,现实却很骨感——这两个办法都有问题:
WebFlux 中不能使用阻塞的操作
java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-server-epoll-7
subscribe() 只会接收到第一个发出的元素,所以会导致获取不全的问题(太长的 Body 会被截断)。这个问题网上有人用 AtomicReference<String> 来包装获取到字符串,有人用 StringBuilder/StringBuffer
以上两个问题在网上找了半天,也没找到一个靠谱的解决办法,都是人云亦云。特别是第二个问题的所谓的“解决办法”,大家无非就在是不遗余力的在展示 DataBuffer 转 String 的 N 种写法,而没有从根本上解决被截断的问题。
正确姿势
2019.08.26 更新:
评论里有网友提醒到 Spring Cloud Gateway 2.1.2 下 DefaultServerRequest、CachedBodyOutputMessage 类的访问权限已经改了。这一块我看了一下,源码确实改动了一些,不过 DefaultServerRequest 这个类已经不需要了,而 CachedBodyOutputMessage 类我们可以模(chao)仿(xi)它的实现。
其实这里的实现不管再怎么变,我们只要死盯着 ModifyRequestBodyGatewayFilterFactory 就行了。即使以后这里边的相关类的访问权限都改成 Default 了,我们也不用一个个去抄一遍,只要在org.springframework.cloud.gateway.filter.factory.rewrite 这个 package 下写我们自己的类就好了。
———– 分割线 ———-
最终找到解决方案还是通过研读 SCG 的源码。
本文使用的版本:
Spring Cloud: Greenwich.RC2
Spring Boot: 2.1.1.RELEASE
在 org.springframework.cloud.gateway.filter.factory.rewrite 包下有个 ModifyRequestBodyGatewayFilterFactory,顾名思义,这就是修改 Request Body 的过滤器工厂类。
但是这个类我们无法直接使用,因为要用的话这个 FilterFactory 只能用 Fluent API 的方式配置,而无法在配置文件中使用,类似于这样
.route("rewrite_request_upper", r -> r.host("*.rewriterequestupper.org") .filters(f -> f.prefixPath("/httpbin") .addResponseHeader("X-TestHeader", "rewrite_request_upper") .modifyRequestBody(String.class, String.class, (exchange, s) -> { return Mono.just(s.toUpperCase()+s.toUpperCase()); }) ).uri(uri) )
我更喜欢用配置文件来配置路由,所以这种方式并不是我的菜。
这时候我就需要自己弄一个 GlobalFilter 了。既然官方已经提供了“葫芦”,那么我们就画个“瓢”吧。
如果了解的 GatewayFilterFactory 和 GatewayFilter 的关系的话,不用我说你就知道该怎么办了。不知道也没关系,我们把 ModifyRequestBodyGatewayFilterFactory 中红框部分 copy 出来,粘贴到我们之前创建的 ValidateFilter#filter 中
我们稍作修改,即可实现读取并修改 Request Body 的功能了(核心部分见上图黄色箭头处)
/** * @author yibo */ public class ValidateFilter implements GlobalFilter, Ordered { @Override public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { ServerRequest serverRequest = new DefaultServerRequest(exchange); // mediaType MediaType mediaType = exchange.getRequest().getHeaders().getContentType(); // read & modify body Mono<String> modifiedBody = serverRequest.bodyToMono(String.class) .flatMap(body -> { if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) { // origin body map Map<String, Object> bodyMap = decodeBody(body); // TODO decrypt & auth // new body map Map<String, Object> newBodyMap = new HashMap<>(); return Mono.just(encodeBody(newBodyMap)); } return Mono.empty(); }); BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, String.class); HttpHeaders headers = new HttpHeaders(); headers.putAll(exchange.getRequest().getHeaders()); // the new content type will be computed by bodyInserter // and then set in the request decorator 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()); 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()); })); } @Override public int getOrder() { return 0; } private Map<String, Object> decodeBody(String body) { return Arrays.stream(body.split("&")) .map(s -> s.split("=")) .collect(Collectors.toMap(arr -> arr[0], arr -> arr[1])); } private String encodeBody(Map<String, Object> map) { return map.entrySet().stream().map(e -> e.getKey() + "=" + e.getValue()).collect(Collectors.joining("&")); } }
至于拿到 Body 后具体要做什么,也就上边代码中的TODO部分,就由你自己来发挥吧~ 别玩坏就好
建议大家可以多关注关注 SCG 的源码,说不定什么时候就会多出一些有用的 Filter 或 FilterFactory。
另外,目前 ModifyRequestBodyGatewayFilterFactory 上的 Javadoc 有这么一句话:
This filter is BETA and may be subject to change in a future release.
所以大家要保持关注呀~
以上这篇Spring Cloud Gateway(读取、修改 Request Body)的操作就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持我们。