使用 Netty 实现群聊功能的步骤和注意事项

前言

通过之前的文章介绍，我们可以深刻认识到Netty在网络编程领域的卓越表现和强大实力。这篇文章将介绍如何利用 Netty 框架开发一个 WebSocket 服务端，从而实现一个简单的在线聊天功能。

声明

文章中所提供的代码仅供参考，旨在帮助无 Netty 经验的开发人员快速上手。请注意，这些代码并不适用于实际应用中。

功能说明

聊天页面：

• 用户进入页面后，会看到一个简单的文本框，可以用来发送消息。
• 页面下方会显示聊天的消息内容。

服务端主要有以下三个功能：

• 响应聊天页面：用来接收和响应聊天页面的请求。
• 处理消息：对接收到的消息进行处理。
• 实现群聊功能：提供群聊的功能，使多个用户能够在同一个聊天室中进行交流。

功能很简单，但是可以通过这个示例实现更多复杂的场景。

实现步骤

创建一个简单的 Maven 项目，直接引入 netty-all 包即可编码。

<dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.28.Final</version>
</dependency>

实现该功能共有五个类，如下：

├── MakeIndexPage.java
├── ProcessWsIndexPageHandler.java
├── ProcesssWsFrameHandler.java
├── WebSocketServer.java
└── WebSocketServerInitializer.java

下面对实现该功能所涉及的五个类的代码进行详细说明

WebSocket 服务启动

这个类是一个基于 Netty 启动的常规服务端。它包含了一些配置项，包括 Reactor 模式、IO 类型以及消息处理配置，大部分都是这样。代码如下：

/*** 类说明：*/
public final class WebSocketServer {/*创建 DefaultChannelGroup，用来保存所有已经连接的 WebSocket Channel，群发和一对一功能可以用上*/private final static ChannelGroup channelGroup =new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);static final boolean SSL = false;//是否启用ssl/*通过ssl访问端口为8443，否则为8080*/static final int PORT= Integer.parseInt(System.getProperty("port", SSL? "8443" : "80"));public static void main(String[] args) throws Exception {/*SSL配置*/final SslContext sslCtx;if (SSL) {SelfSignedCertificate ssc = new SelfSignedCertificate();sslCtx = SslContextBuilder.forServer(ssc.certificate(),ssc.privateKey()).build();} else {sslCtx = null;}EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap b = new ServerBootstrap();b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new WebSocketServerInitializer(sslCtx,channelGroup));Channel ch = b.bind(PORT).sync().channel();System.out.println("打开浏览器访问： " +(SSL? "https" : "http") + "://127.0.0.1:" + PORT + '/');ch.closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

Channel 初始化

这个类的主要功能是创建了一个 ChannelInitializer，用于初始化 ChannelPipeline，并添加了一些通道处理器。这些处理器包括由Netty提供的处理SSL协议、处理HTTP协议和支持WebSocket协议的功能，还有一些由业务自定义的处理器，用于处理页面展示和处理WebSocket数据。代码如下：

/*** 类说明：增加handler*/
public class WebSocketServerInitializerextends ChannelInitializer<SocketChannel> {private final ChannelGroup group;/*websocket访问路径*/private static final String WEBSOCKET_PATH = "/chat";private final SslContext sslCtx;public WebSocketServerInitializer(SslContext sslCtx,ChannelGroup group) {this.sslCtx = sslCtx;this.group = group;}@Overridepublic void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();if (sslCtx != null) {pipeline.addLast(sslCtx.newHandler(ch.alloc()));}/*增加对http的支持*/pipeline.addLast(new HttpServerCodec());pipeline.addLast(new HttpObjectAggregator(65536));/*Netty提供，支持WebSocket应答数据压缩传输*/pipeline.addLast(new WebSocketServerCompressionHandler());/*Netty提供，对整个websocket的通信进行了初始化(发现http报文中有升级为websocket的请求)，包括握手，以及以后的一些通信控制*/pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH,null, true));/*浏览器访问时展示index页面*/pipeline.addLast(new ProcessWsIndexPageHandler(WEBSOCKET_PATH));/*对websocket的数据进行处理*/pipeline.addLast(new ProcesssWsFrameHandler(group));}
}

HTTP 请求处理

这个类的主要功能是在收到 HTTP 请求时，当 URI 为“/”或“/index.html”时，会返回一个聊天界面作为响应。代码如下：

/*** 类说明：对http请求，将index的页面返回给前端*/
public class ProcessWsIndexPageHandlerextends SimpleChannelInboundHandler<FullHttpRequest> {private final String websocketPath;public ProcessWsIndexPageHandler(String websocketPath) {this.websocketPath = websocketPath;}@Overrideprotected void channelRead0(ChannelHandlerContext ctx,FullHttpRequest req) throws Exception {// 处理错误或者无法解析的http请求if (!req.decoderResult().isSuccess()) {sendHttpResponse(ctx, req,new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));return;}//只允许Get请求if (req.method() != GET) {sendHttpResponse(ctx, req,new DefaultFullHttpResponse(HTTP_1_1, FORBIDDEN));return;}// 发送index页面的内容if ("/".equals(req.uri()) || "/index.html".equals(req.uri())) {//生成WebSocket的访问地址，写入index页面中String webSocketLocation= getWebSocketLocation(ctx.pipeline(), req,websocketPath);System.out.println("WebSocketLocation:["+webSocketLocation+"]");//生成index页面的具体内容,并送往浏览器ByteBuf content= MakeIndexPage.getContent(webSocketLocation);FullHttpResponse res = new DefaultFullHttpResponse(HTTP_1_1, OK, content);res.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/html; charset=UTF-8");HttpUtil.setContentLength(res, content.readableBytes());sendHttpResponse(ctx, req, res);} else {sendHttpResponse(ctx, req,new DefaultFullHttpResponse(HTTP_1_1, NOT_FOUND));}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();}/*发送应答*/private static void sendHttpResponse(ChannelHandlerContext ctx,FullHttpRequest req,FullHttpResponse res) {// 错误的请求进行处理 （code<>200).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);//对于不是长连接或者错误的请求直接关闭连接if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) {f.addListener(ChannelFutureListener.CLOSE);}}/*根据用户的访问，告诉用户的浏览器，WebSocket的访问地址*/private static String getWebSocketLocation(ChannelPipeline cp,HttpRequest req,String path) {String protocol = "ws";if (cp.get(SslHandler.class) != null) {protocol = "wss";}return protocol + "://" + req.headers().get(HttpHeaderNames.HOST)+ path;}
}

HTTP 页面内容

这个类的主要目的是生成一个包含消息发送框和内容展示功能的HTML页面，并实现WebSocket的相关功能，包括建立连接、向服务端发送消息以及接收服务端的响应。当然，也可以单独写一个HTML文件。代码如下：

/*** 类说明：生成index页面的内容*/
public final class MakeIndexPage {private static final String NEWLINE = "\r\n";public static ByteBuf getContent(String webSocketLocation) {return Unpooled.copiedBuffer("<html><head><title>Web Socket Test</title><meta charset=\"utf-8\" /></head>"+ NEWLINE +"<body>" + NEWLINE +"<script type=\"text/javascript\">" + NEWLINE +"var socket;" + NEWLINE +"if (!window.WebSocket) {" + NEWLINE +"  window.WebSocket = window.MozWebSocket;" + NEWLINE +'}' + NEWLINE +"if (window.WebSocket) {" + NEWLINE +"  socket = new WebSocket(\"" + webSocketLocation + "\");"+ NEWLINE +"  socket.onmessage = function(event) {" + NEWLINE +"    var ta = document.getElementById('responseText');"+ NEWLINE +"    ta.value = ta.value + '\\n' + event.data" + NEWLINE +"  };" + NEWLINE +"  socket.onopen = function(event) {" + NEWLINE +"    var ta = document.getElementById('responseText');"+ NEWLINE +"    ta.value = \"Web Socket opened!\";" + NEWLINE +"  };" + NEWLINE +"  socket.onclose = function(event) {" + NEWLINE +"    var ta = document.getElementById('responseText');"+ NEWLINE +"    ta.value = ta.value + \"Web Socket closed\"; "+ NEWLINE +"  };" + NEWLINE +"} else {" + NEWLINE +"  alert(\"Your browser does not support Web Socket.\");"+ NEWLINE +'}' + NEWLINE +NEWLINE +"function send(message) {" + NEWLINE +"  if (!window.WebSocket) { return; }" + NEWLINE +"  if (socket.readyState == WebSocket.OPEN) {" + NEWLINE +"    socket.send(message);" + NEWLINE +"  } else {" + NEWLINE +"    alert(\"The socket is not open.\");" + NEWLINE +"  }" + NEWLINE +'}' + NEWLINE +"</script>" + NEWLINE +"<form οnsubmit=\"return false;\">" + NEWLINE +"<input type=\"text\" name=\"message\" " +"value=\"Hi, 你好啊\"/>" +"<input type=\"button\" value=\"发送\""+ NEWLINE +"       οnclick=\"send(this.form.message.value)\" />"+ NEWLINE +"<h3>消息内容</h3>" + NEWLINE +"<textarea id=\"responseText\" " +"style=\"width:500px;height:300px;\"></textarea>"+ NEWLINE +"</form>" + NEWLINE +"</body>" + NEWLINE +"</html>" + NEWLINE, CharsetUtil.UTF_8);}}

WebSocket 请求处理

这个类的主要功能是处理与 Channel 相关的事件。例如，当一个 Channel 连接成功时，会将该 Channel 添加到一个 ChannelGroup 中。当接收到该 Channel 的数据时，可以通过向 ChannelGroup 写入数据来实现群聊效果。代码如下

/*** 类说明：对websocket的数据进行处理*/
public class ProcesssWsFrameHandlerextends SimpleChannelInboundHandler<WebSocketFrame> {private final ChannelGroup group;public ProcesssWsFrameHandler(ChannelGroup group) {this.group = group;}private static final Logger logger= LoggerFactory.getLogger(ProcesssWsFrameHandler.class);@Overrideprotected void channelRead0(ChannelHandlerContext ctx,WebSocketFrame frame) throws Exception {//判断是否为文本帧，目前只处理文本帧if (frame instanceof TextWebSocketFrame) {// Send the uppercase string back.String request = ((TextWebSocketFrame) frame).text();logger.info("{} received {}", ctx.channel(), request);
//            ctx.channel().writeAndFlush(
//                    new TextWebSocketFrame(request.toUpperCase(Locale.CHINA)));/*群发实现:一对一道理一样*/group.writeAndFlush(new TextWebSocketFrame(ctx.channel().remoteAddress() + " :" + request.toUpperCase(Locale.CHINA)));} else {String message = "unsupported frame type: "+ frame.getClass().getName();throw new UnsupportedOperationException(message);}}/*重写 userEventTriggered()方法以处理自定义事件*/@Overridepublic void userEventTriggered(ChannelHandlerContext ctx,Object evt) throws Exception {/*检测事件，如果是握手成功事件，做点业务处理*/if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {//通知所有已经连接的 WebSocket 客户端新的客户端已经连接上了group.writeAndFlush(new TextWebSocketFrame("Client " + ctx.channel().remoteAddress() + " joined"));//将新的 WebSocket Channel 添加到 ChannelGroup 中，// 以便它可以接收到所有的消息group.add(ctx.channel());} else {super.userEventTriggered(ctx, evt);}}
}

效果展示

服务端启动

聊天页面1

聊天页面2

总结

总的来说，基于 Netty 实现一个 WebSocket 功能是非常方便且高效的，但是我们需要知其所以然，要理解 Websocket 协议，也要懂的在 Netty 中，通过添加 ChannelHandler 来处理各种异常情况，例如握手失败、连接关闭等，当然，还要考虑安全性问题，例如处理跨站脚本攻击(XSS)、防止恶意数据传输等。