zoukankan      html  css  js  c++  java
  • 一款基于Netty开发的WebSocket服务器

    代码地址如下:
    http://www.demodashi.com/demo/13577.html

    一款基于Netty开发的WebSocket服务器

    这是一款基于Netty框架开发的服务端,通信协议为WebSocket。主要用于Java后台服务器向浏览器进行消息推送。

    需求

    对于一个Web项目而言,客户端一般均为各种各样的浏览器,如何从后端服务器向浏览器客户端进行消息推送,便成了一个棘手的问题,好在在HTTP1.1之后,HTTP可以支持长连接,由此,我在Netty框架的基础上开发了这个WebSocket服务端。

    当然,你依旧可以下载源码进行测试,集成,二次开发等等。

    环境

    • Intellij IDEA2018
    • JDK 1.8
    • 插件:Simple WebSocket Client 0.1.3
      • FireFox Quantum 60.0.1 (64 位)
      • Google Chrome 67.0.3396.99(正式版本)(64 位)

    运行结果

    请看下图:

    推送结果图

    实现步骤及源码

    WebSocket

    WebSocket是HTML5开始提供的一种浏览器与服务器间进行全双工通讯的网络技术。 WebSocket通信协议于2011年被IETF定为标准RFC 6455,WebSocketAPI被W3C定为标准。 在WebSocket API中,浏览器和服务器只需要要做一个握手的动作,然后,浏览器和服务器之间就形成了一条快速通道。两者之间就直接可以数据互相传送。

    目录结构

    目录结构

    核心类讲解

    • WebSocketChildHandler
    • WebSocketServerHandler

    WebSocketChildHandler

    这个类的主要作用就是为Netty的通道注册事件。其核心代码见下,其中webSocketUrl就是客户端与服务端进行连接的请求路径,我将其写入了配置文件,交由Spring管理,以注入的方式传递到WebSocketServerHandler中。

    ChannelPipeline pipeline = socketChannel.pipeline();
            // 将请求与应答消息编码或者解码为HTTP消息
            pipeline.addLast("http-codec", new HttpServerCodec());
            // 将http消息的多个部分组合成一条完整的HTTP消息
            pipeline.addLast("aggregator", new HttpObjectAggregator(HttpObjectConstant.MAX_CONTENT_LENGTH));
            // 向客户端发送HTML5文件。主要用于支持浏览器和服务端进行WebSocket通信
            pipeline.addLast("http-chunked", new ChunkedWriteHandler());
            // 服务端Handler
            pipeline.addLast("handler", new WebSocketServerHandler(webSocketUrl));
    

    WebSocketServerHandler

    这个类是真正的核心类,这个类的主要功能为:

    • 进行第一次握手
    • 对消息进行处理
      • 可以实现点对点通信
      • 可以实现广播功能
      • 可以实现点对端通信
    /**
         * 接收客户端发送的消息
         *
         * @param channelHandlerContext ChannelHandlerContext
         * @param receiveMessage        消息
         */
        @Override
        protected void messageReceived(ChannelHandlerContext channelHandlerContext, Object receiveMessage) throws Exception {
            // 传统http接入 第一次需要使用http建立握手
            if (receiveMessage instanceof FullHttpRequest) {
                FullHttpRequest fullHttpRequest = (FullHttpRequest) receiveMessage;
                LOGGER.info("├ [握手]: {}", fullHttpRequest.uri());
                // 握手
                handlerHttpRequest(channelHandlerContext, fullHttpRequest);
                // 发送连接成功给客户端
                channelHandlerContext.channel().write(new TextWebSocketFrame("连接成功"));
            }
            // WebSocket接入
            else if (receiveMessage instanceof WebSocketFrame) {
                WebSocketFrame webSocketFrame = (WebSocketFrame) receiveMessage;
                handlerWebSocketFrame(channelHandlerContext, webSocketFrame);
            }
        }
    
        /**
         * 第一次握手
         *
         * @param channelHandlerContext channelHandlerContext
         * @param req                   请求
         */
        private void handlerHttpRequest(ChannelHandlerContext channelHandlerContext, FullHttpRequest req) {
            // 构造握手响应返回,本机测试
            WebSocketServerHandshakerFactory wsFactory
                    = new WebSocketServerHandshakerFactory(webSocketUrl, Constant.NULL, Constant.FALSE);
            // region 从连接路径中截取连接用户名
            String uri = req.uri();
            int i = uri.lastIndexOf("/");
            String userName = uri.substring(i + 1, uri.length());
            // endregion
            Channel connectChannel = channelHandlerContext.channel();
            // 加入在线用户
            WebSocketUsers.put(userName, connectChannel);
            socketServerHandShaker = wsFactory.newHandshaker(req);
            if (socketServerHandShaker == null) {
                // 发送版本错误
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(connectChannel);
            } else {
                // 握手响应
                socketServerHandShaker.handshake(connectChannel, req);
            }
        }
    
        /**
         * webSocket处理逻辑
         *
         * @param channelHandlerContext channelHandlerContext
         * @param frame                 webSocketFrame
         */
        private void handlerWebSocketFrame(ChannelHandlerContext channelHandlerContext, WebSocketFrame frame) throws IOException {
            Channel channel = channelHandlerContext.channel();
            // region 判断是否是关闭链路的指令
            if (frame instanceof CloseWebSocketFrame) {
                LOGGER.info("├ 关闭与客户端[{}]链接", channel.remoteAddress());
                socketServerHandShaker.close(channel, (CloseWebSocketFrame) frame.retain());
                return;
            }
            // endregion
            // region 判断是否是ping消息
            if (frame instanceof PingWebSocketFrame) {
                LOGGER.info("├ [Ping消息]");
                channel.write(new PongWebSocketFrame(frame.content().retain()));
                return;
            }
            // endregion
            // region 纯文本消息
            if (frame instanceof TextWebSocketFrame) {
                String text = ((TextWebSocketFrame) frame).text();
                LOGGER.info("├ [{} 接收到客户端的消息]: {}", new Date(), text);
                channel.write(new TextWebSocketFrame(new Date() + " 服务器将你发的消息原样返回:" + text));
            }
            // endregion
            // region 二进制消息 此处使用了MessagePack编解码方式
            if (frame instanceof BinaryWebSocketFrame) {
                BinaryWebSocketFrame binaryWebSocketFrame = (BinaryWebSocketFrame) frame;
                ByteBuf content = binaryWebSocketFrame.content();
                LOGGER.info("├ [二进制数据]:{}", content);
                final int length = content.readableBytes();
                final byte[] array = new byte[length];
                content.getBytes(content.readerIndex(), array, 0, length);
                MessagePack messagePack = new MessagePack();
                WebSocketMessageEntity webSocketMessageEntity = messagePack.read(array, WebSocketMessageEntity.class);
                LOGGER.info("├ [解码数据]: {}", webSocketMessageEntity);
                WebSocketUsers.sendMessageToUser(webSocketMessageEntity.getAcceptName(), webSocketMessageEntity.getContent());
            }
            // endregion
        }
    

    至此,服务端算是开发完成。但可以看出,服务端中仍有很大的发展空间,细心的同学可以发现我在第一次握手时,将Channel存储了起来,对于上述的三种情况也有简易的实现方案。

    如果有必要,我也会将非浏览器客户端代码(非Js客户端)写成例子,共享出来。

    一款基于Netty开发的WebSocket服务器

    代码地址如下:
    http://www.demodashi.com/demo/13577.html

    注:本文著作权归作者,由demo大师代发,拒绝转载,转载需要作者授权

  • 相关阅读:
    Java基本语法--程序流程控制
    Java基本语法--控制台输入(Scanner类)
    Java基本语法--运算符
    curl:出现SSL错误提示
    升级ruby的版本
    Linux之expect非交互式功能
    Rsync备份同步数据工具
    Linux发展历史
    解决DDOS攻击生产案例
    用shell脚本监控MySQL主从同步
  • 原文地址:https://www.cnblogs.com/demodashi/p/9443544.html
Copyright © 2011-2022 走看看