Spring Boot内嵌Netty服务器：如何无响应关闭/重置HTTP/2流

我们的应用基于启用HTTP/2的Spring Boot内嵌Netty服务器，通过Rest API对外提供服务。存在一种场景：在Rest API调用时，我们希望忽略HTTP/2请求且不返回任何响应。

环境信息 #

• Reactor版本：reactor-netty-http : 1.0.16
• JVM版本：17.0.1
• Spring Boot版本：2.7.4

当前代码实现 #

GreetingController.java #

public class GreetingController {

    private final GreetingService greetingService;

    public GreetingController(GreetingService greetingService) {
        this.greetingService = greetingService;
    }

    @GetMapping("/{name}")
    private Mono<String> greet(@PathVariable String name) {
        return greetingService.greet(name);
    }
}

DiscardOutBoundHandler.java #

public class DiscardOutBoundHandler extends ChannelOutboundHandlerAdapter {

    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println("Outbound Handler name: " + ctx.channel().pipeline().names());
        promise.channel().close();
    }

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.print("exceptionCaught call");
        cause.printStackTrace();
        ctx.close();
    }
}

NettyWebServerFactoryPortCustomizer.java #

public class NettyWebServerFactoryPortCustomizer implements WebServerFactoryCustomizer<NettyReactiveWebServerFactory> {

    @Override
    public void customize(NettyReactiveWebServerFactory serverFactory) {
        Http2 h2 = new Http2();
        serverFactory.setHttp2(h2);
        serverFactory.addServerCustomizers(new PortCustomizer(8443));
        serverFactory.addServerCustomizers(
                httpServer -> httpServer.doOnChannelInit((connectionObserver, channel, remoteAddress) -> {
                    ChannelPipeline channelPipeline = channel.pipeline();
                    channelPipeline.addLast("encoder", new DiscardOutBoundHandler());
                }));

    }

    private static class PortCustomizer implements NettyServerCustomizer {

        private final int port;

        private PortCustomizer(int port) {
            this.port = port;
        }

        @Override
        public HttpServer apply(HttpServer httpServer) {
            return httpServer.port(port);
        }
    }
}

答案 #

Netty Server完全支持在不返回响应的情况下单独关闭HTTP/2流（而非整个TCP连接）。

你当前的实现存在问题：DiscardOutBoundHandler中直接调用promise.channel().close()会关闭整个TCP连接，而HTTP/2是多路复用协议，同一连接上可能存在多个并发流，关闭连接会影响所有流的正常处理，不符合仅忽略单个请求的需求。

正确实现思路 #

HTTP/2中每个请求对应一个独立的流，要忽略单个请求，需要针对目标流发送RST_STREAM帧主动重置流，而非关闭整个连接。以下是两种可行的实现方式：

方式一：自定义Netty Handler处理HTTP/2流

编写专门针对HTTP/2流的Handler，识别流帧并发送重置指令：

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.Http2FrameCodec;
import io.netty.handler.codec.http2.Http2StreamFrame;

public class Http2StreamDiscardHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if (msg instanceof Http2StreamFrame streamFrame) {
            int streamId = streamFrame.streamId();
            // 发送RST_STREAM帧重置当前流，无响应返回
            ctx.channel().pipeline().get(Http2FrameCodec.class)
                .writeRSTStream(ctx, streamId, 0, promise);
            promise.setSuccess();
            return;
        }
        // 非HTTP/2流帧正常处理
        super.write(ctx, msg, promise);
    }
}

在NettyWebServerFactoryPortCustomizer中替换原来的Handler，注意要将Handler添加到Http2FrameCodec之后的位置，确保能正确处理HTTP/2帧：

serverFactory.addServerCustomizers(
        httpServer -> httpServer.doOnChannelInit((connectionObserver, channel, remoteAddress) -> {
            ChannelPipeline channelPipeline = channel.pipeline();
            // 找到Http2FrameCodec的位置，在其后添加自定义Handler
            int codecIndex = channelPipeline.names().indexOf("http2FrameCodec");
            if (codecIndex != -1) {
                channelPipeline.addAfter("http2FrameCodec", "streamDiscard", new Http2StreamDiscardHandler());
            }
        }));

方式二：Spring WebFlux过滤器层面处理

在WebFlux过滤器中判断需要忽略的请求，直接重置对应HTTP/2流：

import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono;

public class Http2StreamDiscardFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        // 根据业务规则判断是否需要忽略当前请求
        if (needIgnoreRequest(request)) {
            // 直接完成响应，并重置HTTP/2流
            return exchange.getResponse().setComplete()
                    .then(Mono.defer(() -> {
                        // 获取底层Netty的HTTP/2流信息并发送重置指令
                        return Mono.fromRunnable(() -> {
                            // 通过request.nativeRequest()获取Netty HttpRequest，进而获取流ID
                            // 找到Http2FrameCodec发送RST_STREAM帧
                            // 示例：
                            // io.netty.handler.codec.http.HttpRequest nettyReq = (io.netty.handler.codec.http.HttpRequest) request.nativeRequest();
                            // int streamId = ((Http2StreamChannel) nettyReq.channel()).stream().id();
                            // 后续调用Http2FrameCodec的writeRSTStream方法
                        });
                    }));
        }
        return chain.filter(exchange);
    }

    private boolean needIgnoreRequest(ServerHttpRequest request) {
        // 示例：忽略特定路径的请求
        return request.getPath().value().contains("/ignore");
    }
}

关键注意点 #

• 必须针对单个HTTP/2流操作，避免关闭整个TCP连接
• 确保Handler添加到Netty Pipeline的正确位置，保证能处理HTTP/2帧
• 发送RST_STREAM帧时可以指定错误码（第二个参数），根据场景选择合适的错误码（如0表示正常关闭）

内容的提问来源于stack exchange，提问作者Ankita Jadav