netty实现WebSocket协议

╰半橙微兮° 2023-07-04 12:49 62阅读 0赞

### 文章目录 ###

*  WebSocket协议
 *  服务端开发
 *  客户端
 *  运行
 *  测试全双工--netty服务端主动推送数据
 *   *  定时器推送消息
     *  维护channel关系
     *  运行

> **注：更多netty相关文章请访问博主专栏： [netty专栏][netty]**

# WebSocket协议 #

一般web应用都是使用的HTTP协议。HTTP协议有以下特点：

*  支持`客户端-服务端`模式
 *  使用简单：只需要知道服务端URL，携带参数发送请求即可
 *  支持多种传输数据类型，由消息头中content-type标识
 *  无状态，使得HTTP服务轻量级

HTTP协议也存在一些缺点：

*  半双工通信：同一时刻，数据只能往同一方向传输。比如向服务器发送消息时，服务器此时不可以向客户端发送消息。（不过目前HTTP2已经支持了全双工通信）
 *  HTTP消息冗长：HTTP消息包含消息头，消息体，换行符等。且大多采用文本传输。所以HTTP消息会有很多冗余消息并且消息占用字节数大，消耗过多的带宽
 *  针对服务器推送的黑客攻击，如长时间轮询

因此在一些场景使用WebSocket来实现更加妥当，比如需要全双工通信的需求（聊天室、消息推送）。WebSocket有以下特点：

*  底层采用单一TCP链接，全双工通信
 *  对代理、防火墙、路由器透明
 *  无头部消息，cookie，身份验证
 *  无安全开销
 *  通过ping/pong消息保持心跳

WebSocket稳定性很高，它实质是一个TCP链接，且支持全双工。WebSocket是当下必学知识点。

# 服务端开发 #

package com.example;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    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.codec.http.websocketx.*;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    import io.netty.handler.stream.ChunkedFile;
    import io.netty.handler.stream.ChunkedWriteHandler;
    import io.netty.util.CharsetUtil;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    import javax.activation.MimetypesFileTypeMap;
    import java.io.File;
    import java.io.FileNotFoundException;
    import java.io.RandomAccessFile;
    import java.io.UnsupportedEncodingException;
    import java.net.InetSocketAddress;
    import java.net.URLDecoder;
    import java.util.Date;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentMap;
    import java.util.regex.Pattern;
    
    /** * netty http 文件下载 服务器 */
    public class MyWebSocketServer { 
    
        int port;
    
        public MyWebSocketServer(int port) { 
            this.port = port;
        }
    
        public void start() { 
            ServerBootstrap bootstrap = new ServerBootstrap();
            EventLoopGroup boss = new NioEventLoopGroup();
            EventLoopGroup work = new NioEventLoopGroup();
            try { 
                bootstrap.group(boss, work)
                        .handler(new LoggingHandler(LogLevel.DEBUG))
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new WebSocketServerInitializer());
    
                ChannelFuture f = bootstrap.bind(new InetSocketAddress(port)).sync();
                System.out.println("http server started . port : " + port);
                f.channel().closeFuture().sync();
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
                boss.shutdownGracefully();
                work.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) { 
            MyWebSocketServer server = new MyWebSocketServer(8080);// 8081为启动端口
            server.start();
        }
    }
    
    class WebSocketServerInitializer extends ChannelInitializer<SocketChannel> { 
    
        @Override
        protected void initChannel(SocketChannel channel) { 
            ChannelPipeline pipeline = channel.pipeline();
            pipeline.addLast(new HttpServerCodec())// http 编解器
                    // http 消息聚合器 512*1024为接收的最大contentlength
                    .addLast("httpAggregator", new HttpObjectAggregator(512 * 1024))
                    // 支持异步发送大的码流(大的文件传输),但不占用过多的内存，防止java内存溢出
                    .addLast("http-chunked", new ChunkedWriteHandler())
                    .addLast(new WebSocketRequestHandler());// 请求处理器
        }
    }
    
    class WebSocketRequestHandler extends SimpleChannelInboundHandler<Object> { 
    
        private WebSocketServerHandshaker handshaker;
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { 
            System.out.println("收到消息：" + msg);
            if (msg instanceof FullHttpRequest) { 
                //以http请求形式接入，但是走的是websocket
                handleHttpRequest(ctx, (FullHttpRequest) msg);
            } else if (msg instanceof WebSocketFrame) { 
                //处理websocket客户端的消息
                handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
            }
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
            ctx.flush();
        }
    
        /* 对WebSocket请求进行处理 */
        private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) { 
            // 判断是否关闭链路的指令
            if (frame instanceof CloseWebSocketFrame) { 
                handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
                return;
            }
            // 判断是否ping消息，如果是，则构造pong消息返回。用于心跳检测
            if (frame instanceof PingWebSocketFrame) { 
                ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
                return;
            }
    
            // 本例程仅支持文本消息，不支持二进制消息
            if (!(frame instanceof TextWebSocketFrame)) { 
                System.out.println("本例程仅支持文本消息，不支持二进制消息");
                throw new UnsupportedOperationException(
                        String.format("%s frame types not supported", frame.getClass().getName()));
            }
    
            //处理客户端请求并返回应答消息
            String request = ((TextWebSocketFrame) frame).text();
            System.out.println("服务端收到：" + request);
            
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("time",format.format(new Date()));
            jsonObject.put("channelId",ctx.channel().id().asShortText());
            jsonObject.put("request",request);
    
            TextWebSocketFrame tws = new TextWebSocketFrame(jsonObject.toJSONString());
    
            // 返回【谁发的发给谁】
             ctx.channel().writeAndFlush(tws);
        }
    
        /** * 唯一的一次http请求。 * 该方法用于处理websocket握手请求 */
        private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { 
            //如果HTTP解码失败，返回异常。要求Upgrade为websocket，过滤掉get/Post
            if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) { 
                //若不是websocket方式，则创建BAD_REQUEST（400）的req，返回给客户端
                sendHttpResponse(ctx, req, new DefaultFullHttpResponse(
                        HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
                return;
            }
    
    // 构造握手响应返回，本机测试
            WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
                    "ws://localhost:8080/websocket", null, false);
            //通过工厂来创建WebSocketServerHandshaker实例
            handshaker = wsFactory.newHandshaker(req);
            if (handshaker == null) { 
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else { 
                /* 通过WebSocketServerHandshaker来构建握手响应消息返回给客户端。 同时将WebSocket相关编解码类添加到ChannelPipeline中，该功能需要阅读handshake的源码。 */
                handshaker.handshake(ctx.channel(), req);
            }
        }
    
        /** * 拒绝不合法的请求，并返回错误信息 */
        private static void sendHttpResponse(ChannelHandlerContext ctx,
                                             FullHttpRequest req, DefaultFullHttpResponse res) { 
            // 返回应答给客户端
            if (res.status().code() != 200) { 
                ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
                res.content().writeBytes(buf);
                buf.release();
                HttpUtil.setContentLength(res, res.content().readableBytes());
            }
    
            ChannelFuture f = ctx.channel().writeAndFlush(res);
    
            // 如果是非Keep-Alive，关闭连接
            if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) { 
                f.addListener(ChannelFutureListener.CLOSE);
            }
        }
    }

注释都已经写好，这里需要注意的几点是：

*  netty的websocket协议是在HTTP协议基础之上完成的，要使用WebSocket协议，需要将HTTP请求头中添加`Upgrade:WebSocket`
 *  WebSocket相关的编解码是在`handshaker.handshake(ctx.channel(), req);`中添加进去的。handshaker是WebSocketServerHandshaker的对象。handshake方法中建握手响应消息返回给客户端。同时将WebSocket相关编解码类添加到ChannelPipeline中。

handshake源码：

public final ChannelFuture handshake(Channel channel, FullHttpRequest req,
                                        HttpHeaders responseHeaders, final ChannelPromise promise) { 
    
        if (logger.isDebugEnabled()) { 
            logger.debug("{} WebSocket version {} server handshake", channel, version());
        }
        //构造握手响应
        FullHttpResponse response = newHandshakeResponse(req, responseHeaders);
    //下面将channelpipeline中的HttpObjectAggregator和HttpContentCompressor移除，并且添加WebSocket编解码器newWebSocketEncoder和newWebsocketDecoder
        ChannelPipeline p = channel.pipeline();
        if (p.get(HttpObjectAggregator.class) != null) { 
            p.remove(HttpObjectAggregator.class);
        }
        if (p.get(HttpContentCompressor.class) != null) { 
            p.remove(HttpContentCompressor.class);
        }
        ChannelHandlerContext ctx = p.context(HttpRequestDecoder.class);
        final String encoderName;
        if (ctx == null) { 
            // this means the user use an HttpServerCodec
            ctx = p.context(HttpServerCodec.class);
            if (ctx == null) { 
                promise.setFailure(
                        new IllegalStateException("No HttpDecoder and no HttpServerCodec in the pipeline"));
                return promise;
            }
            p.addBefore(ctx.name(), "wsencoder", newWebSocketEncoder());
            p.addBefore(ctx.name(), "wsdecoder", newWebsocketDecoder());
            encoderName = ctx.name();
        } else { 
            p.replace(ctx.name(), "wsdecoder", newWebsocketDecoder());
    
            encoderName = p.context(HttpResponseEncoder.class).name();
            p.addBefore(encoderName, "wsencoder", newWebSocketEncoder());
        }
    //将response消息返回给客户端
        channel.writeAndFlush(response).addListener(new ChannelFutureListener() { 
            @Override
            public void operationComplete(ChannelFuture future) throws Exception { 
                if (future.isSuccess()) { 
                    ChannelPipeline p = future.channel().pipeline();
                    p.remove(encoderName);
                    promise.setSuccess();
                } else { 
                    promise.setFailure(future.cause());
                }
            }
        });
        return promise;
    }

# 客户端 #

客户端采用HTML实现，支持WebSocket的浏览器都可以，如果不支持会提示`ERROR:您的浏览器不支持WebSocket！！`。  
客户端的功能是想服务器发送请求消息，服务器返回服务器时间以及channelID。

<!DOCTYPE html>
    <html>
    <head>
        <meta charset="UTF-8">
        <title>my websocket client</title>
    </head>
    <body>
    <textarea id="msgBoxs"></textarea><br>
    待发送消息`：<input type="text" id="msg">
    <input type="button" id="sendBtn" onclick="send()" value="发送">
    <script type="application/javascript"> var socket ; if(!window.WebSocket){ window.WebSocket = window.MozWebSocket; } if(window.WebSocket){ var msgBoxs = document.getElementById("msgBoxs") var msgBox = document.getElementById("msg") socket = new WebSocket("ws://localhost:8080/websocket") socket.onopen = function (evt) { console.log("Connection open ..."); socket.send("Hello WebSocket!"); } socket.onmessage = function (evt) { console.log("Received Message: ", evt.data) msgBoxs.value = msgBoxs.value + "\n" + evt.data } socket.onclose = function (evt) { console.log("Connect closed."); } }else{ alert("ERROR:您的浏览器不支持WebSocket！！"); } function send() { var msg = msgBox.value socket.send(msg) //msgBox.value = "" } </script>
    </body>
    </html>

# 运行 #

启动服务端。然后用浏览器打开上面的HTML。如果你的浏览器支持WebSocket，那么会出现以下界面：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]  
上图中将请求与响应的关键点标识出来了：`Upgrade: websocket`。  
再看一下服务端输出：

http server started . port : 8080
    收到消息：HttpObjectAggregator$AggregatedFullHttpRequest(decodeResult: success, version: HTTP/1.1, content: CompositeByteBuf(ridx: 0, widx: 0, cap: 0, components=0))
    GET /websocket HTTP/1.1
    Host: localhost:8080
    Connection: Upgrade
    Pragma: no-cache
    Cache-Control: no-cache
    User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.130 Safari/537.36
    Upgrade: websocket
    Origin: file://
    Sec-WebSocket-Version: 13
    Accept-Encoding: gzip, deflate, br
    Accept-Language: zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7
    Sec-WebSocket-Key: hoOY0UnvE6Hcf3QIl3nP+A==
    Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
    content-length: 0
    收到消息：TextWebSocketFrame(data: PooledUnsafeDirectByteBuf(ridx: 0, widx: 16, cap: 16))
    服务端收到：Hello WebSocket!
    收到消息：TextWebSocketFrame(data: PooledUnsafeDirectByteBuf(ridx: 0, widx: 3, cap: 3))
    服务端收到：111

# 测试全双工–netty服务端主动推送数据 #

上面的实现还是请求应答模式。接下来让服务器每一秒主动推送一次服务器时间。

这种应用场景非常多，比如手机APP的消息推送，或者聊天室群发消息等。

## 定时器推送消息 ##

这里定时推送采用的是定时器来操作的。在main方法中添加了定时方法：

public void pushMsg(){ 
            //模拟异步发送推送消息
            ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);
            scheduledThreadPool.scheduleWithFixedDelay(() -> { 
                TextWebSocketFrame tws = new TextWebSocketFrame("服务器主动推送消息。当前服务器时间："+System.currentTimeMillis());
                // 群发
                ChannelSupervise.send2All(tws);
            }, 0,1, TimeUnit.SECONDS);
    
        }
        
        public static void main(String[] args) { 
            MyWebSocketServer server = new MyWebSocketServer(8080);// 8081为启动端口
            
            server.pushMsg();
            
            server.start();
        }

## 维护channel关系 ##

这一步是关键点，服务器端需要维护创建连接的channel的集合，保存客户端链接，用于消息的推送。  
当有客户端连接时候会被SimpleChannelInboundHandler中的`channelActive`监听到，当断开时会被`channelInactive`监听到，一般在这两个方法中去保存/移除客户端的通道信息。  
channel信息保存在`ChannelGroup`中，ChannelGroup是netty提供用于管理web于服务器建立的通道channel的，其本质是一个高度封装的set集合，在服务器广播消息时，可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去。  
但在查找某一个客户端的通道时候比较坑爹，必须通过`channelId`对象去查找，而channelId不能人为创建，所有必须通过map将channelId的字符串和channel保存起来。

代码修改如下：

//添加ChannelSupervise类用于保存客户端channel信息
    class ChannelSupervise { 
        /** * ChannelGroup是netty提供用于管理web于服务器建立的通道channel的， * 其本质是一个高度封装的set集合，在服务器广播消息时， * 可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去 */
        private static ChannelGroup GlobalGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        /** * ChannelMap维护的是channelID和channel的对应关系，用于向指定channel发送消息 */
        private static ConcurrentMap<String, ChannelId> ChannelMap = new ConcurrentHashMap<>();
    
        public static void addChannel(Channel channel) { 
            GlobalGroup.add(channel);
            ChannelMap.put(channel.id().asShortText(), channel.id());
        }
    
        public static void removeChannel(Channel channel) { 
            GlobalGroup.remove(channel);
            ChannelMap.remove(channel.id().asShortText());
        }
    
        //找到某个channel来发送消息
        public static Channel findChannel(String id) { 
            return GlobalGroup.find(ChannelMap.get(id));
        }
    
        public static void send2All(TextWebSocketFrame tws) { 
            GlobalGroup.writeAndFlush(tws);
        }
    }
    
    //WebSocketRequestHandler对其进行修改，在监听到链接创建于销毁时，维护channel信息
    class WebSocketRequestHandler extends SimpleChannelInboundHandler<Object> { 
    
        //省略代码
    
    //添加下面channelActive和channelInactive方法
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            //添加连接
            System.out.println("客户端加入连接：" + ctx.channel());
            ChannelSupervise.addChannel(ctx.channel());
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
            //断开连接
            System.out.println("客户端断开连接：" + ctx.channel());
            ChannelSupervise.removeChannel(ctx.channel());
        }
        
    //代码省略

客户端HTML页面无需修改。

## 运行 ##

启动服务器端。然后重新刷新客户端页面，可以看到服务器端会多打印一行日志：

客户端加入连接：[id: 0x3670424c, L:/0:0:0:0:0:0:0:1:8080 - R:/0:0:0:0:0:0:0:1:62458]

页面上会每隔1秒收到服务器的推送消息，页面上依旧可以发送消息，然后服务器端返回值会输出到文本框中：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]

> **注：更多netty相关文章请访问博主专栏： [netty专栏][netty]**

[netty]: https://blog.csdn.net/fgyibupi/category_9705823.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200211114524973.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200213154102112.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZneWlidXBp,size_16,color_FFFFFF,t_70