Spring 云网关 - 响应 body 记录导致直接内存不足错误

Question

下面是我用来构建 API 网关的堆栈，使用 Spring 云网关 -

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

我从互联网上得到下面的“GlobalFilter”代码来记录 request/response body。这工作正常，但我运行出现

这样的错误

"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);
            }
        };

    }
}

我认为此处用于记录响应的“DataBuffer”导致了内存问题；请求部分看起来不错，因为它来自缓存 object。您能帮忙建议一种更有效的记录响应的方法吗？我尝试删除此过滤器并执行以下步骤 -

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

我可以在日志文件中看到请求和响应 headers+body 但它不是可读格式（我认为是某种十六进制格式）。

此致，雅各布

Answer 1

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

Answer 2

试试这个：

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 -> {
                try{
                    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);
                 }
                 finally{
                    DataBufferUtils.release(dataBuffer);
                }
                }));
            }
            return super.writeWith(body);
        }
    };

Spring 云网关 - 响应 body 记录导致直接内存不足错误

Spring cloud gateway - Response body logging causing out of direct memory error

spring-cloud-gateway