Spring Reactive微服务Netty Server连接ID日志采集异常排查

一、两种集成方式的问题分析 #

1. 自定义ServerBootstrap方式的问题 #

• 编解码器重复配置：HttpServerCodec 已经封装了 HttpRequestDecoder 和 HttpResponseEncoder，重复添加这两个处理器会导致Netty Pipeline的消息处理逻辑混乱，触发「Discarded inbound message. Please check your pipeline configuration.」错误。
• 缺失Spring Reactive核心处理器：完全自定义ServerBootstrap后，没有添加Spring Reactive处理HTTP请求的核心处理器（比如ReactorHttpHandlerAdapter），导致请求无法被Spring控制器处理。

2. WebServerFactoryCustomizer方式的问题 #

• Handler添加位置错误：addHandlerLast 将自定义Handler放在Pipeline末尾，而Spring Reactive的Netty在连接初始化后会重新编排Pipeline结构，末尾的自定义Handler会被后续操作移除，导致日志无法输出。
• AttributeKey使用不规范：每次调用AttributeKey.valueOf("connectionId")会重复创建实例（虽然Netty内部有缓存），可能导致属性获取失败，建议定义为静态常量。

二、优化后的解决方案 #

1. 优化自定义ConnectionLoggingHandler #

使用SLF4J日志框架代替System.out，并将AttributeKey定义为静态常量：

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class ConnectionLoggingHandler extends ChannelInboundHandlerAdapter {
    private static final Logger log = LoggerFactory.getLogger(ConnectionLoggingHandler.class);
    private static final AtomicInteger connectionIdCounter = new AtomicInteger(0);
    private static final AttributeKey<Integer> CONNECTION_ID = AttributeKey.valueOf("connectionId");

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        int connectionId = connectionIdCounter.incrementAndGet();
        log.info("Connection established : ID = {}", connectionId);
        ctx.channel().attr(CONNECTION_ID).set(connectionId);
        super.channelActive(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Integer connectionId = ctx.channel().attr(CONNECTION_ID).get();
        if (connectionId != null) {
            log.info("Connection closed : ID = {}", connectionId);
        }
        super.channelInactive(ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Integer connectionId = ctx.channel().attr(CONNECTION_ID).get();
        if (connectionId != null) {
            log.error("Exception in Connection: ID = {}", connectionId, cause);
        } else {
            log.error("Exception in unknown connection", cause);
        }
        ctx.close();
    }
}

2. 正确的WebServerFactoryCustomizer集成方式 #

使用addHandlerFirst将自定义Handler添加到Pipeline头部，避免被Spring后续操作移除：

import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory;
import org.springframework.boot.web.server.WebServerFactoryCustomizer;
import org.springframework.stereotype.Component;

@Component
public class NettyConnectionCustomizer implements WebServerFactoryCustomizer<NettyReactiveWebServerFactory> {

    @Override
    public void customize(NettyReactiveWebServerFactory factory) {
        factory.addServerCustomizers(server -> 
            server.tcpConfiguration(tcp -> 
                tcp.doOnConnection(connection -> 
                    connection.addHandlerFirst(new ConnectionLoggingHandler())
                )
            )
        );
    }
}

3. 自定义ServerBootstrap的兼容方案（不推荐） #

如果必须自行启动ServerBootstrap，需要手动集成Spring的HttpHandler，示例如下：

import org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.netty.http.server.HttpServer;

public class CustomNettyServer {
    private final RouterFunction<ServerResponse> routerFunction;

    public CustomNettyServer(RouterFunction<ServerResponse> routerFunction) {
        this.routerFunction = routerFunction;
    }

    public void startServer() throws InterruptedException {
        ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(routerFunction);
        HttpServer.create()
                .port(8080)
                .doOnConnection(conn -> conn.addHandlerFirst(new ConnectionLoggingHandler()))
                .handle(adapter)
                .bindNow()
                .onDispose()
                .block();
    }
}

三、验证方式 #

启动服务后发起请求，可在日志中看到：

• 连接建立时输出Connection established : ID = X
• 连接关闭时输出Connection closed : ID = X
• 发生异常时输出包含连接ID的错误日志

内容的提问来源于stack exchange，提问作者Satyam Pareek