netty websocket协议开发

websocket的好处我们就不用多说了，就是用于解决长连接、服务推送等需要的一种技术。

以下我们来看一个例子:

package com.ming.netty.http.websocket;

import java.net.InetSocketAddress;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
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.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;

public class WebSocketServer {

    
    
    public static void main(String[] args) {
        new WebSocketServer().run("127.0.0.1", 8500);
    }
    
    
    public void run(String addr,int port){
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
             ServerBootstrap b=new ServerBootstrap();
              b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 128)
             .childOption(ChannelOption.SO_KEEPALIVE, true)
             .childHandler(new WebSocketServerHandlerInitializer()); 
              
              ChannelFuture f=b.bind(new InetSocketAddress(addr, port)).sync();
              System.out.println("启动服务器:"+f.channel().localAddress());
              //等等服务器端监听端口关闭
              f.channel().closeFuture().sync();
              
        } catch (Exception e) {
            e.printStackTrace();
        }finally{
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    protected class WebSocketServerHandlerInitializer extends ChannelInitializer<SocketChannel>{
        
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline(); 
            pipeline.addLast(new HttpServerCodec());
            pipeline.addLast(new HttpObjectAggregator(64*1024));
            pipeline.addLast(new ChunkedWriteHandler());
            pipeline.addLast(new HttpRequestHandler("/ws"));
            pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));
            pipeline.addLast(new TextWebSocketFrameHandler());
        }
        
    }
}

package com.ming.netty.http.websocket;

import io.netty.channel.Channel;
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.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.URISyntaxException;
import java.net.URL;

public class HttpRequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final String wsUri;
    private static final File INDEX;
 
    // static HTTP request handling operation.
    static {
        URL location = HttpRequestHandler.class.getProtectionDomain().getCodeSource().getLocation();
        try {
           String path = location.toURI() + "WebSocketClient.html";
           path = !path.contains("file:") ? path : path.substring(5);
           INDEX = new File("D:\javaPro\workspace\NettyTest\resources\WebsocketChatClient.html");
         } catch (URISyntaxException e) {
           throw new IllegalStateException("Unable to locate WebsocketChatClient.html", e);
        }
  }
 

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

 
 
 
 @Override
protected void messageReceived(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
     channelRead(ctx,request);
    
}




public void channelRead(ChannelHandlerContext ctx, FullHttpRequest request)
   throws Exception {
  if (wsUri.equalsIgnoreCase(request.getUri())) {
   ctx.fireChannelRead(request.retain());
  } else {
   if (HttpHeaders.is100ContinueExpected(request)) {
    send100Continue(ctx);
   }
   
   RandomAccessFile file = new RandomAccessFile(INDEX, "r");
   HttpResponse response = new DefaultHttpResponse(
     request.getProtocolVersion(), HttpResponseStatus.OK);
     response.headers().set(HttpHeaders.Names.CONTENT_TYPE,
     "text/html; charset=UTF-8");

   boolean keepAlive = HttpHeaders.isKeepAlive(request);
   if (keepAlive) {
    response.headers().set(HttpHeaders.Names.CONTENT_LENGTH,  file.length());
    response.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE);
   }
   ctx.write(response);

   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) {
    future.addListener(ChannelFutureListener.CLOSE);
   }

   file.close();
  }
 }

 private static void send100Continue(ChannelHandlerContext ctx) {
  FullHttpResponse response = new DefaultFullHttpResponse(
    HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE);
  ctx.writeAndFlush(response);
 }

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
   throws Exception {
  Channel incoming = ctx.channel();
  System.out.println("Client:" + incoming.remoteAddress() + "异常");
  
  // 当出现异常就关闭连接
  cause.printStackTrace();
  ctx.close();
 }

}

package com.ming.netty.http.websocket;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;

public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>  {

    public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
          Channel incoming = ctx.channel();
          System.out.println("Client:" + incoming.remoteAddress() + "在线");
    }



    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        System.out.println("Client:" + incoming.remoteAddress() + "掉线");
    }


    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
          Channel incoming = ctx.channel();
          for (Channel channel : channels) {
           channel.writeAndFlush(new TextWebSocketFrame("[SERVER] - "  + incoming.remoteAddress() + " 加入"));
          }
          channels.add(ctx.channel());
          System.out.println("Client:" + incoming.remoteAddress() + "加入");
    }



    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        for (Channel channel : channels) {
          channel.writeAndFlush(new TextWebSocketFrame("[SERVER] - "  + incoming.remoteAddress() + " 离开"));
        }
        System.out.println("Client:" + incoming.remoteAddress() + "离开");
        channels.remove(ctx.channel());
    }

    
    
    


    @Override
    protected void messageReceived(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        Channel incoming = ctx.channel();
          for (Channel channel : channels) {
           if (channel != incoming) {
            //channel.writeAndFlush(new TextWebSocketFrame("[" + incoming.remoteAddress() + "]" + msg.text()));
           } else {
            //返送给指定的
            channel.writeAndFlush(new TextWebSocketFrame("[服务器端返回]：" + msg.text()));
            
        
            //output current message to context. 
            StringBuffer sb = new StringBuffer();
            sb.append(incoming.remoteAddress()).append("->").append(msg.text());
            System.out.println(sb.toString());
           }
          }
        
    }



    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        Channel incoming = ctx.channel();
        System.out.println("Client:" + incoming.remoteAddress() + "异常");
        // 当出现异常就关闭连接
        cause.printStackTrace();
        ctx.close();
    }



    

    

}

<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<title>WebSocket Chat</title>
</head>
<body>
    <script type="text/javascript">
        var socket;
        if (!window.WebSocket) {
            window.WebSocket = window.MozWebSocket;
        }
        if (window.WebSocket) {
            socket = new WebSocket("ws://127.0.0.1:8500/ws");
            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 = "连接开启!";
            };
            socket.onclose = function(event) {
                var ta = document.getElementById('responseText');
                ta.value = ta.value + "连接被关闭";
            };
        } else {
            alert("你的浏览器不支持 WebSocket！");
        }

        function send(message) {
            if (!window.WebSocket) {
                return;
            }
            if (socket.readyState == WebSocket.OPEN) {
                socket.send(message);
            } else {
                alert("连接没有开启.");
            }
        }
    </script>
    <form onsubmit="return false;">
        <h3>WebSocket 聊天室：</h3>
        <textarea id="responseText" style=" 500px; height: 300px;"></textarea>
        <br> 
        <input type="text" name="message"  style=" 300px" value="Welcome to www.waylau.com">
        <input type="button" value="发送消息" onclick="send(this.form.message.value)">
        <input type="button" onclick="javascript:document.getElementById('responseText').value=''" value="清空聊天记录">
    </form>
    <br> 
    <br> 
    
</body>
</html>

运行服务器，然后在浏览器输入：127.0.0.1:8500 就可以看见一个简单的聊天室效果了.