Netty 框架学习 —— 添加 WebSocket 支持

WebSocket 简介

WebSocket 协议是完全重新设计的协议，旨在为 Web 上的双向数据传输问题提供一个切实可行的解决方案，使得客户端和服务器之间可以在任意时刻传输消息

Netty 对于 WebSocket 的支持包含了所有正在使用钟的主要实现，我们将通过创建一个基于 WebSocket 的实时聊天应用程序来演示这一点

WebSocket 应用程序示例

我们将通过使用 WebSocket 协议来实现一个基于浏览器的聊天应用程序，使得多个用户之间可以同时进行相互通信

下图说明了该应用程序的逻辑：

1. 客户端发送一个消息
2. 该消息将被广播到所有其他连接的客户端

所有人都在可以和其他人聊天，在示例中，我们只实现服务器，客户端则是通过 Web 页面访问该聊天室的浏览器

1. 添加 WebSocket 支持

在从标准的 HTTP 或者 HTTPS 协议切换到 WebSocket 时，将会使用一种称为升级握手的机制，使用 WebSocket 的应用协议将始终以 HTTP/S 作为开始，然后再执行升级。这个升级动作发生的确切时刻特定于应用程序，可能会发生在启动时，也可能会发生在请求了某个特定的 URL 之后

我们的应用程序将采用如下约定：如果被请求的 URL 以 /ws 结尾，那么把该协议升级为 WebSocket，否则服务器将使用基本的 HTTP/S

下图解释了 Netty 如何处理 HTTP 以及 WebSocket 协议技术，它由一组 ChannelHandler 实现

2. 处理 HTTP 请求

首先，我们实现处理 HTTP 请求的组件，这个组件将提供用于访问聊天室并显示由连接的客户端发送的消息的网页

public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final String wsUri;
    private static final File INDEX;

    static {
        URL location = HttpRequestHandler.class.getProtectionDomain().getCodeSource().getLocation();
        try {
            String path = location.toURI() + "index.html";
            path = !path.contains("file") ? path : path.substring(5);
            INDEX = new File(path);
        } catch (URISyntaxException e) {
            throw new IllegalStateException("Unable to locate index.html", e);
        }
    }

    public HttpRequestHandler(String wsUri) {
        this.wsUri = wsUri;
    }

    @Override
    protected void messageReceived(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
        if (wsUri.equalsIgnoreCase(request.uri())) {
            // 如果请求了 WebSocket 协议升级，则增加引用计数，并将它传递给下一个 ChannelInboundHandler
            ctx.fireChannelRead(request.retain());
        } else {
            // 读取 index.html
            RandomAccessFile file = new RandomAccessFile(INDEX, "r");
            DefaultHttpResponse response = new DefaultHttpResponse(request.protocolVersion(), HttpResponseStatus.OK);
            response.headers().set("CONTENT_TYPE", "text/html; charset=UTF-8");
            // 将 HttpResponse 写到客户端
            ctx.write(response);
            // 将 index.html 写到客户端
            if (ctx.pipeline().get(SslHandler.class) == null) {
                ctx.write(new DefaultFileRegion(file.getChannel(), 0, file.length()));
            } else {
                ctx.write(new ChunkedNioFile(file.getChannel()));
            }
            // 写 LastHttpContent 并冲刷到客户端
            ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
            // 写操作完成后关闭 Channel
            future.addListener(ChannelFutureListener.CLOSE);
        }
    }
}

3. 处理 WebSocket 帧

由 IETF 发布的 WebSocket RFC 定义了六种帧，Netty 为它们每种都提供了一个 POJO 实现。下表列出了这些帧类型，并描述了它们的用法

下述代码展示了用于处理 TextWebSocketFrame 的 ChannelInboundHandler，其还将在它的 ChannelGroup 中跟踪所有活动的 WebSocket 连接

public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private final ChannelGroup group;

    public TextWebSocketFrameHandler(ChannelGroup group) {
        this.group = group;
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
            // 如果该事件握手成功，则移除 HttpRequestHandler，因为不会再接收到任何 HTTP 消息了
            ctx.pipeline().remove(HttpRequestHandler.class);
            // 通知所有已经连接的 WebSocket 客户端新的客户端已经连接上了
            group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel() + " joined"));
            // 将新的 WebSocket Channel 添加到 ChannelGroup
            group.add(ctx.channel());
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    protected void messageReceived(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        // 增加消息的引用计数，并将它写到 ChannelGroup 中所有已经连接的客户端
        group.writeAndFlush(msg.retain());
    }
}

4. 初始化 ChannelPipeline

为了将 ChannelHandler 安装到 ChannelPipeline 中，我们需要扩展 ChannelInitializer 并实现 initChannel() 方法

public class ChatServerInitializer extends ChannelInitializer<Channel> {

    private final ChannelGroup group;

    public ChatServerInitializer(ChannelGroup group) {
        this.group = group;
    }

    @Override
    protected void initChannel(Channel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new HttpServerCodec());
        pipeline.addLast(new ChunkedWriteHandler());
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));
        pipeline.addLast(new HttpRequestHandler("/ws"));
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
        pipeline.addLast(new TextWebSocketFrameHandler(group));
    }
}

对于 initChannel() 方法的调用，通过安装所有必需的 ChannelHandler 来设置该新注册的 Channel 的 ChannelPipeline。Netty 的 WebSocketServerProtocolHandler 处理了所有委托管理的 WebSocket 帧类型以及升级握手本身。如果握手本身，那么所需的 ChannelHandler 将被添加到 ChannelPipeline 中，而那些不再需要的 ChannelHandler 则会被移除

5. 引导

最后一步是引导该服务器，并安装 ChatServerInitializer 的代码，这将由 ChatServer 类处理

public class ChatServer {

    private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
    private final EventLoopGroup group = new NioEventLoopGroup();
    private Channel channel;

    public ChannelFuture start(InetSocketAddress address) {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .childHandler(createInitializer(channelGroup));
        ChannelFuture future = bootstrap.bind(address);
        future.syncUninterruptibly();
        channel = future.channel();
        return future;
    }

    protected ChannelInitializer<Channel> createInitializer(ChannelGroup group) {
        return new ChatServerInitializer(group);
    }

    public void destroy() {
        if (channel != null) {
            channel.close();
        }
        channelGroup.close();
        group.shutdownGracefully();
    }

    public static void main(String[] args) {
        if (args.length != 1) {
            System.err.println("Please give port as argument");
            System.exit(1);
        }
        int port = Integer.parseInt(args[0]);
        final ChatServer endpoint = new ChatServer();
        ChannelFuture future = endpoint.start(new InetSocketAddress(port));
        Runtime.getRuntime().addShutdownHook(new Thread(endpoint::destroy));
        future.channel().closeFuture().syncUninterruptibly();
    }
}