SpringBoot2+Netty+WebSocket(netty实现websocket，支持URL参数)

╰半夏微凉° 2022-01-17 00:17 738阅读 0赞

关于Netty

Netty 是一个利用 Java 的高级网络的能力，隐藏其背后的复杂性而提供一个易于使用的 API 的客户端/服务器框架。

更新

  • 2019-7-11 新增URL参数支持，并解决了带参URL导致的连接自动断开问题，感谢大家的支持。

MAVEN依赖

  1. <dependencies>
  2.         <!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
  3.         <dependency>
  4.             <groupId>io.netty</groupId>
  5.             <artifactId>netty-all</artifactId>
  6.             <version>4.1.36.Final</version>
  7.         </dependency>
  9.     </dependencies>

SpringBootApplication

启动器中需要new一个NettyServer，并显式调用启动netty。

  1. @SpringBootApplication
  2. public class SpringCloudStudyDemoApplication { 
  3.     public static void main(String[] args) { 
  4.         SpringApplication.run(SpringCloudStudyDemoApplication.class,args);
  5.         try { 
  6.             new NettyServer(12345).start();
  7.             System.out.println("https://blog.csdn.net/moshowgame");
  8.             System.out.println("http://127.0.0.1:6688/netty-websocket/index");
  9.         }catch(Exception e) { 
  10.             System.out.println("NettyServerError:"+e.getMessage());
  11.         }
  12.     }
  13. }

NettyServer

启动的NettyServer，这里进行配置

  1. /** * NettyServer Netty服务器配置 * @author zhengkai.blog.csdn.net * @date 2019-06-12 */
  2. public class NettyServer { 
  3.     private final int port;
  5.     public NettyServer(int port) { 
  6.         this.port = port;
  7.     }
  9.     public void start() throws Exception { 
  10.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  12.         EventLoopGroup group = new NioEventLoopGroup();
  13.         try { 
  14.             ServerBootstrap sb = new ServerBootstrap();
  15.             sb.option(ChannelOption.SO_BACKLOG, 1024);
  16.             sb.group(group, bossGroup) // 绑定线程池
  17.                     .channel(NioServerSocketChannel.class) // 指定使用的channel
  18.                     .localAddress(this.port)// 绑定监听端口
  19.                     .childHandler(new ChannelInitializer<SocketChannel>() {  // 绑定客户端连接时候触发操作
  21.                         @Override
  22.                         protected void initChannel(SocketChannel ch) throws Exception { 
  23.                             System.out.println("收到新连接");
  24.                             //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
  25.                             ch.pipeline().addLast(new HttpServerCodec());
  26.                             //以块的方式来写的处理器
  27.                             ch.pipeline().addLast(new ChunkedWriteHandler());
  28.                             ch.pipeline().addLast(new HttpObjectAggregator(8192));
  29.                             ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
  30.                             ch.pipeline().addLast(new MyWebSocketHandler());
  31.                         }
  32.                     });
  33.             ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
  34.             System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
  35.             cf.channel().closeFuture().sync(); // 关闭服务器通道
  36.         } finally { 
  37.             group.shutdownGracefully().sync(); // 释放线程池资源
  38.             bossGroup.shutdownGracefully().sync();
  39.         }
  40.     }
  41. }

MyChannelHandlerPool

通道组池，管理所有websocket连接

  1. /** * MyChannelHandlerPool * 通道组池，管理所有websocket连接 * @author zhengkai.blog.csdn.net * @date 2019-06-12 */
  2. public class MyChannelHandlerPool { 
  4.     public MyChannelHandlerPool(){ }
  6.     public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  8. }

MyWebSocketHandler

处理ws一下几种情况：

  • channelActive与客户端建立连接
  • channelInactive与客户端断开连接

  • channelRead0客户端发送消息处理

    /* NettyServer Netty服务器配置 @author zhengkai.blog.csdn.net @date 2019-06-12 */
    public class NettyServer { 

    1. private final int port;
    3. public NettyServer(int port) { 
    4.     this.port = port;
    5. }
    7. public void start() throws Exception { 
    8.     EventLoopGroup bossGroup = new NioEventLoopGroup();
    10.     EventLoopGroup group = new NioEventLoopGroup();
    11.     try { 
    12.         ServerBootstrap sb = new ServerBootstrap();
    13.         sb.option(ChannelOption.SO_BACKLOG, 1024);
    14.         sb.group(group, bossGroup) // 绑定线程池
    15.                 .channel(NioServerSocketChannel.class) // 指定使用的channel
    16.                 .localAddress(this.port)// 绑定监听端口
    17.                 .childHandler(new ChannelInitializer<SocketChannel>() {  // 绑定客户端连接时候触发操作
    19.                     @Override
    20.                     protected void initChannel(SocketChannel ch) throws Exception { 
    21.                         System.out.println("收到新连接");
    22.                         //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
    23.                         ch.pipeline().addLast(new HttpServerCodec());
    24.                         //以块的方式来写的处理器
    25.                         ch.pipeline().addLast(new ChunkedWriteHandler());
    26.                         ch.pipeline().addLast(new HttpObjectAggregator(8192));
    27.                         ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
    28.                         ch.pipeline().addLast(new MyWebSocketHandler());
    29.                     }
    30.                 });
    31.         ChannelFuture cf = sb.bind().sync(); // 服务器异步创建绑定
    32.         System.out.println(NettyServer.class + " 启动正在监听： " + cf.channel().localAddress());
    33.         cf.channel().closeFuture().sync(); // 关闭服务器通道
    34.     } finally { 
    35.         group.shutdownGracefully().sync(); // 释放线程池资源
    36.         bossGroup.shutdownGracefully().sync();
    37.     }
    38. }

    }

socket.html

主要是连接ws，发送消息，以及消息反馈

  1. <!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" "http://www.w3.org/TR/xhtml1/DTD/xhtml1transitional.dtd">
  2. <html xmlns="http://www.w3.org/1999/xhtml">
  3. <head>
  4.     <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
  5.     <title>Netty-Websocket</title>
  6.     <script type="text/javascript"> // by zhengkai.blog.csdn.net var socket; if(!window.WebSocket){ window.WebSocket = window.MozWebSocket; } if(window.WebSocket){ socket = new WebSocket("ws://127.0.0.1:12345/ws"); socket.onmessage = function(event){ var ta = document.getElementById('responseText'); ta.value += event.data+"\r\n"; }; socket.onopen = function(event){ var ta = document.getElementById('responseText'); ta.value = "Netty-WebSocket服务器。。。。。。连接 \r\n"; }; socket.onclose = function(event){ var ta = document.getElementById('responseText'); ta.value = "Netty-WebSocket服务器。。。。。。关闭 \r\n"; }; }else{ alert("您的浏览器不支持WebSocket协议！"); } function send(message){ if(!window.WebSocket){ return;} if(socket.readyState == WebSocket.OPEN){ socket.send(message); }else{ alert("WebSocket 连接没有建立成功！"); } } </script>
  7. </head>
  8. <body>
  9. <form onSubmit="return false;">
  10.     <label>ID</label><input type="text" name="uid" value="${uid!!}" /> <br />
  11.     <label>TEXT</label><input type="text" name="message" value="这里输入消息" /> <br />
  12.     <br /> <input type="button" value="发送ws消息" onClick="send(this.form.uid.value+':'+this.form.message.value)" />
  13.     <hr color="black" />
  14.     <h3>服务端返回的应答消息</h3>
  15.     <textarea id="responseText" style="width: 1024px;height: 300px;"></textarea>
  16. </form>
  17. </body>
  18. </html>

Controller

写好了html当然还需要一个controller来引导页面。

  1. @RestController
  2. public class IndexController { 
  4.     @GetMapping("/index")
  5.     public ModelAndView  index(){ 
  6.         ModelAndView mav=new ModelAndView("socket");
  7.         mav.addObject("uid", RandomUtil.randomNumbers(6));
  8.         return mav;
  9.     }
  11. }

效果演示

在这里插入图片描述
在这里插入图片描述

思路优化" class="reference-link">在这里插入图片描述 思路优化

由于netty不能像默认的websocket一样设置一些PathVariable例如{uid}等参数(暂未发现可以,如果有发现欢迎补充),所以很多时候发送到后台的报文可以设置一些特殊的格式,例如上文的004401:大家好,可以分解为userid:text,当然userid也可以是加密的一些报文,甚至可以学习其他报文一样设置加密区,这取决于大家的业务需要. （已更新解决方案）

后言

项目已经整合进开源项目spring-cloud-study的子模块spring-cloud-study-netty-websocket，作为对websocket体系的补充，对SpringBoot2.0集成WebSocket，实现后台向前端推送信息 的完善。

改造netty支持url参数

最新改造的项目代码已经上传，克服了使用url会导致连接断开的问题，详情请看spring-cloud-study

  1. 首先，调整一下加载handler的顺序，优先MyWebSocketHandler在WebSocketServerProtocolHandler之上。

    ch.pipeline().addLast(new MyWebSocketHandler());
    ch.pipeline().addLast(new WebSocketServerProtocolHandler(“/ws”, null, true, 65536 * 10));

  2. 其次，改造MyWebSocketHandler 的channelRead方法，首次连接会是一个FullHttpRequest类型，可以通过FullHttpRequest.uri()获取完整ws的URL地址，之后接受信息的话，会是一个TextWebSocketFrame类型。

    public class MyWebSocketHandler extends SimpleChannelInboundHandler { 

    1. @Override
    2. public void channelActive(ChannelHandlerContext ctx) throws Exception { 
    3.     System.out.println("与客户端建立连接，通道开启！");
    5.     //添加到channelGroup通道组
    6.     MyChannelHandlerPool.channelGroup.add(ctx.channel());
    7. }
    9. @Override
    10. public void channelInactive(ChannelHandlerContext ctx) throws Exception { 
    11.     System.out.println("与客户端断开连接，通道关闭！");
    12.     //添加到channelGroup 通道组
    13.     MyChannelHandlerPool.channelGroup.remove(ctx.channel());
    14. }
    16. @Override
    17. public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
    18.     //首次连接是FullHttpRequest，处理参数 by zhengkai.blog.csdn.net
    19.     if (null != msg && msg instanceof FullHttpRequest) { 
    20.         FullHttpRequest request = (FullHttpRequest) msg;
    21.         String uri = request.uri();
    23.         Map paramMap=getUrlParams(uri);
    24.         System.out.println("接收到的参数是："+JSON.toJSONString(paramMap));
    25.         //如果url包含参数，需要处理
    26.         if(uri.contains("?")){ 
    27.             String newUri=uri.substring(0,uri.indexOf("?"));
    28.             System.out.println(newUri);
    29.             request.setUri(newUri);
    30.         }
    32.     }else if(msg instanceof TextWebSocketFrame){ 
    33.         //正常的TEXT消息类型
    34.         TextWebSocketFrame frame=(TextWebSocketFrame)msg;
    35.         System.out.println("客户端收到服务器数据：" +frame.text());
    36.         sendAllMessage(frame.text());
    37.     }
    38.     super.channelRead(ctx, msg);
    39. }
    41. @Override
    42. protected void channelRead0(ChannelHandlerContext channelHandlerContext, TextWebSocketFrame textWebSocketFrame) throws Exception { 
    44. }
    46. private void sendAllMessage(String message){ 
    47.     //收到信息后，群发给所有channel
    48.     MyChannelHandlerPool.channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    49. }
    51. private static Map getUrlParams(String url){ 
    52.     Map<String,String> map = new HashMap<>();
    53.     url = url.replace("?",";");
    54.     if (!url.contains(";")){ 
    55.         return map;
    56.     }
    57.     if (url.split(";").length > 0){ 
    58.         String[] arr = url.split(";")[1].split("&");
    59.         for (String s : arr){ 
    60.             String key = s.split("=")[0];
    61.             String value = s.split("=")[1];
    62.             map.put(key,value);
    63.         }
    64.         return  map;
    66.     }else{ 
    67.         return map;
    68.     }
    69. }

    }

  3. html中的ws地址也进行改造

    socket = new WebSocket(“ws://127.0.0.1:12345/ws?uid=666&gid=777”);

  4. 改造后控制台输出情况

    收到新连接
    与客户端建立连接，通道开启！
    接收到的参数是：{ “uid”:”666”,”gid”:”777”}
    /ws
    客户端收到服务器数据：142531:这里输入消息
    客户端收到服务器数据：142531:这里输入消息
    客户端收到服务器数据：142531:这里输入消息

failed: WebSocket opening handshake timed out

听说是ssl wss的情况下才会出现，来自 @around-gao 的解决方法：

把MyWebSocketHandler和WebSocketServerProtocolHandler调下顺序就好了。

发表评论

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

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

相关阅读