Netty实现聊天室

约定不等于承诺〃 2023-07-05 07:58 84阅读 0赞

文章目录

注：更多netty相关文章请访问博主专栏： netty专栏

本文内容基于上一篇博客 netty实现WebSocket协议，一些基本使用请参考该博客。

本例实现的功能:

  • 有新成员加入时，群广播消息，欢迎加入
  • 有成员退出时，群广播消息，退出
  • 每个成员都可以发送消息，消息广播给群内的每个人

完整的服务器代码如下：

  1. package com.example;
  3. import com.alibaba.fastjson.JSONObject;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.buffer.ByteBuf;
  6. import io.netty.buffer.Unpooled;
  7. import io.netty.channel.*;
  8. import io.netty.channel.group.ChannelGroup;
  9. import io.netty.channel.group.DefaultChannelGroup;
  10. import io.netty.channel.nio.NioEventLoopGroup;
  11. import io.netty.channel.socket.SocketChannel;
  12. import io.netty.channel.socket.nio.NioServerSocketChannel;
  13. import io.netty.handler.codec.http.*;
  14. import io.netty.handler.codec.http.websocketx.*;
  15. import io.netty.handler.logging.LogLevel;
  16. import io.netty.handler.logging.LoggingHandler;
  17. import io.netty.handler.stream.ChunkedFile;
  18. import io.netty.handler.stream.ChunkedWriteHandler;
  19. import io.netty.util.CharsetUtil;
  20. import io.netty.util.concurrent.GlobalEventExecutor;
  22. import javax.activation.MimetypesFileTypeMap;
  23. import java.io.File;
  24. import java.io.FileNotFoundException;
  25. import java.io.RandomAccessFile;
  26. import java.io.UnsupportedEncodingException;
  27. import java.net.InetSocketAddress;
  28. import java.net.URLDecoder;
  29. import java.text.SimpleDateFormat;
  30. import java.util.Date;
  31. import java.util.HashMap;
  32. import java.util.Map;
  33. import java.util.concurrent.*;
  34. import java.util.regex.Pattern;
  36. /** * netty实现聊天室 */
  37. public class MyChatRoomServer { 
  39.     int port;
  41.     public MyChatRoomServer(int port) { 
  42.         this.port = port;
  43.     }
  45.     public void start() { 
  46.         ServerBootstrap bootstrap = new ServerBootstrap();
  47.         EventLoopGroup boss = new NioEventLoopGroup();
  48.         EventLoopGroup work = new NioEventLoopGroup();
  49.         try { 
  50.             bootstrap.group(boss, work)
  51.                     .handler(new LoggingHandler(LogLevel.DEBUG))
  52.                     .channel(NioServerSocketChannel.class)
  53.                     .childHandler(new ChatRoomServerInitializer());
  55.             ChannelFuture f = bootstrap.bind(new InetSocketAddress(port)).sync();
  56.             System.out.println("http server started . port : " + port);
  57.             f.channel().closeFuture().sync();
  59.         } catch (Exception e) { 
  60.             e.printStackTrace();
  61.         } finally { 
  62.             boss.shutdownGracefully();
  63.             work.shutdownGracefully();
  64.         }
  65.     }
  67.     public static void main(String[] args) { 
  68.         MyChatRoomServer server = new MyChatRoomServer(8080);// 8081为启动端口
  69.         server.start();
  70.     }
  71. }
  73. class ChatRoomServerInitializer extends ChannelInitializer<SocketChannel> { 
  75.     @Override
  76.     protected void initChannel(SocketChannel channel) { 
  77.         ChannelPipeline pipeline = channel.pipeline();
  78.         pipeline.addLast(new HttpServerCodec())// http 编解器
  79.                 // http 消息聚合器 512*1024为接收的最大contentlength
  80.                 .addLast("httpAggregator", new HttpObjectAggregator(512 * 1024))
  81.                 // 支持异步发送大的码流(大的文件传输),但不占用过多的内存，防止java内存溢出
  82.                 .addLast("http-chunked", new ChunkedWriteHandler())
  83.                 .addLast(new ChatRoomRequestHandler());// 请求处理器
  84.     }
  85. }
  87. class ChatRoomRequestHandler extends SimpleChannelInboundHandler<Object> { 
  89.     private WebSocketServerHandshaker handshaker;
  91.     @Override
  92.     protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { 
  93.         System.out.println("收到消息：" + msg);
  94.         if (msg instanceof FullHttpRequest) { 
  95.             //以http请求形式接入，但是走的是websocket
  96.             handleHttpRequest(ctx, (FullHttpRequest) msg);
  97.         } else if (msg instanceof WebSocketFrame) { 
  98.             //处理websocket客户端的消息
  99.             handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
  100.         }
  101.     }
  103.     @Override
  104.     public void channelActive(ChannelHandlerContext ctx) throws Exception { 
  105.         //添加连接
  106.         System.out.println("客户端加入连接：" + ctx.channel());
  107.         ChannelSupervise.addChannel(ctx.channel());
  109.         TextWebSocketFrame tws = new TextWebSocketFrame(
  110.                 "欢迎 " + ctx.channel().id().asShortText() + "; 当前在线总人数：" + ChannelSupervise.count());
  111.         ChannelSupervise.send2All(tws);
  113.     }
  115.     @Override
  116.     public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
  117.         //断开连接
  118.         System.out.println("客户端断开连接：" + ctx.channel());
  119.         ChannelSupervise.removeChannel(ctx.channel());
  121.         TextWebSocketFrame tws = new TextWebSocketFrame(
  122.                 "再见 " + ctx.channel().id().asShortText() + "; 当前在线总人数：" + ChannelSupervise.count());
  123.         ChannelSupervise.send2All(tws);
  124.     }
  126.     @Override
  127.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
  128.         ctx.flush();
  129.     }
  131.     /* 对WebSocket请求进行处理 */
  132.     private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) { 
  133.         // 判断是否关闭链路的指令
  134.         if (frame instanceof CloseWebSocketFrame) { 
  135.             handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
  136.             return;
  137.         }
  138.         // 判断是否ping消息，如果是，则构造pong消息返回。用于心跳检测
  139.         if (frame instanceof PingWebSocketFrame) { 
  140.             ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
  141.             return;
  142.         }
  144.         // 本例程仅支持文本消息，不支持二进制消息
  145.         if (!(frame instanceof TextWebSocketFrame)) { 
  146.             System.out.println("本例程仅支持文本消息，不支持二进制消息");
  147.             throw new UnsupportedOperationException(
  148.                     String.format("%s frame types not supported", frame.getClass().getName()));
  149.         }
  151.         //处理客户端请求并返回应答消息
  152.         String request = ((TextWebSocketFrame) frame).text();
  153.         System.out.println(request);
  154.         TextWebSocketFrame tws = new TextWebSocketFrame(ctx.channel().id().asShortText() + "：" + request);
  156.         // 群发
  157.         ChannelSupervise.send2All(tws);
  158.     }
  160.     /** * 唯一的一次http请求。 * 该方法用于处理websocket握手请求 */
  161.     private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) { 
  162.         //如果HTTP解码失败，返回异常。要求Upgrade为websocket，过滤掉get/Post
  163.         if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) { 
  164.             //若不是websocket方式，则创建BAD_REQUEST（400）的req，返回给客户端
  165.             sendHttpResponse(ctx, req, new DefaultFullHttpResponse(
  166.                     HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
  167.             return;
  168.         }
  170. // 构造握手响应返回，本机测试
  171.         WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
  172.                 "ws://localhost:8080/websocket", null, false);
  173.         //通过工厂来创建WebSocketServerHandshaker实例
  174.         handshaker = wsFactory.newHandshaker(req);
  175.         if (handshaker == null) { 
  176.             WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
  177.         } else { 
  178.             /* 通过WebSocketServerHandshaker来构建握手响应消息返回给客户端。 同时将WebSocket相关编解码类添加到ChannelPipeline中，该功能需要阅读handshake的源码。 */
  179.             handshaker.handshake(ctx.channel(), req);
  180.         }
  181.     }
  183.     /** * 拒绝不合法的请求，并返回错误信息 */
  184.     private static void sendHttpResponse(ChannelHandlerContext ctx,
  185.                                          FullHttpRequest req, DefaultFullHttpResponse res) { 
  186.         // 返回应答给客户端
  187.         if (res.status().code() != 200) { 
  188.             ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
  189.             res.content().writeBytes(buf);
  190.             buf.release();
  191.             HttpUtil.setContentLength(res, res.content().readableBytes());
  192.         }
  194.         ChannelFuture f = ctx.channel().writeAndFlush(res);
  196.         // 如果是非Keep-Alive，关闭连接
  197.         if (!HttpUtil.isKeepAlive(req) || res.status().code() != 200) { 
  198.             f.addListener(ChannelFutureListener.CLOSE);
  199.         }
  200.     }
  201. }
  203. public class ChannelSupervise { 
  204.     /** * ChannelGroup是netty提供用于管理web于服务器建立的通道channel的， * 其本质是一个高度封装的set集合，在服务器广播消息时， * 可以直接通过它的writeAndFlush将消息发送给集合中的所有通道中去 */
  205.     private static ChannelGroup GlobalGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  206.     /** * ChannelMap维护的是channelID和channel的对应关系，用于向指定channel发送消息 */
  207.     private static ConcurrentMap<String, ChannelId> ChannelMap = new ConcurrentHashMap<>();
  209.     public static void addChannel(Channel channel) { 
  210.         GlobalGroup.add(channel);
  211.         ChannelMap.put(channel.id().asShortText(), channel.id());
  212.     }
  214.     public static void removeChannel(Channel channel) { 
  215.         GlobalGroup.remove(channel);
  216.         ChannelMap.remove(channel.id().asShortText());
  217.     }
  219.     //找到某个channel来发送消息
  220.     public static Channel findChannel(String id) { 
  221.         return GlobalGroup.find(ChannelMap.get(id));
  222.     }
  224.     public static void send2All(TextWebSocketFrame tws) { 
  225.         GlobalGroup.writeAndFlush(tws);
  226.     }
  228.     public static int count() { 
  229.         return GlobalGroup.size();
  230.     }
  231. }

客户端页面代码：

  1. <!DOCTYPE html>
  2. <html>
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>my websocket client</title>
  6. </head>
  7. <body>
  8. <textarea id="msgBoxs"></textarea><br>
  9. 待发送消息`：<input type="text" id="msg">
  10. <input type="button" id="sendBtn" onclick="send()" value="发送">
  11. <script type="application/javascript"> var socket ; if(!window.WebSocket){ window.WebSocket = window.MozWebSocket; } if(window.WebSocket){ var msgBoxs = document.getElementById("msgBoxs") var msgBox = document.getElementById("msg") socket = new WebSocket("ws://localhost:8080/websocket") socket.onopen = function (evt) { console.log("Connection open ..."); socket.send("Hello WebSocket!"); } socket.onmessage = function (evt) { console.log("Received Message: ", evt.data) msgBoxs.value = msgBoxs.value + "\n" + evt.data } socket.onclose = function (evt) { console.log("Connect closed."); } }else{ alert("ERROR:您的浏览器不支持WebSocket！！"); } function send() { var msg = msgBox.value socket.send(msg) //msgBox.value = "" } </script>
  12. </body>
  13. </html>

依次打开三个浏览器，作为三个客户端，以channelID作为用户ID。
启动以后可以看到当有用户加入时，前面新加入的用户会受到欢迎的消息。每个用户都可以发送消息，消息会被广播到群里的所有人：
在这里插入图片描述

当关闭一个窗口时，其他窗口会收到再见消息。
注意这里最好在客户端检测退出关闭socket的请求，页面关闭前主动释放与服务器的链接，释放资源，并且其他用户可以第一时间感知到下线，否则需要等待TCP超时后才可以受到下线消息。

简易聊天室完成

注：更多netty相关文章请访问博主专栏： netty专栏

发表评论

表情：
评论列表 （有 0 条评论，84人围观）

还没有评论，来说两句吧...

相关阅读