netty(4)高級篇-Websocket協議開發

一、HTTP協議的弊端

將HTTP協議的主要弊端總結如下：

• (1) 半雙工協議：可以在客戶端和服務端2個方向上傳輸，但是不能同時傳輸。同一時刻，只能在一個方向上傳輸。
• (2) HTTP消息冗長：相比於其他二進制協議，有點繁瑣。
• (3) 針對服務器推送的黑客攻擊，例如長時間輪詢。

 

現在很多網站的消息推送都是使用輪詢，即客戶端每隔1S或者其他時間給服務器發送請求，然后服務器返回最新的數據給客戶端。HTTP協議中的Header非常冗長，因此會占用很多的帶寬和服務器資源。

比較新的技術是Comet，使用了AJAX。雖然可以雙向通信，但是依然需要發送請求，而且在Comet中，普遍采用了長連接，也會大量消耗服務器的帶寬和資源。

為了解決這個問題，HTML5定義的WebSocket協議。

二、WebSocket協議介紹

在WebSocket API中，瀏覽器和服務器只需要一個握手的動作，然后，瀏覽器和服務器之間就形成了一條快速通道，兩者就可以直接互相傳送數據了。

WebSocket基於TCP雙向全雙工協議，即在同一時刻，即可以發送消息，也可以接收消息，相比於HTTP協議，是一個性能上的提升。

特點：

• 　　單一的TCP連接，全雙工；
• 　　對代理、防火牆和路由器透明；
• 　　無頭部信息、Cookie和身份驗證；
• 　　無安全開銷；
• 　　通過"ping/pong"幀保持鏈路激活;
• 　　服務器可以主動傳遞消息給客戶端，不再需要客戶端輪詢；

擁有以上特點的WebSocket就是為了取代輪詢和Comet技術，使得客戶端瀏覽器具備像C/S架構下桌面系統一樣的實時能力。

瀏覽器通過js建立一個WebSocket的請求，連接建立后，客戶端和服務器端可以通過TCP直接交換數據。

因為WebSocket本質上是一個TCP連接，穩定，所以在Comet和輪詢比擁有性能優勢，如圖所示：

 

 

三、WebSocket連接

3.1 連接建立

client端發送握手請求，請求消息如圖所示：

 

• 這個請求和普通的HTTP請求不同，包含了一些附加頭信息，其中附加頭信息"Upgrade: Websocket"表明這是一個申請協議升級的HTTP請求。
• 服務器嘗試解析這個信息，然后返回應答信息給客戶端，因此客戶端和服務器端的WebSocket連接就建立起來了，雙方可以通過這個連接通道自由的傳遞信息。
• 這個連接會持續到某一方主動斷開連接。

服務端的應答請求如圖所示：

 

 client消息中的"Sec-WebSocket-Key"是隨機的，服務器端會用這些數據來構造一個"SHA-1"的信息摘要，把"Sec-WebSocket-Key"加上一個魔幻字符串。使用"SHA-1"加密，然后進行BASE64編碼，將結果作為"Sec-Webscoket-Accept"頭的值。

3.2 生命周期

• 握手成功，連接建立后，以"Messages"的方式通信。
• 一個消息由一個或者多個"幀"組成。
• 幀都有自己的類型，同一消息的多個幀類型相同。
• 廣義上，類型可以是文本、二進制、控制幀如信號。

 

3.3 連接關閉

• 安全方法是關閉底層TCP連接以及TLS會話。
• 底層的TCP連接，正常情況下，應該由服務器先關閉。
• 異常時(比如合理的時間內沒有接收到服務器的TCP Close)，可以由客戶端發起TCP Close。因此，在client發起TCP Close時，服務器應該立即發起一個TCP Close操作；客戶端則等待服務器的TCP Close；
• 關閉消息帶有一個狀態碼和可選的關閉原因，它必須按照協議要求發送一個Close控制幀。

四、協議開發

官方demo: http://netty.io/4.1/xref/io/netty/example/http/websocketx/server/package-summary.html

功能介紹：

服務器端開發：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.stream.ChunkedWriteHandler;

/**
 * @author lilinfeng
 * @version 1.0
 * @date 2014年2月14日
 */
public class WebSocketServer {
    public void run(int port) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch)
                                throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("http-codec",
                                    new HttpServerCodec());
                            pipeline.addLast("aggregator",
                                    new HttpObjectAggregator(65536));
                            ch.pipeline().addLast("http-chunked",
                                    new ChunkedWriteHandler());
                            pipeline.addLast("handler",
                                    new WebSocketServerHandler());
                        }
                    });

            Channel ch = b.bind(port).sync().channel();
            System.out.println("Web socket server started at port " + port
                    + '.');
            System.out
                    .println("Open your browser and navigate to http://localhost:"
                            + port + '/');

            ch.closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args.length > 0) {
            try {
                port = Integer.parseInt(args[0]);
            } catch (NumberFormatException e) {
                e.printStackTrace();
            }
        }
        new WebSocketServer().run(port);
    }
}

 

HttpServerCodec：將請求和應答消息解碼為HTTP消息

HttpObjectAggregator：將HTTP消息的多個部分合成一條完整的HTTP消息

ChunkedWriteHandler：向客戶端發送HTML5文件

看上去和HTTP協議的非常類似，下面從Handler中來尋找答案：

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;

import java.util.logging.Level;
import java.util.logging.Logger;

import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

/**
 * @author lilinfeng
 * @version 1.0
 * @date 2014年2月14日
 */
public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    private static final Logger logger = Logger
            .getLogger(WebSocketServerHandler.class.getName());

    private WebSocketServerHandshaker handshaker;


    @Override
    public void channelRead0(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        // 傳統的HTTP接入
        if (msg instanceof FullHttpRequest) {
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        }
        // WebSocket接入
        else if (msg instanceof WebSocketFrame) {
            handleWebSocketFrame(ctx, (WebSocketFrame) msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    private void handleHttpRequest(ChannelHandlerContext ctx,
                                   FullHttpRequest req) throws Exception {

        // 如果HTTP解碼失敗，返回HHTP異常
        if (!req.decoderResult().isSuccess()
                || (!"websocket".equals(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1,
                    BAD_REQUEST));
            return;
        }

        // 構造握手響應返回，本機測試
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws://localhost:8080/websocket", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory
                    .sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

    private void handleWebSocketFrame(ChannelHandlerContext ctx,
                                      WebSocketFrame frame) {

        // 判斷是否是關閉鏈路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(),
                    (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判斷是否是Ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(
                    new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 本例程僅支持文本消息，不支持二進制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format(
                    "%s frame types not supported", frame.getClass().getName()));
        }

        // 返回應答消息
        String request = ((TextWebSocketFrame) frame).text();
        if (logger.isLoggable(Level.FINE)) {
            logger.fine(String.format("%s received %s", ctx.channel(), request));
        }
        ctx.channel().write(
                new TextWebSocketFrame(request
                        + " , 歡迎使用Netty WebSocket服務，現在時刻："
                        + new java.util.Date().toString()));
    }

    private static void sendHttpResponse(ChannelHandlerContext ctx,
                                         FullHttpRequest req, FullHttpResponse res) {
        // 返回應答給客戶端
        if (res.getStatus().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.getStatus().toString(),
                    CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
            HttpUtil.setContentLength(res, res.content().readableBytes());
        }

        // 如果是非Keep-Alive，關閉連接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

(1) 第一次握手由HTTP協議承載，所以是一個HTTP消息，根據消息頭中是否包含"Upgrade"字段來判斷是否是websocket。

(2) 通過校驗后，構造WebSocketServerHandshaker，通過它構造握手響應信息返回給客戶端，同時將WebSocket相關的編碼和解碼類動態添加到ChannelPipeline中。

下面分析鏈路建立之后的操作：

(1) 客戶端通過文本框提交請求給服務端，Handler收到之后已經解碼之后的WebSocketFrame消息。

(2) 如果是關閉按鏈路的指令就關閉鏈路

(3) 如果是維持鏈路的ping消息就返回Pong消息。

(4) 否則就返回應答消息

五、客戶端以及測試

html5中的JS代碼：

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    Netty WebSocket 時間服務器
</head>
<br>
<body>
<br>
<script type="text/javascript">
    var socket;
    if (!window.WebSocket) {
        window.WebSocket = window.MozWebSocket;
    }
    if (window.WebSocket) {
        socket = new WebSocket("ws://localhost:8080/websocket");
        socket.onmessage = function (event) {
            var ta = document.getElementById('responseText');
            ta.value = "";
            ta.value = event.data
        };
        socket.onopen = function (event) {
            var ta = document.getElementById('responseText');
            ta.value = "打開WebSocket服務正常，瀏覽器支持WebSocket!";
        };
        socket.onclose = function (event) {
            var ta = document.getElementById('responseText');
            ta.value = "";
            ta.value = "WebSocket 關閉!";
        };
    }
    else {
        alert("抱歉，您的瀏覽器不支持WebSocket協議!");
    }

    function send(message) {
        if (!window.WebSocket) {
            return;
        }
        if (socket.readyState == WebSocket.OPEN) {
            socket.send(message);
        }
        else {
            alert("WebSocket連接沒有建立成功!");
        }
    }
</script>
<form onsubmit="return false;">
    <input type="text" name="message" value="Netty最佳實踐"/>
    <br><br>
    <input type="button" value="發送WebSocket請求消息" onclick="send(this.form.message.value)"/>
    <hr color="blue"/>
    <h3>服務端返回的應答消息</h3>
    <textarea id="responseText" style="width:500px;height:300px;"></textarea>
</form>
</body>
</html>

演示效果大致如下：

 

這里只是對WebSocket協議最基本的演示，WebSocket支持多種協議，文本，二進制，控制幀。