gateway、webflux、reactor-netty請求日志輸出的方式是什么

本篇內容介紹了“gateway、webflux、reactor-netty請求日志輸出的方式是什么”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

gateway、webflux、reactor-netty請求日志輸出

場景

在使用spring cloud gateway時想要輸出請求日志，考慮到兩種實現方案

方案一

官網中使用Reactor Netty Access Logs方案，配置“-Dreactor.netty.http.server.accessLogEnabled=true”開啟日志記錄。

輸出如下：

reactor.netty.http.server.AccessLog      :
10.2.20.177 - - [02/Dec/2020:16:41:57 +0800] "GET /fapi/gw/hi/login HTTP/1.1" 200 319 8080 626 ms

• 優點:簡單方便

• 缺點:格式固定，信息量少

方案二

創建一個logfilter，在logfilter中解析request，并輸出請求信息

• 優點:可以自定義日志格式和內容,可以獲取body信息

• 缺點:返回信息需要再寫一個filter,沒有匹配到路由時無法進入到logfilter中

思路

對方案一進行改造，使其滿足需求。對reactor-netty源碼分析，主要涉及

• AccessLog：日志工具，日志結構體

• AccessLogHandler：http1.1協議日志控制，我們主要使用這個。

• AccessLogHandler2：http2協議日志控制

代碼如下：

package reactor.netty.http.server; 
import reactor.util.Logger;
import reactor.util.Loggers;
 
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Locale;
import java.util.Objects;
 
final class AccessLog {
    static final Logger log = Loggers.getLogger("reactor.netty.http.server.AccessLog");
    static final DateTimeFormatter DATE_TIME_FORMATTER =
            DateTimeFormatter.ofPattern("dd/MMM/yyyy:HH:mm:ss Z", Locale.US);
    static final String COMMON_LOG_FORMAT =
            "{} - {} [{}] \"{} {} {}\" {} {} {} {} ms";
    static final String MISSING = "-"; 
    final String zonedDateTime;
 
    String address;
    CharSequence method;
    CharSequence uri;
    String protocol;
    String user = MISSING;
    CharSequence status;
    long contentLength;
    boolean chunked;
    long startTime = System.currentTimeMillis();
    int port;
 
    AccessLog() {
        this.zonedDateTime = ZonedDateTime.now().format(DATE_TIME_FORMATTER);
    }
 
    AccessLog address(String address) {
        this.address = Objects.requireNonNull(address, "address");
        return this;
    }
 
    AccessLog port(int port) {
        this.port = port;
        return this;
    }
 
    AccessLog method(CharSequence method) {
        this.method = Objects.requireNonNull(method, "method");
        return this;
    }
 
    AccessLog uri(CharSequence uri) {
        this.uri = Objects.requireNonNull(uri, "uri");
        return this;
    }
 
    AccessLog protocol(String protocol) {
        this.protocol = Objects.requireNonNull(protocol, "protocol");
        return this;
    }
 
    AccessLog status(CharSequence status) {
        this.status = Objects.requireNonNull(status, "status");
        return this;
    }
 
    AccessLog contentLength(long contentLength) {
        this.contentLength = contentLength;
        return this;
    }
 
    AccessLog increaseContentLength(long contentLength) {
        if (chunked) {
            this.contentLength += contentLength;
        }
        return this;
    }
 
    AccessLog chunked(boolean chunked) {
        this.chunked = chunked;
        return this;
    }
 
    long duration() {
        return System.currentTimeMillis() - startTime;
    }
 
    void log() {
        if (log.isInfoEnabled()) {
            log.info(COMMON_LOG_FORMAT, address, user, zonedDateTime,
                    method, uri, protocol, status, (contentLength > -1 ? contentLength : MISSING), port, duration());
        }
    }
}

• AccessLogHandler：日志控制

package reactor.netty.http.server; 
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufHolder;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.LastHttpContent;
 
/**
 * @author Violeta Georgieva
 */
final class AccessLogHandler extends ChannelDuplexHandler {
 
    AccessLog accessLog = new AccessLog();
 
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof HttpRequest) {
            final HttpRequest request = (HttpRequest) msg;
            final SocketChannel channel = (SocketChannel) ctx.channel();
 
            accessLog = new AccessLog()
                    .address(channel.remoteAddress().getHostString())
                    .port(channel.localAddress().getPort())
                    .method(request.method().name())
                    .uri(request.uri())
                    .protocol(request.protocolVersion().text());
        }
        ctx.fireChannelRead(msg);
    }
 
    @Override
    @SuppressWarnings("FutureReturnValueIgnored")
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        if (msg instanceof HttpResponse) {
            final HttpResponse response = (HttpResponse) msg;
            final HttpResponseStatus status = response.status();
 
            if (status.equals(HttpResponseStatus.CONTINUE)) {
                //"FutureReturnValueIgnored" this is deliberate
                ctx.write(msg, promise);
                return;
            }
 
            final boolean chunked = HttpUtil.isTransferEncodingChunked(response);
            accessLog.status(status.codeAsText())
                     .chunked(chunked);
            if (!chunked) {
                accessLog.contentLength(HttpUtil.getContentLength(response, -1));
            }
        }
        if (msg instanceof LastHttpContent) {
            accessLog.increaseContentLength(((LastHttpContent) msg).content().readableBytes());
            ctx.write(msg, promise.unvoid())
               .addListener(future -> {
                   if (future.isSuccess()) {
                       accessLog.log();
                   }
               });
            return;
        }
        if (msg instanceof ByteBuf) {
            accessLog.increaseContentLength(((ByteBuf) msg).readableBytes());
        }
        if (msg instanceof ByteBufHolder) {
            accessLog.increaseContentLength(((ByteBufHolder) msg).content().readableBytes());
        }
        //"FutureReturnValueIgnored" this is deliberate
        ctx.write(msg, promise);
    }
}

執行順序

AccessLogHandler.channelRead > GlobalFilter.filter > AbstractLoadBalance.choose >response.writeWith >AccessLogHandler.write

解決方案

對AccessLog和AccessLogHandler進行重寫，輸出自己想要的內容和樣式。

AccessLogHandler中重寫了ChannelDuplexHandler中的channelRead和write方法，還可以對ChannelInboundHandler和ChannelOutboundHandler中的方法進行重寫，覆蓋請求的整個生命周期。

spring-webflux、gateway、springboot-start-web問題

Spring-webflux

當兩者一起時配置的并不是webflux web application, 仍然時一個spring mvc web application。

官方文檔中有這么一段注解：

很多開發者添加spring-boot-start-webflux到他們的spring mvc web applicaiton去是為了使用reactive WebClient. 如果希望更改webApplication 類型需要顯示的設置，如SpringApplication.setWebApplicationType(WebApplicationType.REACTIVE).

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

結論一：

當兩者一起時配置的并不是webflux web application, 仍然時一個spring mvc web application。但是啟動不會報錯，可以正常使用，但是webflux功能失效

Spring-gateway

因為gateway和zuul不一樣，gateway用的是長連接，netty-webflux，zuul1.0用的就是同步webmvc。

所以你的非gateway子項目啟動用的是webmvc，你的gateway啟動用的是webflux. spring-boot-start-web和spring-boot-start-webflux相見分外眼紅。

不能配置在同一pom.xml，或者不能在同一項目中出現，不然就會啟動報錯

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-gateway</artifactId>
    </dependency>

結論二：

當spring-cloud-gateway和spring-boot-starer-web兩者一起時配置的時候, 啟動直接報錯，依賴包沖突不兼容

“gateway、webflux、reactor-netty請求日志輸出的方式是什么”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！