Spring cloud gateway-响应体日志记录导致直接内存错误

Question

下面是我使用Spring Cloud gateway构建API网关的堆栈-

SCG - Hoxton.RELEASE
Java - 1.8
Spring Boot - 2.2.1.RELEASE

我从互联网上得到了下面的“GlobalFilter”代码来记录请求/响应正文。这很好，但我会遇到这样的错误

"failed to allocate 16777216 byte(s) of direct memory (used: 1023410183, max: 1038876672)"

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.List;

import org.reactivestreams.Publisher;
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.DataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.server.ServerWebExchange;

import com.xx.GatewayJsonLogger;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Component
@Slf4j
public class RequestResponseLogFilter implements GlobalFilter, Ordered {

    private static final String MAGIC_HEADER = "x-debug";
    private static final String START_TIME = "startTime";

    private static final String HTTP_SCHEME = "http";

    private static final String HTTPS_SCHEME = "https";

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        List<String> debugHeader = exchange.getRequest().getHeaders().get(MAGIC_HEADER);
        if (!log.isDebugEnabled() && debugHeader == null) {
            // DO NOTHING
            return chain.filter(exchange);
        }
        ServerHttpRequest request = exchange.getRequest();
        URI requestURI = request.getURI();
        String scheme = requestURI.getScheme();
        if (debugHeader != null) {
            String debugHeaderContent = debugHeader.get(0);
            if (!debugHeaderContent.equalsIgnoreCase("true") && !requestURI.getPath().toLowerCase().contains(debugHeaderContent.toLowerCase())) {
                return chain.filter(exchange);
            }
        }
        if ((!HTTP_SCHEME.equalsIgnoreCase(scheme) && !HTTPS_SCHEME.equals(scheme))) {
            return chain.filter(exchange);
        }
        long startTime = System.currentTimeMillis();
        exchange.getAttributes().put(START_TIME, startTime);
        logRequest(request, exchange.getAttribute("cachedRequestBodyObject"));
        return chain.filter(exchange.mutate().response(logResponse(exchange)).build());
    }

    @Override
    public int getOrder() {
        return Integer.MIN_VALUE;
    }

    private void logRequest(ServerHttpRequest request, String body) {
        URI requestURI = request.getURI();
        String scheme = requestURI.getScheme();
        HttpHeaders headers = request.getHeaders();
        log.info("Request Scheme:{},Path:{}", scheme, requestURI.getPath());
        log.info("Request Method:{},IP:{},Host:{}", request.getMethod(), request.getRemoteAddress(), requestURI.getHost());
        headers.forEach((key, value) -> log.debug("Request Headers:Key->{},Value->{}", key, value));
        MultiValueMap<String, String> queryParams = request.getQueryParams();
        if (!queryParams.isEmpty()) {
            queryParams.forEach((key, value) -> log.info("Request Query Param :Key->({}),Value->({})", key, value));
        }
        MediaType contentType = headers.getContentType();
        long length = headers.getContentLength();
        log.info("Request ContentType:{},Content Length:{}", contentType, length);
        if (body != null) {
            GatewayJsonLogger.log(null, requestURI.getPath(), body.replaceAll("\"password\"\\s*:\\s*\".*\"", "\"password\":\"*******\""), "REQUEST");
        }
    }

    private ServerHttpResponseDecorator logResponse(ServerWebExchange exchange) {
        ServerHttpResponse origResponse = exchange.getResponse();
        Long startTime = exchange.getAttribute(START_TIME);
        log.info("Response HttpStatus:{}", origResponse.getStatusCode());
        HttpHeaders headers = origResponse.getHeaders();
        headers.forEach((key, value) -> log.debug("[RequestLogFilter]Headers:Key->{},Value->{}", key, value));
        MediaType contentType = headers.getContentType();
        long length = headers.getContentLength();
        log.info("Response ContentType:{},Content Length:{}", contentType, length);
        Long executeTime = (System.currentTimeMillis() - startTime);
        log.info("Response Original Path:{},Cost:{} ms", exchange.getRequest().getURI().getPath(), executeTime);
        DataBufferFactory bufferFactory = origResponse.bufferFactory();
        return new ServerHttpResponseDecorator(origResponse) {
            @Override
            public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                if (body instanceof Flux) {
                    Flux<? extends DataBuffer> fluxBody = (Flux<? extends DataBuffer>) body;
                    return super.writeWith(fluxBody.map(dataBuffer -> {
                        byte[] content = new byte[dataBuffer.readableByteCount()];
                        dataBuffer.read(content);
                        String bodyContent = new String(content, StandardCharsets.UTF_8);
                        GatewayJsonLogger.log(null, exchange.getRequest().getURI().getPath(), bodyContent, "RESPONSE");
                        return bufferFactory.wrap(content);
                    }));
                }
                return super.writeWith(body);
            }
        };

    }
}

1. Enable DEBUG logging for reactor.netty
2. Add below property to application.yml
   spring.cloud.gateway.httpserver.wiretap = true

它的工作原理是，我可以在日志文件中看到请求和响应头+正文，但它不是一种可读的格式（我认为是某种十六进制格式）。

你好，雅各布

Answer 1

似乎应该使用dataBuffer.release()发布您使用的dataBuffer，因为您使用bufferFactory.wrap(content)提供了一个新的dataBuffer，所以原始dataBuffer永远不会发布，并导致内存泄漏。