Netty5 + WebSocket 练习

朴灿烈づ我的快乐病毒、 2022-06-12 02:41 254阅读 0赞

# [Netty5 + WebSocket 练习][Netty5 _ WebSocket] #

1. 了解WebSocket知识  
　　略  
2. websocket实现系统简单反馈时间

WebSocketServerHandler.java

![复制代码][copycode.gif]

1 package com.jieli.nettytest.websocketserver;
     2 
     3 import io.netty.bootstrap.ServerBootstrap;
     4 import io.netty.channel.ChannelFuture;
     5 import io.netty.channel.ChannelInitializer;
     6 import io.netty.channel.EventLoopGroup;
     7 import io.netty.channel.nio.NioEventLoopGroup;
     8 import io.netty.channel.socket.SocketChannel;
     9 import io.netty.channel.socket.nio.NioServerSocketChannel;
    10 import io.netty.handler.codec.http.HttpObjectAggregator;
    11 import io.netty.handler.codec.http.HttpServerCodec;
    12 import io.netty.handler.stream.ChunkedWriteHandler;
    13 
    14 public class WebSocketServer {
    15     
    16     public void run(int port){
    17         EventLoopGroup bossGroup = new NioEventLoopGroup();
    18         EventLoopGroup workerGroup = new NioEventLoopGroup();
    19         try {
    20             ServerBootstrap b = new ServerBootstrap();
    21             b.group(bossGroup, workerGroup)
    22              .channel(NioServerSocketChannel.class)
    23              .childHandler(new ChannelInitializer<SocketChannel>() {
    24                 @Override
    25                 protected void initChannel(SocketChannel ch) throws Exception {
    26                     //HttpServerCodec将请求和应答消息编码或解码为HTTP消息
    27                     //通常接收到的http是一个片段，如果想要完整接受一次请求所有数据，我们需要绑定HttpObjectAggregator
    28                     //然后就可以收到一个FullHttpRequest完整的请求信息了
    29                     //ChunkedWriteHandler 向客户端发送HTML5文件，主要用于支持浏览器和服务器进行WebSocket通信
    30                     //WebSocketServerHandler自定义Handler
    31                     ch.pipeline().addLast("http-codec", new HttpServerCodec())
    32                                  .addLast("aggregator", new HttpObjectAggregator(65536)) //定义缓冲大小
    33                                  .addLast("http-chunked", new ChunkedWriteHandler())
    34                                  .addLast("handler", new WebSocketServerHandler());
    35                 }
    36             });
    37             
    38             ChannelFuture f = b.bind(port).sync();
    39             System.out.println("start...");
    40             f.channel().closeFuture().sync();
    41         } catch (Exception e) {
    42             e.printStackTrace();
    43         } finally {
    44             workerGroup.shutdownGracefully();
    45             bossGroup.shutdownGracefully();
    46         }
    47     }
    48     
    49     public static void main(String[] args) {
    50         new WebSocketServer().run(7777);
    51     }
    52 }

![复制代码][copycode.gif]

WebSocketServerHandler.java

![复制代码][copycode.gif]

1 package com.jieli.nettytest.websocketserver;
      2 
      3 import java.util.logging.Level;
      4 import java.util.logging.Logger;
      5 
      6 import io.netty.buffer.ByteBuf;
      7 import io.netty.buffer.Unpooled;
      8 import io.netty.channel.ChannelFuture;
      9 import io.netty.channel.ChannelFutureListener;
     10 import io.netty.channel.ChannelHandlerContext;
     11 import io.netty.channel.SimpleChannelInboundHandler;
     12 import io.netty.handler.codec.http.DefaultFullHttpResponse;
     13 import io.netty.handler.codec.http.FullHttpRequest;
     14 import io.netty.handler.codec.http.FullHttpResponse;
     15 import io.netty.handler.codec.http.HttpHeaderUtil;
     16 import io.netty.handler.codec.http.HttpResponseStatus;
     17 import io.netty.handler.codec.http.HttpVersion;
     18 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
     19 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
     20 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
     21 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
     22 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
     23 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
     24 import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
     25 import io.netty.util.CharsetUtil;
     26 
     27 public class WebSocketServerHandler extends SimpleChannelInboundHandler<Object>{
     28 
     29     /**
     30      * 日志
     31      */
     32     private static final Logger logger = 
     33             Logger.getLogger(WebSocketServerHandler.class.getName());
     34     /**
     35      * 全局websocket
     36      */
     37     private WebSocketServerHandshaker handshaker;
     38     
     39     @Override
     40     protected void messageReceived(ChannelHandlerContext ctx, Object msg)
     41             throws Exception {
     42         //普通HTTP接入
     43         if(msg instanceof FullHttpRequest){
     44             handleHttpRequest(ctx, (FullHttpRequest) msg);
     45         }else if(msg instanceof WebSocketFrame){ //websocket帧类型 已连接
     46             //BinaryWebSocketFrame CloseWebSocketFrame ContinuationWebSocketFrame 
     47             //PingWebSocketFrame PongWebSocketFrame TextWebScoketFrame
     48             handleWebSocketFrame(ctx, (WebSocketFrame) msg);
     49         }
     50     }
     51     
     52     @Override
     53     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
     54         ctx.flush();
     55     }
     56     
     57     private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request){
     58         //如果http解码失败 则返回http异常 并且判断消息头有没有包含Upgrade字段(协议升级)
     59         if(!request.decoderResult().isSuccess() 
     60                 || (!"websocket".equals( request.headers().get("Upgrade")))    ){
     61             sendHttpResponse(ctx, request, new DefaultFullHttpResponse(
     62                     HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
     63             return ;
     64         }
     65         //构造握手响应返回
     66         WebSocketServerHandshakerFactory ws = new WebSocketServerHandshakerFactory("", null, false);
     67         handshaker = ws.newHandshaker(request);
     68         if(handshaker == null){
     69             //版本不支持
     70             WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
     71         }else{
     72             handshaker.handshake(ctx.channel(), request);
     73         }
     74     }
     75     /**
     76      * websocket帧
     77      * @param ctx
     78      * @param frame
     79      */
     80     private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame){
     81         //判断是否关闭链路指令
     82         if(frame instanceof CloseWebSocketFrame){
     83             handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
     84             return ;
     85         }
     86         //判断是否Ping消息 -- ping/pong心跳包
     87         if(frame instanceof PingWebSocketFrame){
     88             ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
     89             return ;
     90         }
     91         //本程序仅支持文本消息， 不支持二进制消息
     92         if(!(frame instanceof TextWebSocketFrame)){
     93             throw new UnsupportedOperationException(
     94                     String.format("%s frame types not supported", frame.getClass().getName()));
     95         }
     96         
     97         //返回应答消息 text文本帧
     98         String request = ((TextWebSocketFrame) frame).text();
     99         //打印日志
    100         if(logger.isLoggable(Level.FINE)){
    101             logger.fine(String.format("%s received %s", ctx.channel(), request));
    102         }
    103         //发送到客户端websocket
    104         ctx.channel().write(new TextWebSocketFrame(request 
    105                 + ", 欢迎使用Netty WebSocket服务， 现在时刻:" 
    106                 + new java.util.Date().toString()));
    107     }
    108     
    109     /**
    110      * response
    111      * @param ctx
    112      * @param request
    113      * @param response
    114      */
    115     private static void sendHttpResponse(ChannelHandlerContext ctx, 
    116             FullHttpRequest request, FullHttpResponse response){
    117         //返回给客户端
    118         if(response.status().code() != HttpResponseStatus.OK.code()){
    119             ByteBuf buf = Unpooled.copiedBuffer(response.status().toString(), CharsetUtil.UTF_8);
    120             response.content().writeBytes(buf);
    121             buf.release();
    122             HttpHeaderUtil.setContentLength(response, response.content().readableBytes());
    123         }
    124         //如果不是keepalive那么就关闭连接
    125         ChannelFuture f = ctx.channel().writeAndFlush(response);
    126         if(!HttpHeaderUtil.isKeepAlive(response) 
    127                 || response.status().code() != HttpResponseStatus.OK.code()){
    128             f.addListener(ChannelFutureListener.CLOSE);
    129         }
    130     }
    131     
    132     /**
    133      * 异常 出错
    134      */
    135     @Override
    136     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
    137             throws Exception {
    138         cause.printStackTrace();
    139         ctx.close();
    140     }
    141 }

![复制代码][copycode.gif]

WebSocketServer.html (这个随便放都可以，到时候双击打开这个即可，不由服务器提供)

![复制代码][copycode.gif]

1 <html>
     2     <head>
     3         <meta charset="utf-8">
     4         <title>Netty websocket 时间服务器</title>
     5     </head>
     6     <body>
     7         <form action="" onsubmit="return false;">
     8             <input type="text" name="message" value="..."/>
     9             <br>
    10             <input type="button" value="send" value="发送websocket请求消息" onclick="send(this.form.message.value);" /> 
    11             <hr color="blue">
    12             <h3>服务器返回信息</h3>
    13             <textarea id="responseText" rows="10" cols=""></textarea>
    14         </form>
    15     </body>
    16     
    17     <script type="text/javascript">
    18         var socket;
    19         if(!window.WebSocket){
    20             window.WebSocket = window.MozWebSocket;
    21         }
    22         if(window.WebSocket){
    23             socket = new WebSocket("ws://localhost:7777/websocket");
    24             socket.onmessage = function(event){
    25                 var ta = document.getElementById('responseText');
    26                 ta.value="";
    27                 ta.value=event.data;
    28             };
    29             socket.onopen = function(event){
    30                 var ta = document.getElementById('responseText');
    31                 ta.value = "打开websocket服务正常";
    32             }
    33             socket.onclose = function(event){
    34                 var ta = document.getElementById('responseText');
    35                 ta.value="";
    36                 ta.value="websocket关闭";
    37             }
    38         }else{
    39             alert("对不起，您的浏览器不支持WebSocket.");
    40         }
    41         
    42         function send(message){
    43             if(!window.WebSocket){
    44                 return ;
    45             }
    46             if(socket.readyState == WebSocket.OPEN){
    47                 socket.send(message);
    48             }else{
    49                 alert("WebSocket 连接创建失败.");
    50             }
    51         }
    52     </script>
    53 </html>

![复制代码][copycode.gif]

运行结果

![485067-20160303203954737-818138828.png][]

![485067-20160303204005112-365447270.png][]

3. websocket实现简单聊天室

WebSocketChatServer.java

![复制代码][copycode.gif]

1 package com.jieli.nettytest.websocket;
     2 
     3 import io.netty.bootstrap.ServerBootstrap;
     4 import io.netty.channel.ChannelFuture;
     5 import io.netty.channel.ChannelOption;
     6 import io.netty.channel.EventLoopGroup;
     7 import io.netty.channel.nio.NioEventLoopGroup;
     8 import io.netty.channel.socket.nio.NioServerSocketChannel;
     9 
    10 public class WebsocketChatServer {
    11     
    12     private int port;
    13     
    14     public WebsocketChatServer(int port){
    15         this.port = port;
    16     }
    17     
    18     public void run() throws Exception{
    19         EventLoopGroup bossGroup = new NioEventLoopGroup();
    20         EventLoopGroup workerGroup = new NioEventLoopGroup();
    21         try {
    22             ServerBootstrap b = new ServerBootstrap();
    23             b.group(bossGroup, workerGroup)
    24              .channel(NioServerSocketChannel.class)
    25              .childHandler(new WebsocketChatServerInitializer())
    26              .option(ChannelOption.SO_BACKLOG, 128)
    27              .childOption(ChannelOption.SO_KEEPALIVE, true);
    28             
    29             System.out.println("websocket start..");
    30 
    31             ChannelFuture f = b.bind(port).sync();
    32             
    33             f.channel().closeFuture().sync();
    34         } catch (Exception e) {
    35             e.printStackTrace();
    36         } finally {
    37             workerGroup.shutdownGracefully();
    38             bossGroup.shutdownGracefully();
    39             System.out.println("websocket close.");
    40         }
    41     }
    42     
    43     public static void main(String[] args) throws Exception{
    44         new WebsocketChatServer(8080).run();
    45     }
    46 }

![复制代码][copycode.gif]

WebsocketChatServerInitializer.java

![复制代码][copycode.gif]

1 package com.jieli.nettytest.websocket;
     2 
     3 import io.netty.channel.ChannelInitializer;
     4 import io.netty.channel.ChannelPipeline;
     5 import io.netty.channel.socket.SocketChannel;
     6 import io.netty.handler.codec.http.HttpObjectAggregator;
     7 import io.netty.handler.codec.http.HttpServerCodec;
     8 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
     9 import io.netty.handler.stream.ChunkedWriteHandler;
    10 
    11 public class WebsocketChatServerInitializer extends ChannelInitializer<SocketChannel>{
    12 
    13     @Override
    14     protected void initChannel(SocketChannel ch) throws Exception {
    15         ChannelPipeline pipeline = ch.pipeline();
    16         pipeline.addLast(new HttpServerCodec())
    17                 .addLast(new HttpObjectAggregator(64*1024))
    18                 .addLast(new ChunkedWriteHandler())
    19                 .addLast(new HttpRequestHandler("/ws")) //如果访问的是RUI"/ws"，处理WebSocket升级
    20                 .addLast(new WebSocketServerProtocolHandler("/ws"))
    21                 .addLast(new TextWebSocketFrameHandler());
    22     }
    23     
    24 }

![复制代码][copycode.gif]

HttpRequestHandler.java

![ContractedBlock.gif][]

html/index.html

![ContractedBlock.gif][]

运行结果，依次运行1，2，3

![485067-20160304093535830-2059786334.png][]

服务器打印结果

![485067-20160304093607768-640669622.png][]

[Netty5 _ WebSocket]: http://www.cnblogs.com/wunaozai/p/5240006.html
[copycode.gif]: /images/20220612/238ab857da194206bde2907c426801bf.png
[485067-20160303203954737-818138828.png]: /images/20220612/045c3fe94fdf42628c6d4c310ff6f82d.png
[485067-20160303204005112-365447270.png]: /images/20220612/22bd088a79a44c9d8680cdc06b6c7c31.png
[ContractedBlock.gif]: /images/20220612/6eb7a80c9e954bf19606400f12d207ec.png
[485067-20160304093535830-2059786334.png]: /images/20220612/bae394534248410eb751bad3788f966b.png
[485067-20160304093607768-640669622.png]: /images/20220612/52bfa230b6b54b08b6a68c04f022a11e.png