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

╰半夏微凉° 2023-10-15 11:44 59阅读 0赞

#### 文章目录 ####

*  前言
 *  声明
 *  功能说明
 *  实现步骤
 *   *  WebSocket 服务启动
     *  Channel 初始化
     *  HTTP 请求处理
     *  HTTP 页面内容
     *  WebSocket 请求处理
 *  效果展示
 *  总结

## 前言 ##

通过之前的文章介绍，我们可以深刻认识到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 WebSocketServerInitializer
            extends 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;
        }
    
        @Override
        public 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 ProcessWsIndexPageHandler
            extends SimpleChannelInboundHandler<FullHttpRequest> {
        
    
        private final String websocketPath;
    
        public ProcessWsIndexPageHandler(String websocketPath) {
        
            this.websocketPath = websocketPath;
        }
    
        @Override
        protected 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));
            }
        }
    
        @Override
        public 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 ProcesssWsFrameHandler
            extends SimpleChannelInboundHandler<WebSocketFrame> {
        
    
        private final ChannelGroup group;
    
        public ProcesssWsFrameHandler(ChannelGroup group) {
        
            this.group = group;
        }
    
        private static final Logger logger
                = LoggerFactory.getLogger(ProcesssWsFrameHandler.class);
    
        @Override
        protected 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()方法以处理自定义事件*/
        @Override
        public 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);
            }
        }
    }

## 效果展示 ##

**服务端启动**

![在这里插入图片描述][3065a53bf57349208c65067b0886715f.png]

**聊天页面1**

![在这里插入图片描述][2eb507e89cd94a94861eb3558402ecfc.png]

**聊天页面2**

![在这里插入图片描述][42e49df7b815431fb40f8d70a176afe3.png]

## 总结 ##

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

[3065a53bf57349208c65067b0886715f.png]: https://img-blog.csdnimg.cn/3065a53bf57349208c65067b0886715f.png
[2eb507e89cd94a94861eb3558402ecfc.png]: https://img-blog.csdnimg.cn/2eb507e89cd94a94861eb3558402ecfc.png
[42e49df7b815431fb40f8d70a176afe3.png]: https://img-blog.csdnimg.cn/42e49df7b815431fb40f8d70a176afe3.png