Netty之WebSocket协议开发

一.WebSocket定义

        websocket是html5开始提供的一种浏览器与服务器间进行全双工通信的网络技术，websocket通信协议于2011年被IETF定为标准RFC6455，websocket API被W3C定位标准。

        在websocket API中，浏览器与服务器只需要做一个握手的动作，然后，浏览器和服务器之间就形成了一条快速通道，两者就可以直接互相传送数据了。websocket基于TCP双向全双工进行消息传递，在同一时刻，既可以发送消息，也可以接收消息，相比于HTTP的半双工协议，性能得到很大提升。

         websocket特点：

         单一的TCP连接，采用全双工模式通信；

         对代理，防火墙和路由器透明；

         无头部消息，cookie和身份认证；

         无安全开销；

         通过"ping/pong"帧保持链路激活

         服务器可以主动传递消息给客户端，不再需要客户端轮询。

二.WebSocket协议开发

2.1 服务启动类WebSocketServer实现

package websocket;

 

import server.TimeServer;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.Channel;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.EventLoopGroup;

importio.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

importio.netty.handler.codec.http.HttpObjectAggregator;

importio.netty.handler.codec.http.HttpServerCodec;

importio.netty.handler.stream.ChunkedWriteHandler;

 

public class WebSocketServer {

         publicvoid run(int port) throws Exception{

                   EventLoopGroupbossGroup=new NioEventLoopGroup();

                   EventLoopGroupworkerGroup=new NioEventLoopGroup();

                   try{

                            ServerBootstrapb=new ServerBootstrap();

                            b.group(bossGroup,workerGroup)

                            .channel(NioServerSocketChannel.class)

                            .childHanlder(newChannelInitializer<SocketChannel>() {

                                     @Override

                                     protectedvoid initChannel(SocketChannel ch) throws Exception{

                                               Channelpipelinepipeline=ch.pipeline();

                                     //添加HttpServerCode，将请求消息和应答消息编码或者解码为HTTP消息

                                               pipeline.addLast("http-codec",newHttpServerCodec());

                                               //增加HttpObjectAggregator，目的是将HTTP消息的多个部分组合成一条完整的HTTP消息

                                               pipeline.addLast("aggregator",newHttpObjectAggregator(65536));

                                               //  添加ChunkedWriteHandler,来向客户端发送HTML5文件

                                               //它主要用于支持浏览器和服务端进行WebSocket通信

                                               ch.pipeline().addLast("http-chunked",newChunkedWriteHandler());

                                               //增加WebSocket服务端handler

                                               pipeline.addLast("handler",newWebSocketServerHandler());

                                     }

                            });

                            Channelch=b.bind(port).sync().channel();

                            System.out.println("Websocket server started at port"+port+'.');

                            System.out.println("Openyour browser and navigate "+port+'/');

                            ch.closeFuture().sync();

                   }catch (Exception e) {

                            //TODO: handle exception

                   }finally{

                            bossGroup.shutdownGracefully();

                            workerGroup.shutdownGracefully();

                   }

         }

        

         publicstatic void main(String[] args) throws Exception{

                   intport=8080;

                   if(args!=null&&args.length>0){

                            try{

                                     port=Integer.valueOf(args[0]);

                            }catch (Exception e) {

                                     //TODO: handle exception

                                     e.printStackTrace();

                            }

                   }

                   newWebSocketServer().run(port);

         }

 

}

2.2 WebSocketHandler的实现

package websocket;

 

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.ChannelFuture;

importio.netty.channel.ChannelFutureListener;

importio.netty.channel.ChannelHandlerContext;

importio.netty.handler.codec.http.FullHttpRequest;

importio.netty.handler.codec.http.FullHttpResponse;

importio.netty.handler.codec.http.websocketx.CloseWebSocketFrame;

importio.netty.handler.codec.http.websocketx.PingWebSocketFrame;

importio.netty.handler.codec.http.websocketx.PongWebSocketFrame;

importio.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import io.netty.handler.codec.http.websocketx.WebSocketFrame;

importio.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;

import io.netty.util.CharsetUtil;

importio.netty.util.ResourceLeakDetector.Level;

 

importcom.sun.istack.internal.logging.Logger;

 

public class WebSocketServerHandler

   extends SimpleChannel InboundHandler<Object>{

          private static final Loggerlogger=Logger.getLogger(WebSocketServerHandler.class.getName());

          

          private WebSocketServerHandshaker handshaker;

          

          @Override

          public void messageReceived(ChannelHandlerContextctx,Object msg) throws Exception {

                    // 第一次握手请求消息由HTTP协议承载，所以它是一个HTTP消息，

                    // 执行handleHttpRequest方法来处理WebSocket握手请求

                    // 传统的HTTP接入

                    if(msg instanceof FullHttpRequest){

                             handleHttpRequest(ctx,(FullHttpRequest) msg);

                    }

                    // WebSocket接入

                    else if(msg instanceof WebSocketFrame){

                             handleWebSocketFrame(ctx,(WebSocketFrame)msg);

                    }

          }

          

          @Override

          public voidchannelReadComplete(ChannelHandlerContext ctx) throws Exception{

                    ctx.flush();

          }

    

          private void handleHttpRequest(ChannelHandlerContextctx,FullHttpRequest req) throws Exception{

                     // 对握手请求消息进行判断，如果消息头中没有包含Upgrade字段或者值不是WebSocket,则返回400响应

                    // 如果HTTP解码失败，返回HTTP异常

                    if(!req.getDecoderResult().isSuccess()||(!"websocket".equals(req.headers().get("Upgrade")))){

                             sendHttpResponse(ctx,req,newDefaultFullHttpResponse(HTTP_1_1,BAD_REQUEST));

                             return;

                    }

                    // 构造握手响应返回，本机测试

                    WebSocketServerHandshakeFactoty wsFactory =new

                                      WebSocketServerHandshakeFactoty("ws://localhost:8080/websocket",null,false);

                    if(handshaker==null){

                             WebSocketServerHandshakeFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());

                             

                    } else{

                             handshaker.handshake(ctx.channel(),req);

                    }

          }

          /*

           * 客户端通过文本框请求消息给服务端，websocketserverhandler接收到的是已经编码后

           * 的websocketframe消息

           */

          private voidhandleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame){

                  

                    /*

                     * 对websocket请求消息进行处理，首先需要对控制帧进行判断

                     *

                     * 关闭websocket连接，

                     */

                    // 判断是否是关闭链路的指令

                    // 如果是关闭链路的控制消息，就调用websocketserverhandlshaker的close方法

                    if(frame instanceof CloseWebSocketFrame){

                             handshaker.close(ctx.channel()),(CloseWebSocketFrame)frame.retain());

                             return;

                    }

                    // 判断是否是Ping消息

                    // 如果是维持链路的ping消息，则构造ping消息返回

                    if(frame instanceof PingWebSocketFrame){

                             ctx.channel().write(newPongWebSocketFrame(frame.content().retain()));

                             return;

                    }

                    /*

                     * 由于本例程的websocket通信双方使用的都是文本消息，所以对请求消息的类型进行判断

                     */

                    // 本例程仅支持文本消息，不支持二进制消息

                    if(!(frame instanceof TextWebSocketFrame)) {

                             throw new UnsupportedOperationException(String.format("%sframe types not supported",frame.getClass().getName());

                    }

                    /*

                     * 从textwebsocket中获取请求消息字符串，对它处理后通过构造新的textwebsocketframe

                     * 消息返回给客户端，

                     */

                    // 返回应答消息

                    String request=((TextWebSocketFrame)frame).text();

                    if(logger.isLoggable(Level.FINE)) {

                             Logger.fine(String.format("%s received%s",ctx.channel(),request));

                             

                    }

                    // 由于握手应答时动态增加了textwebsocketframe的编码类

                    ctx.channel().write(

                                      new TextWebSocketFrame(request+",欢迎使用Netty WebSocket服务，现在时刻:"+newjava.uti.Date.toString()));

                                      )

                  

          }

          private static voidsendHttpResponse(ChannelHandlerContext ctx,

                             FullHttpRequest req,FullHttpResponse res) {

                    // 返回应答给客户端

                    if(res.getStatus().code()!=200) {

                             ByteBufbuf=Unpooled.copiedBuffer(res.getSatus().toString(),CharsetUtil.UTF-8);

                             res.content().writeBytes(buf);

                             buf.release();

                             setContentLength(res,res.content().readableBytes());

                             

                             

                    }

                    // 如果是非Keep-Alive，关闭链接

                    ChannelFuturef=ctx.channel().waitAndFlush(res);

                    if(!isKeepAlive(req)||res.getStatus.code()!=200){

                             f.addListener(ChannelFutureListener.CLOSE);

                    }

          }

          

          @Override

          public voidexceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception{

                    cause.printStackTrace();

                    ctx.close();

          }

}