本文主要研究一下reactor-netty的AccessLog
对于使用webflux的应用,没有这么对应的配置,但是可以通过-Dreactor.netty.http.server.accessLogEnabled=true来开启
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"; //...... }
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)); } } //...... }
reactor.netty.http.server.accessLogEnabled
)的属性,读取不到默认为false;HttpServerBind有个Http1Initializer类,它的accept方法会判断ACCESS_LOG是否为true,如果为true则会往Channel的pipeline添加名为accessLogHandler( NettyPipeline.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); } }
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()); } } }
{} - {} [{}] "{} {} {}" {} {} {} {} ms
),分别是address, user, zonedDateTime, method, uri, protocol, status, contentLength, port, duration reactor.netty.http.server.accessLogEnabled
)的属性,读取不到默认为false;HttpServerBind有个Http1Initializer类,它的accept方法会判断ACCESS_LOG是否为true,如果为true则会往Channel的pipeline添加名为accessLogHandler( NettyPipeline.AccessLogHandler
)的AccessLogHandler {} - {} [{}] "{} {} {}" {} {} {} {} ms
),分别是address, user, zonedDateTime, method, uri, protocol, status, contentLength, port, duration