序

本文主要研究一下reactor-netty的AccessLog

开启access log

对于使用tomcat的spring boot应用，可以server.tomcat.accesslog.enabled=true来开启对于使用jetty的spring boot应用，可以server.jetty.accesslog.enabled=true来开启对于使用undertow的spring boot应用，可以server.undertow.accesslog.enabled=true来开启

对于使用webflux的应用，没有这么对应的配置，但是可以通过-Dreactor.netty.http.server.accessLogEnabled=true来开启

ReactorNetty

reactor-netty-0.8.5.RELEASE-sources.jar!/reactor/netty/ReactorNetty.java

/** * Internal helpers for reactor-netty contracts * * @author Stephane Maldini */public final class ReactorNetty { //......​ // System properties names​ /** * Default worker thread count, fallback to available processor * (but with a minimum value of 4) */ public static final String IO_WORKER_COUNT = "reactor.netty.ioWorkerCount"; /** * Default selector thread count, fallback to -1 (no selector thread) */ public static final String IO_SELECT_COUNT = "reactor.netty.ioSelectCount"; /** * Default worker thread count for UDP, fallback to available processor * (but with a minimum value of 4) */ public static final String UDP_IO_THREAD_COUNT = "reactor.netty.udp.ioThreadCount";​​ /** * Default value whether the native transport (epoll, kqueue) will be preferred, * fallback it will be preferred when available */ public static final String NATIVE = "reactor.netty.native";​​ /** * Default max connections, if -1 will never wait to acquire before opening a new * connection in an unbounded fashion. Fallback to * available number of processors (but with a minimum value of 16) */ public static final String POOL_MAX_CONNECTIONS = "reactor.netty.pool.maxConnections"; /** * Default acquisition timeout (milliseconds) before error. If -1 will never wait to * acquire before opening a new * connection in an unbounded fashion. Fallback 45 seconds */ public static final String POOL_ACQUIRE_TIMEOUT = "reactor.netty.pool.acquireTimeout";​​ /** * Default SSL handshake timeout (milliseconds), fallback to 10 seconds */ public static final String SSL_HANDSHAKE_TIMEOUT = "reactor.netty.tcp.sslHandshakeTimeout"; /** * Default value whether the SSL debugging on the client side will be enabled/disabled, * fallback to SSL debugging disabled */ public static final String SSL_CLIENT_DEBUG = "reactor.netty.tcp.ssl.client.debug"; /** * Default value whether the SSL debugging on the server side will be enabled/disabled, * fallback to SSL debugging disabled */ public static final String SSL_SERVER_DEBUG = "reactor.netty.tcp.ssl.server.debug";​​ /** * Specifies whether the Http Server access log will be enabled. * By default it is disabled. */ public static final String ACCESS_LOG_ENABLED = "reactor.netty.http.server.accessLogEnabled";​ //......}

ReactorNetty定义了ACCESS_LOG_ENABLED常量，其值为reactor.netty.http.server.accessLogEnabled

HttpServerBind

reactor-netty-0.8.5.RELEASE-sources.jar!/reactor/netty/http/server/HttpServerBind.java

final class HttpServerBind extends HttpServer implements Function<ServerBootstrap, ServerBootstrap> {​ static final HttpServerBind INSTANCE = new HttpServerBind();​ static final Function<DisposableServer, DisposableServer> CLEANUP_GLOBAL_RESOURCE = DisposableBind::new;​ static final boolean ACCESS_LOG = Boolean.parseBoolean(System.getProperty(ACCESS_LOG_ENABLED, "false"));​ //......​ static final class Http1Initializer implements BiConsumer<ConnectionObserver, Channel> {​ final int line; final int header; final int chunk; final boolean validate; final int buffer; final int minCompressionSize; final BiPredicate<HttpServerRequest, HttpServerResponse> compressPredicate; final boolean forwarded; final ServerCookieEncoder cookieEncoder; final ServerCookieDecoder cookieDecoder;​ Http1Initializer(int line, int header, int chunk, boolean validate, int buffer, int minCompressionSize, @Nullable BiPredicate<HttpServerRequest, HttpServerResponse> compressPredicate, boolean forwarded, ServerCookieEncoder encoder, ServerCookieDecoder decoder) { this.line = line; this.header = header; this.chunk = chunk; this.validate = validate; this.buffer = buffer; this.minCompressionSize = minCompressionSize; this.compressPredicate = compressPredicate; this.forwarded = forwarded; this.cookieEncoder = encoder; this.cookieDecoder = decoder; }​ @Override public void accept(ConnectionObserver listener, Channel channel) { ChannelPipeline p = channel.pipeline();​ p.addLast(NettyPipeline.HttpCodec, new HttpServerCodec(line, header, chunk, validate, buffer));​ if (ACCESS_LOG) { p.addLast(NettyPipeline.AccessLogHandler, new AccessLogHandler()); }​ boolean alwaysCompress = compressPredicate == null && minCompressionSize == 0;​ if (alwaysCompress) { p.addLast(NettyPipeline.CompressionHandler, new SimpleCompressionHandler()); }​ p.addLast(NettyPipeline.HttpTrafficHandler, new HttpTrafficHandler(listener, forwarded, compressPredicate, cookieEncoder, cookieDecoder)); } }​ //......} 

HttpServerBind有个ACCESS_LOG属性，它读取ReactorNetty的ACCESS_LOG_ENABLED(reactor.netty.http.server.accessLogEnabled)的属性，读取不到默认为false；HttpServerBind有个Http1Initializer类，它的accept方法会判断ACCESS_LOG是否为true，如果为true则会往Channel的pipeline添加名为accessLogHandler(NettyPipeline.AccessLogHandler)的AccessLogHandler

AccessLogHandler

reactor-netty-0.8.5.RELEASE-sources.jar!/reactor/netty/http/server/AccessLogHandler.java

final class AccessLogHandler extends ChannelDuplexHandler {​ AccessLog accessLog = new AccessLog();​ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 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()); } super.channelRead(ctx, msg); }​ @Override 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)) { 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) .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()); } ctx.write(msg, promise); }}

AccessLogHandler继承了ChannelDuplexHandler；在channelRead的时候创建了AccessLog对象，在write的时候更新AccessLog对象；当msg为LastHttpContent时，则添加了一个listener，在成功回调时执行accessLog.log()

AccessLog

reactor-netty-0.8.5.RELEASE-sources.jar!/reactor/netty/http/server/AccessLog.java

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

AccessLog的log方法直接通过logger输出日志，其日志格式为COMMON_LOG_FORMAT({} - {} [{}] "{} {} {}" {} {} {} {} ms)，分别是address, user, zonedDateTime, method, uri, protocol, status, contentLength, port, duration

小结

对于使用webflux的应用，可以通过-Dreactor.netty.http.server.accessLogEnabled=true来开启access logHttpServerBind有个ACCESS_LOG属性，它读取ReactorNetty的ACCESS_LOG_ENABLED(reactor.netty.http.server.accessLogEnabled)的属性，读取不到默认为false；HttpServerBind有个Http1Initializer类，它的accept方法会判断ACCESS_LOG是否为true，如果为true则会往Channel的pipeline添加名为accessLogHandler(NettyPipeline.AccessLogHandler)的AccessLogHandlerAccessLogHandler继承了ChannelDuplexHandler；在channelRead的时候创建了AccessLog对象，在write的时候更新AccessLog对象；当msg为LastHttpContent时，则添加了一个listener，在成功回调时执行accessLog.log()；AccessLog的log方法直接通过logger输出日志，其日志格式为COMMON_LOG_FORMAT({} - {} [{}] "{} {} {}" {} {} {} {} ms)，分别是address, user, zonedDateTime, method, uri, protocol, status, contentLength, port, duration

doc

Spring Boot Reactor Netty Configuration