Netty就是这么回事（九）

很久没更新了，以前一段时间一直在忙双11，外加其余项目实在是没有时间写，好在如今快过春节了，终于有时间写了。这一章主要介绍netty的websocket协议开发，websocket协议其实不是很新鲜了，可是比较重要，自己设计它的目的就是为了取代轮训和Comet技术，使客户端具备像C/S架构下桌面系统同样的实时通讯能力。websocket协议有以下特色：

• 单一的TCP链接，采用全双工模式通讯；
• 对代理、防火墙和路由器透明；
• 无头部信息、Cookie和身份验证；
• 无安全开销；
• 经过”ping/pong“帧保持链路激活；
• 服务端能够主动传递消息给客户端，再也不须要客户端轮训；

下面看下websocket在netty是怎么实现的？废话很少说，直接上代码。

服务端代码：

package com.dlb.note.websocket;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
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.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;

/**
 * 做者：dlb
 * 功能: websocket服务器
 * 日期: 2018/1/16 11:26
 * 版本：V0.1
 */
public class WebSocketServer {
    /**
     * 主函数
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        int port = 8888;
        new WebSocketServer().start(port);
    }

    /**
     * 执行函数
     * @param port
     * @throws Exception
     */
    public void start(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 {
                            ch.pipeline()
                                    // 将请求和应答消息编码或解码为HTTP消息
                                    .addLast("http-codec", new HttpServerCodec())
                                    // 将HTTP消息的多个部分组合成一个完整的HTTP消息
                                    .addLast("aggregator", new HttpObjectAggregator(65536))
                                    // 向客户端发送HTML5文件，主要用于支持浏览器和服务端进行Websocket通讯
                                    .addLast("http-chunked", new ChunkedWriteHandler())
                                    // websocket处理器
                                    .addLast("handler", new WebSocketServerHandler());
                        }
                    });

            Channel ch = b.bind(port).sync().channel();
            System.out.println("Websocket server start,port=" + port);

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

/**
 * websocket处理器
 */
class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {
    private WebSocketServerHandshaker handshaker;

    @Override
    protected void messageReceived(ChannelHandlerContext channelHandlerContext, Object o)
            throws Exception {
        if (o instanceof WebSocketFrame) { // websocket
            System.out.println("websocket body=【 " + o + " 】");
            handleWebSocketFrame(channelHandlerContext, (WebSocketFrame) o);
        }else if(o instanceof FullHttpRequest){ // http
            System.out.println("http body=【 " + o + " 】");
            handleHttpRequest(channelHandlerContext, (FullHttpRequest) o);

        }
    }

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

    /**
     * 处理http请求
     * @param ctx
     * @param req
     * @throws Exception
     */
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) throws Exception {
        // 若是HTTP解码失败或者不是websocket请求，返回HTTP异常
        if (!req.getDecoderResult().isSuccess()
                || (!"websocket".equals(req.headers().get("Upgrade")))) {
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                    HttpResponseStatus.BAD_REQUEST));
            return;
        }

        // 构造握手响应返回，本机测试
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                "ws://localhost:8888/websocket", null, false);

        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

    /**
     * 处理websocket请求
     * @param ctx
     * @param frame
     */
    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 data = ((TextWebSocketFrame) frame).text();
        System.out.println("服务器接收数据=" + data);

        ctx.channel().write(new TextWebSocketFrame(data + " , welocme to："
                + new java.util.Date().toString()));
    }

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

   /** 
    * 返回应答给客户端
    * @param ctx
    * @param req
    * @param res
    */
    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();
            HttpHeaders.setContentLength(res, res.content().readableBytes());
        }

        // 若是是非Keep-Alive，关闭链接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!HttpHeaders.isKeepAlive(req) || res.getStatus().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }
}

客户端代码：

<!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:8888/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>