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

立即下载用AI写一个

发布时间：2018-07-14

71人 | 浏览：6894次 | 收藏 |

技术：SpringBoot + Netty + WebSocket

运行环境：JDK1.8+、Netty5.0、Intellij Idea

概述

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

详细

# 一款基于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 位）

## 运行结果

请看下图：

![推送结果图](http://www.demodashi.com/contentImages/image/20180713/qWgfzpV16LEuPWDGizm.png "推送结果图")

## 实现步骤及源码

### WebSocket

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

### 目录结构

![目录结构](http://www.demodashi.com/contentImages/image/20180713/ZUbBw3vxbfoewFlPX7D.png "目录结构")

### 核心类讲解

- WebSocketChildHandler
- WebSocketServerHandler

#### WebSocketChildHandler

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

```java
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

这个类是真正的核心类，这个类的主要功能为：

- 进行第一次握手
- 对消息进行处理
  - 可以实现点对点通信
  - 可以实现广播功能
  - 可以实现点对端通信

```java
/**
     * 接收客户端发送的消息
     *
     * @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客户端)写成例子，共享出来。

#注
后续的更新迭代，我会以新的DEMO来完善，修改。谢谢大家。此DEMO如果在你的机器上不能正常运行，请大家看一下配置是否正确，亲测可行

## 硬件环境

| 名称 |              规格               |
| :--: | :-----------------------------: |
| 机器 |          Mac Book Pro           |
| CPU  |      2.4 GHz Intel Core i7      |
| 内存 |      16 GB 1867 MHz LPDDR3      |
| 显卡 | Intel Iris Graphics 540 1536 MB |

## 软件环境

|    名称     |                             规格                             |
| :---------: | :----------------------------------------------------------: |
|     IDE     |          IntelliJ IDEA 2018.3.6 (Ultimate Edition)           |
|     JDK     | java version "1.8.0_161"<br />Java(TM) SE Runtime Environment (build 1.8.0_161-b12)<br />Java HotSpot(TM) 64-Bit Server VM (build 25.161-b12, mixed mode) |
|    maven    |                      Apache Maven 3.5.2                      |
| Spring-boot |                        2.0.3.RELEASE                         |
|    Netty    |                         5.0.0.Alpha2                         |

心跳机制的集成，我会放在接下来的DEMO中共享，2019-04-10日之前提交到Demo大师中。

本实例支付的费用只是购买源码的费用，如有疑问欢迎在文末留言交流，如需作者在线代码指导、定制等，在作者开启付费服务后，可以点击“购买服务”进行实时联系，请知悉，谢谢

感谢

手机上随时阅读、收藏该文章？请扫下方二维码

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

概述

详细

相似例子推荐

评论

作者

评分详细

作者例子