第十一章：WebSocket

本章介紹

WebSocket

ChannelHandler,Decoder and Encoder

引導一個Netty基礎程序

測試WebSocket

使用Netty附帶的WebSocket，咱們不須要關注協議內部實現，只須要使用Netty提供的一些簡單的方法就能夠實現

11.1 WebSockets some background

        關於WebSocket的一些概念和背景，能夠查詢網上相關介紹。這裏不贅述。

11.2 面臨的挑戰

要顯示「real-time」支持的WebSocket，應用程序將顯示如何使用Netty中的WebSocket實現一個在瀏覽器中進行聊天的IRC應用程序。

在這個應用程序中，不一樣的用戶能夠同時交談，很是像IRC(Internet Relay Chat，互聯網中繼聊天)。

上圖顯示的邏輯很簡單：

一個客戶端發送一條消息

消息被廣播到其餘已鏈接的客戶端

它的工做原理就像聊天室同樣，在這裏例子中，咱們將編寫服務器，而後使用瀏覽器做爲客戶端。帶着這樣的思路，咱們將會很簡單的實現它。

11.3 實現

WebSocket使用HTTP升級機制從一個普通的HTTP鏈接WebSocket，由於這個應用程序使用WebSocket老是開始於HTTP(s)，而後再升級。

在這裏，若是url的結尾以/ws結束，咱們將只會升級到WebSocket，不然服務器將發送一個網頁給客戶端。升級後的鏈接將經過WebSocket傳輸全部數據。邏輯圖以下：

11.3.1 處理http請求

服務器將做爲一種混合式以容許同時處理http和websocket，因此服務器還須要html頁面，html用來充當客戶端角色，鏈接服務器並交互消息。所以，若是客戶端不發送/ws的uri，咱們須要寫一個ChannelInboundHandler用來處理FullHttpRequest。看下面代碼：

package netty.in.action;  
import io.netty.channel.ChannelFuture;  
import io.netty.channel.ChannelFutureListener;  
import io.netty.channel.ChannelHandlerContext;  
import io.netty.channel.DefaultFileRegion;  
import io.netty.channel.SimpleChannelInboundHandler;  
import io.netty.handler.codec.http.DefaultFullHttpResponse;  
import io.netty.handler.codec.http.DefaultHttpResponse;  
import io.netty.handler.codec.http.FullHttpRequest;  
import io.netty.handler.codec.http.FullHttpResponse;  
import io.netty.handler.codec.http.HttpHeaders;  
import io.netty.handler.codec.http.HttpResponse;  
import io.netty.handler.codec.http.HttpResponseStatus;  
import io.netty.handler.codec.http.HttpVersion;  
import io.netty.handler.codec.http.LastHttpContent;  
import io.netty.handler.ssl.SslHandler;  
import io.netty.handler.stream.ChunkedNioFile;  
import java.io.RandomAccessFile;  
/** 
 * WebSocket，處理http請求 
 */  
public class HttpRequestHandler extends  
        SimpleChannelInboundHandler<FullHttpRequest> {  
    //websocket標識  
    private final String wsUri;  
  
    public HttpRequestHandler(String wsUri) {  
        this.wsUri = wsUri;  
    }  
    @Override  
    protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg)  
            throws Exception {  
        //若是是websocket請求，請求地址uri等於wsuri  
        if (wsUri.equalsIgnoreCase(msg.getUri())) {  
            //將消息轉發到下一個ChannelHandler  
            ctx.fireChannelRead(msg.retain());  
        } else {//若是不是websocket請求  
            if (HttpHeaders.is100ContinueExpected(msg)) {  
                //若是HTTP請求頭部包含Expect: 100-continue，  
                //則響應請求  
                FullHttpResponse response = new DefaultFullHttpResponse(  
                        HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);  
                ctx.writeAndFlush(response);  
            }  
            //獲取index.html的內容響應給客戶端  
            RandomAccessFile file = new RandomAccessFile(  
                    System.getProperty("user.dir") + "/index.html", "r");  
            HttpResponse response = new DefaultHttpResponse(  
                    msg.getProtocolVersion(), HttpResponseStatus.OK);  
            response.headers().set(HttpHeaders.Names.CONTENT_TYPE,  
                    "text/html; charset=UTF-8");  
            boolean keepAlive = HttpHeaders.isKeepAlive(msg);  
            //若是http請求保持活躍，設置http請求頭部信息  
            //並響應請求  
            if (keepAlive) {  
                response.headers().set(HttpHeaders.Names.CONTENT_LENGTH,  
                        file.length());  
                response.headers().set(HttpHeaders.Names.CONNECTION,  
                        HttpHeaders.Values.KEEP_ALIVE);  
            }  
            ctx.write(response);  
            //若是不是https請求，將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()));  
            }  
            //標識響應內容結束並刷新通道  
            ChannelFuture future = ctx  
                    .writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);  
            if (!keepAlive) {  
                //若是http請求不活躍，關閉http鏈接  
                future.addListener(ChannelFutureListener.CLOSE);  
            }  
            file.close();  
        }  
    }  
    @Override  
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  
            throws Exception {  
        cause.printStackTrace();  
        ctx.close();  
    }  
}

11.3.2 處理WebSocket框架

WebSocket支持6種不一樣框架，以下圖：

咱們的程序只須要使用下面4個框架：

CloseWebSocketFrame

PingWebSocketFrame

PongWebSocketFrame

TextWebSocketFrame

咱們只須要顯示處理TextWebSocketFrame，其餘的會自動由WebSocketServerProtocolHandler處理，看下面代碼：

package netty.in.action;  
import io.netty.channel.ChannelHandlerContext;  
import io.netty.channel.SimpleChannelInboundHandler;  
import io.netty.channel.group.ChannelGroup;  
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;  
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;  
/** 
 * 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 {  
        //若是WebSocket握手完成  
        if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {  
            //刪除ChannelPipeline中的HttpRequestHandler  
            ctx.pipeline().remove(HttpRequestHandler.class);  
            //寫一個消息到ChannelGroup  
            group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel()  
                    + " joined"));  
            //將Channel添加到ChannelGroup  
            group.add(ctx.channel());  
        }else {  
            super.userEventTriggered(ctx, evt);  
        }  
    }  
  
    @Override  
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg)  
            throws Exception {  
        //將接收的消息經過ChannelGroup轉發到因此已鏈接的客戶端  
        group.writeAndFlush(msg.retain());  
    }  
}

11.3.3 初始化ChannelPipeline

 看下面代碼：

package netty.in.action;  
import io.netty.channel.Channel;  
import io.netty.channel.ChannelInitializer;  
import io.netty.channel.ChannelPipeline;  
import io.netty.channel.group.ChannelGroup;  
import io.netty.handler.codec.http.HttpObjectAggregator;  
import io.netty.handler.codec.http.HttpServerCodec;  
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;  
import io.netty.handler.stream.ChunkedWriteHandler;  
/** 
 * WebSocket,初始化ChannelHandler 
 */  
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();  
        //編解碼http請求  
        pipeline.addLast(new HttpServerCodec());  
        //寫文件內容  
        pipeline.addLast(new ChunkedWriteHandler());  
        //聚合解碼HttpRequest/HttpContent/LastHttpContent到FullHttpRequest  
        //保證接收的Http請求的完整性  
        pipeline.addLast(new HttpObjectAggregator(64 * 1024));  
        //處理FullHttpRequest  
        pipeline.addLast(new HttpRequestHandler("/ws"));  
        //處理其餘的WebSocketFrame  
        pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));  
        //處理TextWebSocketFrame  
        pipeline.addLast(new TextWebSocketFrameHandler(group));  
    }  
}

WebSocketServerProtcolHandler不只處理Ping/Pong/CloseWebSocketFrame，還和它本身握手並幫助升級WebSocket。這是執行完成握手和成功修改ChannelPipeline，而且添加須要的編碼器/解碼器和刪除不須要的ChannelHandler。

 看下圖：

ChannelPipeline經過ChannelInitializer的initChannel(...)方法完成初始化，完成握手後就會更改事情。一旦這樣作了，WebSocketServerProtocolHandler將取代HttpRequestDecoder、WebSocketFrameDecoder13和HttpResponseEncoder、WebSocketFrameEncoder13。另外也要刪除全部不須要的ChannelHandler已得到最佳性能。這些都是HttpObjectAggregator和HttpRequestHandler。下圖顯示ChannelPipeline握手完成：

咱們甚至沒注意到它，由於它是在底層執行的。以很是靈活的方式動態更新ChannelPipeline讓單獨的任務在不一樣的ChannelHandler中實現。

11.4 結合在一塊兒使用

一如既往，咱們要將它們結合在一塊兒使用。使用Bootstrap引導服務器和設置正確的ChannelInitializer。看下面代碼：

package netty.in.action;  
  
import io.netty.bootstrap.ServerBootstrap;  
import io.netty.channel.Channel;  
import io.netty.channel.ChannelFuture;  
import io.netty.channel.ChannelInitializer;  
import io.netty.channel.EventLoopGroup;  
import io.netty.channel.group.ChannelGroup;  
import io.netty.channel.group.DefaultChannelGroup;  
import io.netty.channel.nio.NioEventLoopGroup;  
import io.netty.channel.socket.nio.NioServerSocketChannel;  
import io.netty.util.concurrent.ImmediateEventExecutor;  
  
import java.net.InetSocketAddress;  
/** 
 * 訪問地址：http://localhost:2048 
 */  
public class ChatServer {  
  
    private final ChannelGroup group = new DefaultChannelGroup(  
            ImmediateEventExecutor.INSTANCE);  
    private final EventLoopGroup workerGroup = new NioEventLoopGroup();  
    private Channel channel;  
  
    public ChannelFuture start(InetSocketAddress address) {  
        ServerBootstrap b = new ServerBootstrap();  
        b.group(workerGroup).channel(NioServerSocketChannel.class)  
                .childHandler(createInitializer(group));  
        ChannelFuture f = b.bind(address).syncUninterruptibly();  
        channel = f.channel();  
        return f;  
    }  
  
    public void destroy() {  
        if (channel != null)  
            channel.close();  
        group.close();  
        workerGroup.shutdownGracefully();  
    }  
  
    protected ChannelInitializer<Channel> createInitializer(ChannelGroup group) {  
        return new ChatServerInitializer(group);  
    }  
  
    public static void main(String[] args) {  
        final ChatServer server = new ChatServer();  
        ChannelFuture f = server.start(new InetSocketAddress(2048));  
        Runtime.getRuntime().addShutdownHook(new Thread() {  
            @Override  
            public void run() {  
                server.destroy();  
            }  
        });  
        f.channel().closeFuture().syncUninterruptibly();  
    }  
}

另外，須要將index.html文件放在項目根目錄，index.html內容以下：

<html>  
<head>  
<title>Web Socket Test</title>  
</head>  
<body>  
<script type="text/javascript">  
var socket;  
if (!window.WebSocket) {  
  window.WebSocket = window.MozWebSocket;  
}  
if (window.WebSocket) {  
  socket = new WebSocket("ws://localhost:2048/ws");  
  socket.onmessage = function(event) {  
    var ta = document.getElementById('responseText');  
    ta.value = ta.value + '\n' + event.data  
  };  
  socket.onopen = function(event) {  
    var ta = document.getElementById('responseText');  
    ta.value = "Web Socket opened!";  
  };  
  socket.onclose = function(event) {  
    var ta = document.getElementById('responseText');  
    ta.value = ta.value + "Web Socket closed";   
  };  
} else {  
  alert("Your browser does not support Web Socket.");  
}  
  
function send(message) {  
  if (!window.WebSocket) { return; }  
  if (socket.readyState == WebSocket.OPEN) {  
    socket.send(message);  
  } else {  
    alert("The socket is not open.");  
  }  
}  
</script>  
    <form onsubmit="return false;">  
        <input type="text" name="message" value="Hello, World!"><input  
            type="button" value="Send Web Socket Data"  
            onclick="send(this.form.message.value)">  
        <h3>Output</h3>  
        <textarea id="responseText" style="width: 500px; height: 300px;"></textarea>  
    </form>  
</body>  
</html>

最後在瀏覽器中輸入：http://localhost:2048，多開幾個窗口就能夠聊天了。

11.5 給WebSocket加密

上面的應用程序雖然工做的很好，可是在網絡上收發消息存在很大的安全隱患，因此有必要對消息進行加密。添加這樣一個加密的功能通常比較複雜，須要對代碼有較大的改動。可是使用Netty就能夠很容易的添加這樣的功能，只須要將SslHandler加入到ChannelPipeline中就能夠了。實際上還須要添加SslContext，但這不在本例子範圍內。

首先咱們建立一個用於添加加密Handler的handler初始化類，看下面代碼：

package netty.in.action;  
import io.netty.channel.Channel;  
import io.netty.channel.group.ChannelGroup;  
import io.netty.handler.ssl.SslHandler;  
import javax.net.ssl.SSLContext;  
import javax.net.ssl.SSLEngine;  
public class SecureChatServerIntializer extends ChatServerInitializer {  
    private final SSLContext context;  
    public SecureChatServerIntializer(ChannelGroup group,SSLContext context) {  
        super(group);  
        this.context = context;  
    }  
  
    @Override  
    protected void initChannel(Channel ch) throws Exception {  
        super.initChannel(ch);  
        SSLEngine engine = context.createSSLEngine();  
        engine.setUseClientMode(false);  
        ch.pipeline().addFirst(new SslHandler(engine));  
    }  
}

最後咱們建立一個用於引導配置的類，看下面代碼：

package netty.in.action;  
import io.netty.channel.Channel;  
import io.netty.channel.ChannelFuture;  
import io.netty.channel.ChannelInitializer;  
import io.netty.channel.group.ChannelGroup;  
import java.net.InetSocketAddress;  
import javax.net.ssl.SSLContext;  
/** 
 * 訪問地址：https://localhost:4096 
 */  
public class SecureChatServer extends ChatServer {  
    private final SSLContext context;  
  
    public SecureChatServer(SSLContext context) {  
        this.context = context;  
    }  
  
    @Override  
    protected ChannelInitializer<Channel> createInitializer(ChannelGroup group) {  
        return new SecureChatServerIntializer(group, context);  
    }  
  
    /** 
     * 獲取SSLContext須要相關的keystore文件，這裏沒有 關於HTTPS能夠查閱相關資料，這裏只介紹在Netty中如何使用 
     *  
     * @return 
     */  
    private static SSLContext getSslContext() {  
        return null;  
    }  
  
    public static void main(String[] args) {  
        SSLContext context = getSslContext();  
        final SecureChatServer server = new SecureChatServer(context);  
        ChannelFuture future = server.start(new InetSocketAddress(4096));  
        Runtime.getRuntime().addShutdownHook(new Thread() {  
            @Override  
            public void run() {  
                server.destroy();  
            }  
        });  
        future.channel().closeFuture().syncUninterruptibly();  
    }  
}