Netty教程 / 第 90 节

第9章：常用协议支持

本章导读

Netty 内置了对多种常用协议的支持，包括 HTTP、WebSocket、SSL/TLS 等。本章将讲解如何使用这些内置支持快速构建应用。

9.1 HTTP 协议支持

9.1.1 HTTP 服务器

public class HttpServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            
                            // HTTP 编解码器
                            pipeline.addLast(new HttpServerCodec());
                            
                            // HTTP 聚合器（将多个HTTP消息聚合成一个完整的HTTP请求或响应）
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            
                            // 业务Handler
                            pipeline.addLast(new HttpServerHandler());
                        }
                    });
            
            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("HTTP服务器启动: http://localhost:8080");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
        // 获取请求信息
        String uri = request.uri();
        HttpMethod method = request.method();
        
        System.out.println("收到请求: " + method + " " + uri);
        
        // 构建响应
        String content = "<html><body><h1>Hello Netty HTTP!</h1></body></html>";
        ByteBuf buf = Unpooled.copiedBuffer(content, CharsetUtil.UTF_8);
        
        FullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK,
                buf
        );
        
        // 设置响应头
        response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html; charset=UTF-8");
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, buf.readableBytes());
        
        // 发送响应
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
    }
}

9.1.2 HTTP 客户端

public class HttpClient {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            
                            // HTTP 编解码器
                            pipeline.addLast(new HttpClientCodec());
                            
                            // HTTP 聚合器
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            
                            // 业务Handler
                            pipeline.addLast(new HttpClientHandler());
                        }
                    });
            
            ChannelFuture future = bootstrap.connect("localhost", 8080).sync();
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

class HttpClientHandler extends SimpleChannelInboundHandler<FullHttpResponse> {
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 构建HTTP请求
        FullHttpRequest request = new DefaultFullHttpRequest(
                HttpVersion.HTTP_1_1,
                HttpMethod.GET,
                "/"
        );
        
        request.headers().set(HttpHeaderNames.HOST, "localhost");
        request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
        
        // 发送请求
        ctx.writeAndFlush(request);
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpResponse response) {
        System.out.println("状态: " + response.status());
        System.out.println("响应内容: " + response.content().toString(CharsetUtil.UTF_8));
    }
}

9.2 WebSocket 协议

9.2.1 WebSocket 服务器

public class WebSocketServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            
                            // HTTP 编解码器
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            
                            // WebSocket 协议处理器
                            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
                            
                            // 业务Handler
                            pipeline.addLast(new WebSocketServerHandler());
                        }
                    });
            
            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("WebSocket服务器启动: ws://localhost:8080/ws");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        String text = msg.text();
        System.out.println("收到消息: " + text);
        
        // 回复消息
        ctx.writeAndFlush(new TextWebSocketFrame("服务器收到: " + text));
    }
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        System.out.println("客户端连接: " + ctx.channel().id());
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        System.out.println("客户端断开: " + ctx.channel().id());
    }
}

9.2.2 WebSocket 客户端（HTML）

<!DOCTYPE html>
<html>
<head>
    <title>WebSocket Client</title>
</head>
<body>
    <h1>WebSocket 测试</h1>
    <input type="text" id="message" placeholder="输入消息">
    <button onclick="send()">发送</button>
    <div id="output"></div>
    
    <script>
        const ws = new WebSocket('ws://localhost:8080/ws');
        
        ws.onopen = function() {
            console.log('连接成功');
            document.getElementById('output').innerHTML += '<p>连接成功</p>';
        };
        
        ws.onmessage = function(event) {
            console.log('收到消息:', event.data);
            document.getElementById('output').innerHTML += '<p>收到: ' + event.data + '</p>';
        };
        
        ws.onclose = function() {
            console.log('连接关闭');
            document.getElementById('output').innerHTML += '<p>连接关闭</p>';
        };
        
        function send() {
            const message = document.getElementById('message').value;
            ws.send(message);
            document.getElementById('output').innerHTML += '<p>发送: ' + message + '</p>';
        }
    </script>
</body>
</html>

9.3 HTTPS/SSL/TLS

9.3.1 生成证书

# 生成自签名证书
keytool -genkey -alias netty -keyalg RSA -keystore netty.jks -keysize 2048 -validity 365

9.3.2 HTTPS 服务器

public class HttpsServer {
    public static void main(String[] args) throws Exception {
        // 加载证书
        SslContext sslContext = SslContextBuilder
                .forServer(new File("netty.crt"), new File("netty.key"))
                .build();
        
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            
                            // SSL Handler（必须在最前面）
                            pipeline.addLast(sslContext.newHandler(ch.alloc()));
                            
                            // HTTP 编解码器
                            pipeline.addLast(new HttpServerCodec());
                            pipeline.addLast(new HttpObjectAggregator(65536));
                            
                            // 业务Handler
                            pipeline.addLast(new HttpServerHandler());
                        }
                    });
            
            ChannelFuture future = bootstrap.bind(8443).sync();
            System.out.println("HTTPS服务器启动: https://localhost:8443");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

9.4 Protocol Buffers 集成

9.4.1 定义 .proto 文件

syntax = "proto3";

option java_package = "com.netty.tutorial.proto";
option java_outer_classname = "UserProto";

message User {
    int32 id = 1;
    string name = 2;
    int32 age = 3;
}

9.4.2 使用 Protobuf

// 服务端
pipeline.addLast(new ProtobufVarint32FrameDecoder());
pipeline.addLast(new ProtobufDecoder(UserProto.User.getDefaultInstance()));
pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast(new ProtobufEncoder());
pipeline.addLast(new ProtobufHandler());

class ProtobufHandler extends SimpleChannelInboundHandler<UserProto.User> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, UserProto.User user) {
        System.out.println("收到用户: " + user.getName() + ", 年龄: " + user.getAge());
        
        // 回复
        UserProto.User response = UserProto.User.newBuilder()
                .setId(user.getId())
                .setName("Server")
                .setAge(0)
                .build();
        
        ctx.writeAndFlush(response);
    }
}

// 客户端发送
UserProto.User user = UserProto.User.newBuilder()
        .setId(1)
        .setName("Alice")
        .setAge(25)
        .build();

ctx.writeAndFlush(user);

9.5 JSON 编解码

9.5.1 使用 Jackson

// 添加依赖
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.14.0</version>
</dependency>

// JSON 编码器
public class JsonEncoder extends MessageToByteEncoder<Object> {
    
    private static final ObjectMapper mapper = new ObjectMapper();
    
    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        byte[] json = mapper.writeValueAsBytes(msg);
        out.writeInt(json.length);
        out.writeBytes(json);
    }
}

// JSON 解码器
public class JsonDecoder extends ByteToMessageDecoder {
    
    private static final ObjectMapper mapper = new ObjectMapper();
    private final Class<?> clazz;
    
    public JsonDecoder(Class<?> clazz) {
        this.clazz = clazz;
    }
    
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        if (in.readableBytes() < 4) {
            return;
        }
        
        in.markReaderIndex();
        int length = in.readInt();
        
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        
        byte[] json = new byte[length];
        in.readBytes(json);
        
        Object obj = mapper.readValue(json, clazz);
        out.add(obj);
    }
}

// 使用
pipeline.addLast(new JsonDecoder(User.class));
pipeline.addLast(new JsonEncoder());

9.6 本章小结

本章我们学习了常用协议支持：

✅ HTTP：HttpServerCodec、HttpObjectAggregator
✅ WebSocket：WebSocketServerProtocolHandler
✅ HTTPS/SSL/TLS：SslContext、SslHandler
✅ Protocol Buffers：高效的二进制序列化
✅ JSON：自定义 JSON 编解码器

关键要点

Netty 内置了对主流协议的支持
HTTP 和 WebSocket 可以在同一个端口上共存
SSL Handler 必须放在 Pipeline 的最前面
Protocol Buffers 性能优于 JSON
选择合适的序列化方案很重要

上一章：第8章：粘包与拆包解决方案
下一章：第10章：EventLoop与线程模型