WebSocket 是一种全双工通信协议，可以在客户端和服务器之间建立持久连接，广泛应用于即时聊天、实时数据推送等场景。Netty 提供了对 WebSocket 协议的全面支持，可以快速实现一个高效的 WebSocket 服务。
本文将从基础原理出发，逐步讲解如何使用 Netty 实现 WebSocket 通信。
1. WebSocket 基本原理
1.1 什么是 WebSocket？
- WebSocket 是一个基于 TCP 的全双工协议，通过一次 HTTP 请求升级为 WebSocket 协议后，客户端和服务器可以在单个连接上自由发送消息。
- WebSocket 的默认端口：
- 80（非加密通信）
- 443（加密通信）
1.2 WebSocket 工作流程
- 连接建立：通过 HTTP 请求，使用
Upgrade: websocket升级协议。 - 数据传输：升级后使用帧（Frame）进行数据传输，帧可以是文本、二进制或控制信息。
- 连接关闭：客户端或服务器可随时关闭连接。
2. Netty 对 WebSocket 的支持
Netty 提供了以下 WebSocket 处理器：
WebSocketServerProtocolHandler：处理 WebSocket 协议升级和帧的管理。TextWebSocketFrame：表示 WebSocket 文本帧。BinaryWebSocketFrame：表示 WebSocket 二进制帧。PingWebSocketFrame和PongWebSocketFrame：用于心跳检测。
3. 使用 Netty 实现 WebSocket 服务
以下是完整的实现过程，包括服务端和客户端代码。
3.1 服务端代码
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;public class WebSocketServer {public static void main(String[] args) throws InterruptedException {EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();// HTTP 协议支持pipeline.addLast(new HttpServerCodec());pipeline.addLast(new HttpObjectAggregator(65536));pipeline.addLast(new ChunkedWriteHandler());// WebSocket 协议支持pipeline.addLast(new WebSocketServerProtocolHandler("/ws"));// 自定义处理器pipeline.addLast(new WebSocketFrameHandler());}});System.out.println("WebSocket server started on port 8080");ChannelFuture future = bootstrap.bind(8080).sync();future.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}class WebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {String receivedText = frame.text();System.out.println("Received: " + receivedText);// 返回响应ctx.channel().writeAndFlush(new TextWebSocketFrame("Server received: " + receivedText));}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {System.out.println("Client connected: " + ctx.channel().id().asLongText());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {System.out.println("Client disconnected: " + ctx.channel().id().asLongText());}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}
服务端代码说明
-
HTTP 支持：
HttpServerCodec：HTTP 请求和响应的编解码器。HttpObjectAggregator：将多个 HTTP 消息聚合成一个完整的 FullHttpRequest 或 FullHttpResponse。ChunkedWriteHandler：支持大数据流的写操作。
-
WebSocket 支持：
WebSocketServerProtocolHandler：- 处理 WebSocket 协议的升级（HTTP -> WebSocket）。
/ws：指定 WebSocket 的路径。
-
自定义处理器：
WebSocketFrameHandler：处理 WebSocket 帧，接收文本消息并发送响应。
3.2 客户端代码
客户端可以使用浏览器或自定义 WebSocket 客户端库（如 JavaScript 的 WebSocket 对象）。
以下是一个 JavaScript 客户端示例：
const ws = new WebSocket('ws://localhost:8080/ws');// 连接建立
ws.onopen = () => {console.log('Connected to server');ws.send('Hello Server!');
};// 接收消息
ws.onmessage = (event) => {console.log('Received from server:', event.data);
};// 连接关闭
ws.onclose = () => {console.log('Connection closed');
};
运行代码时，可以在浏览器开发者工具的控制台中看到通信结果。
4. 功能扩展
4.1 广播消息
在服务端实现广播功能，可以向所有连接的客户端发送消息。
修改 WebSocketFrameHandler：
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;public class WebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) throws Exception {String receivedText = frame.text();System.out.println("Received: " + receivedText);// 广播消息到所有客户端for (Channel channel : channels) {channel.writeAndFlush(new TextWebSocketFrame("[Broadcast] " + receivedText));}}@Overridepublic void handlerAdded(ChannelHandlerContext ctx) throws Exception {channels.add(ctx.channel());System.out.println("Client connected: " + ctx.channel().id().asLongText());}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) throws Exception {channels.remove(ctx.channel());System.out.println("Client disconnected: " + ctx.channel().id().asLongText());}
}
4.2 心跳检测
通过 IdleStateHandler 实现心跳检测，检测客户端是否存活。
import io.netty.handler.timeout.IdleStateHandler;@Override
protected void initChannel(Channel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();// 添加心跳检测pipeline.addLast(new IdleStateHandler(60, 30, 0)); // 读超时 60 秒，写超时 30 秒// 其他处理器pipeline.addLast(new WebSocketFrameHandler());
}
5. 测试流程
- 启动服务端代码，监听
8080端口。 - 使用浏览器或 WebSocket 客户端连接
ws://localhost:8080/ws。 - 测试发送和接收消息。
- 多个客户端连接时，验证广播功能。
6. 总结
通过本文实现了一个简单的 WebSocket 服务，展示了 Netty 对 WebSocket 的强大支持。Netty 提供了对 WebSocket 协议的底层支持，以及灵活的扩展功能，适合构建高性能的实时通信服务。
关键点回顾：
- WebSocket 协议升级：通过
WebSocketServerProtocolHandler实现。 - 帧的处理：通过
TextWebSocketFrame和BinaryWebSocketFrame等对象实现数据收发。 - 功能扩展：实现广播消息和心跳检测。
通过这些功能，你可以进一步开发复杂的实时通信系统，如在线聊天室、股票实时推送等应用场景。