SpringCloudGateway获取报文大小

Overview

SpringCloud Gateway使用过程中，希望获取报文大小。由于SpringCloud Gateway底层基于Netty实现，直接读取报文，会大幅影响网关性能。因此本文将通过其他方式获取报文大小。本文基于2.2.9 SpringCloud Gateway开发。

读取请求报文大小

实现自定义Filter，读取请求报文大小，具体可参考以下代码。

import lombok.extern.slf4j.Slf4j;import org.springframework.cloud.gateway.filter.GatewayFilterChain;import org.springframework.cloud.gateway.filter.GlobalFilter;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;@Slf4j@Componentpublic class ReadRequestBodyFilter implements GlobalFilter {    @Override    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {        // 分配空的DataBuffer        DataBuffer emptyBuffer = exchange.getResponse().bufferFactory().allocateBuffer(0);        // 使用 DataBufferUtils.join 将DataBuffer数据流聚合为一个Mono        // 聚合后的 DataBuffer 为一个完整报文的 DataBuffer。        // 如果请求报文为空，使用分配的空 DataBuffer。        return DataBufferUtils.join(exchange.getRequest().getBody().defaultIfEmpty(emptyBuffer))                .flatMap(dataBuffer -> {                    // 获取报文大小，不直接获取报文内容。                    int size = dataBuffer.readableByteCount();                    log.info("=====> request body size: {}", size);                    if (size == 0) {                        // 如果报文内容为空，需要主动释放创建的空的DataBuffer。                        DataBufferUtils.release(dataBuffer);                        return chain.filter(exchange);                    }                    // 复制一份DataBuffer，slice方法不会retain DataBuffer，只是复制的指针坐标。                    Flux cachedFlux = Flux.defer(() -> Flux.just(                            dataBuffer.slice(0, dataBuffer.readableByteCount())));                    // 构建新的 Request，重写 getBody 返回复制的报文                    ServerHttpRequest mutatedRequest = new ServerHttpRequestDecorator(                            exchange.getRequest()) {                        @Override                        public Flux getBody() {                            return cachedFlux;                        }                    };                    // 继续处理请求                    return chain.filter(exchange.mutate().request(mutatedRequest).build());                });    }}

读取响应报文大小

重写NettyWriteResponseFilter，实现响应报文大小获取。此方法有点不够优雅。

// 重写 filter 方法@Overridepublic Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {   // NOTICE: nothing in "pre" filter stage as CLIENT_RESPONSE_CONN_ATTR is not added   // until the NettyRoutingFilter is run   // @formatter:off   return chain.filter(exchange)         .doOnError(throwable -> cleanup(exchange))         .then(Mono.defer(() -> {            Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);            if (connection == null) {               return Mono.empty();            }            if (log.isTraceEnabled()) {               log.trace("NettyWriteResponseFilter start inbound: "                     + connection.channel().id().asShortText() + ", outbound: "                     + exchange.getLogPrefix());            }            ServerHttpResponse response = exchange.getResponse();            // TODO: needed?            final Flux body = connection                  .inbound()                  .receive()                  .retain()                  .map(byteBuf -> wrap(byteBuf, response));            MediaType contentType = null;            try {               contentType = response.getHeaders().getContentType();            } catch (Exception e) {               if (log.isTraceEnabled()) {                  log.trace("invalid media type", e);               }            }                        // 重写部分代码，如果为 stream 类型报文，则忽略报文读取            // stream media type, we can not join databuffer, just ignore body size.            if (isStreamingMediaType(contentType)) {               return response.writeAndFlushWith(body.map(Flux::just));            }            // 分配空的 DataBuffer            DataBuffer emptyBuffer = exchange.getResponse().bufferFactory().allocateBuffer(0);            // 聚合 DataBuffer 构建大的 DataBuffer 包含完整报文            return DataBufferUtils.join(body.defaultIfEmpty(emptyBuffer))                  .flatMap(dataBuffer -> {                     // 读取报文大小                     int size = dataBuffer.readableByteCount();                     log.info("=====> response body size: " + size);                     // 复制 DataBuffer                      Flux cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, size)));                     // 向 response 写入数据                     return response.writeWith(cachedFlux);                  });         })).doOnCancel(() -> cleanup(exchange));}

总结

通过上述方式，可以获取到完整的请求和响应报文大小。根据压测结果，此方法对性能无影响。