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

逃离我推掉我的手 2022-05-11 04:33 256阅读 0赞

> 代码地址如下：  
> [http://www.demodashi.com/demo/13577.html][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 位）

## 运行结果 ##

请看下图：

![推送结果图][qWgfzpV16LEuPWDGizm.png]

## 实现步骤及源码 ##

### WebSocket ###

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

### 目录结构 ###

![目录结构][ZUbBw3vxbfoewFlPX7D.png]

### 核心类讲解 ###

*  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][http_www.demodashi.com_demo_13577.html]

> 注：本文著作权归作者，由demo大师发表，拒绝转载，转载需要作者授权

[http_www.demodashi.com_demo_13577.html]: http://www.demodashi.com/demo/13577.html
[qWgfzpV16LEuPWDGizm.png]: /images/20220511/62935ee0d6434fe19bd4b4b8e733ff5e.png
[ZUbBw3vxbfoewFlPX7D.png]: /images/20220511/025fbd3c71894e9f848f3f3ad42b56cd.png