WebSocket长连接

简介

  1. WebSocket 是 HTML5 中的协议，是构建在 HTTP 协议之上的一个网络通信协议，其以长连接的方式实现了客户端与服务端的全双工通信。HTTP/1.1 版本协议中具有 keep-alive 属性，实现的是半双工通信。

握手原理

Netty高级应用实战篇(上) - 图1

  • Clinet端生产一个随机的Key，保存到浏览器和请求头中。
  • Clinet发送请求给Server端。
  • Server从请求头属性中判断出这事一个ws（WebSocket）请求，读到请求头中的key属性。
  • Server对key加密，放到响应头accept属性中。
  • Server将响应回应给Clinet端。
  • Clinet端从头中判断是一个ws（WebSocket）响应，读取头中的accpet属性。
  • Clinet端进行解密后，将结果保存在浏览中的key进行比较，如果匹配，则连接。否则不连接。

场景需求分析

  1. 在页面上有两个左右并排的文本域，它们的中间有一个“发送”按钮。在左侧文本域中输入文本内容后，单击发送按钮，会显示到右侧文本域中。

客户端页面定义

  1. 在 src/main 下定义一个目录 webapp。在其中定义 html 页面。
  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8">
  5.     <title>index</title>
  6. </head>
  7. <script type="text/javascript">
  9.     // 当前页面一打开就会执行的代码
  10.     var socket;
  12.     if(window.WebSocket) {
  13.         // 创建一个WebSocket连接
  14.         socket = new WebSocket("ws://localhost:8888/gc");
  16.         // 当与服务端的ws连接创建成功后会触发onopen的执行
  17.         socket.onopen = function (ev) {
  18.             // 在右侧文本域中显示连接建立提示
  19.             var ta = document.getElementById("responseText");
  20.             ta.value = "连接已建立";
  21.         }
  23.         // 当接收到服务端发送的消息时会触发onmessage的执行
  24.         socket.onmessage = function (ev) {
  25.             // 将服务端发送来的消息在右侧文本域中显示，在原有内容基础上进行拼接
  26.             var ta = document.getElementById("responseText");
  27.             ta.value = ta.value + "\n" + ev.data;
  28.         }
  30.         // 当与服务端的ws连接断开时会触发onclose的执行
  31.         socket.onclose = function (ev) {
  32.             // 将连接关闭消息在右侧文本域中显示，在原有内容基础上进行拼接
  33.             var ta = document.getElementById("responseText");
  34.             ta.value = ta.value + "\n连接已关闭";
  35.         }
  36.     } else {
  37.         alert("浏览器不支持WebSocket");
  38.     }
  40.     // 定义发送按钮的发送方法
  41.     function send(msg) {
  42.         // 若当前浏览器不支持WebSocket，则直接结束
  43.         if(!window.WebSocket) return;
  45.         // 若ws连接已打开，则向服务器发送消息
  46.         if(socket.readyState == WebSocket.OPEN) {
  47.             // 通过ws连接向服务器发送消息
  48.             socket.send(msg);
  49.         }
  50.     }
  51. </script>
  52. <body>
  53.     <form>
  54.         <textarea id="message" style="width: 150px; height: 150px"></textarea>
  55.         <input type="button" value="发送" onclick="send(this.form.message.value)">
  56.         <textarea id="responseText" style="width: 150px; height: 150px"></textarea>
  57.     </form>
  58. </body>
  59. </html>

服务端定义

  1. package com.gc.socket.server;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.Unpooled;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. import io.netty.handler.codec.DelimiterBasedFrameDecoder;
  11. import io.netty.handler.codec.FixedLengthFrameDecoder;
  12. import io.netty.handler.codec.LineBasedFrameDecoder;
  13. import io.netty.handler.codec.http.HttpObjectAggregator;
  14. import io.netty.handler.codec.http.HttpServerCodec;
  15. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
  16. import io.netty.handler.codec.string.StringDecoder;
  17. import io.netty.handler.codec.string.StringEncoder;
  18. import io.netty.handler.logging.LogLevel;
  19. import io.netty.handler.logging.LoggingHandler;
  20. import io.netty.handler.stream.ChunkedWriteHandler;
  21. import io.netty.util.CharsetUtil;
  23. /**
  24.  * @description:
  25.  * @author: GC
  26.  * @create: 2020-11-10 16:11
  27.  **/
  28. public class Server {
  29.     public static void main(String[] args) {
  31.         //定义用于处理客户端连接的EventLoopGroup
  32.         EventLoopGroup parentEventLoop = new NioEventLoopGroup();
  33.         //定义用于处理客户端请求的EventLoopGroup
  34.         EventLoopGroup childEventLoop = new NioEventLoopGroup();
  35.         try{
  37.             //服务端专用Bootstrap.主要用于将属绑定起来。建立关系
  38.             ServerBootstrap bootstrap = new ServerBootstrap();
  39.             //绑定处理客户端的EventLoopGroup对象
  40.             bootstrap.group(parentEventLoop, childEventLoop)
  41.                     //使用NIO的方式
  42.                     .channel(NioServerSocketChannel.class)
  43.                     .handler(new LoggingHandler(LogLevel.ERROR))
  44.                     //绑定消息收发时的编码/解码器。 和自定义的Handler绑定
  45.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  46.                         @Override
  47.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  48.                             ChannelPipeline pipeline = socketChannel.pipeline();
  49.                             //HttpRequestDecoder和HttpResponseEncoder的组合使
  50.                             pipeline.addLast(new HttpServerCodec());
  51.                             //可以毫无困难地发送大型数据流。
  52.                             pipeline.addLast(new ChunkedWriteHandler());
  53.                             //聚合处理器，聚合请求或者响应
  54.                             pipeline.addLast(new HttpObjectAggregator(111));
  55.                             //处理程序为你运行一个websocket服务器做了所有繁重的工作。
  56.                             // 它负责websocket握手以及控制框架的处理（Close，Ping，Pong）。
  57.                             // 文本和二进制数据帧被传递到管道中的下一个处理程序（由您执行）进行处理。
  58.                             pipeline.addLast(new WebSocketServerProtocolHandler("/gc"));
  59.                             //自定义业务处理器
  60.                             pipeline.addLast(new ServerHandler());
  61.                         }
  62.                     });
  63.             //声明服务端绑定的端口
  64.             ChannelFuture future = bootstrap.bind(8888).sync();
  65.             //当Channel调用了close()方法，并且成功关闭之后，才会调用此代码
  66.             future.channel().closeFuture().sync();
  67.         }catch (Exception e){
  68.             e.printStackTrace();
  69.         }finally {
  70.             //优雅的关闭方式
  71.             parentEventLoop.shutdownGracefully();
  72.             childEventLoop.shutdownGracefully();
  73.         }
  74.     }
  75. }

业务处理器定义

  1. package com.gc.socket.server;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  6. import java.util.Date;
  7. import java.util.Random;
  8. import java.util.UUID;
  9. import java.util.concurrent.TimeUnit;
  11. public class ServerHandler extends ChannelInboundHandlerAdapter {
  13.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  14.         String text = ((TextWebSocketFrame)msg).text();
  15.         ctx.channel().writeAndFlush(new TextWebSocketFrame("客户端："+text));
  16.         System.out.println("本地地址为:"+ctx.channel().localAddress());
  17.         System.out.println("远程地址为:"+ctx.channel().remoteAddress()+" ---> 收到消息:"+msg);
  18.     }
  20.     //当发生Throwable异常时，会调用该方法
  21.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  22.         ctx.fireExceptionCaught(cause);
  23.         ctx.close();
  24.     }
  26. }

Socket实现群聊

简介

  1. 本例要实现一个网络群聊工具。参与聊天的客户端消息是通过服务端进行广播的。

服务端定义

  1. package com.gc.socket.server;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.buffer.ByteBuf;
  5. import io.netty.buffer.Unpooled;
  6. import io.netty.channel.*;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. import io.netty.handler.codec.DelimiterBasedFrameDecoder;
  11. import io.netty.handler.codec.FixedLengthFrameDecoder;
  12. import io.netty.handler.codec.LineBasedFrameDecoder;
  13. import io.netty.handler.codec.http.HttpObjectAggregator;
  14. import io.netty.handler.codec.http.HttpServerCodec;
  15. import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
  16. import io.netty.handler.codec.string.StringDecoder;
  17. import io.netty.handler.codec.string.StringEncoder;
  18. import io.netty.handler.logging.LogLevel;
  19. import io.netty.handler.logging.LoggingHandler;
  20. import io.netty.handler.stream.ChunkedWriteHandler;
  21. import io.netty.util.CharsetUtil;
  23. /**
  24.  * @description:
  25.  * @author: GC
  26.  * @create: 2020-11-10 16:11
  27.  **/
  28. public class Server {
  29.     public static void main(String[] args) {
  31.         //定义用于处理客户端连接的EventLoopGroup
  32.         EventLoopGroup parentEventLoop = new NioEventLoopGroup();
  33.         //定义用于处理客户端请求的EventLoopGroup
  34.         EventLoopGroup childEventLoop = new NioEventLoopGroup();
  35.         try{
  37.             //服务端专用Bootstrap.主要用于将属绑定起来。建立关系
  38.             ServerBootstrap bootstrap = new ServerBootstrap();
  39.             //绑定处理客户端的EventLoopGroup对象
  40.             bootstrap.group(parentEventLoop, childEventLoop)
  41.                     //使用NIO的方式
  42.                     .channel(NioServerSocketChannel.class)
  43.                     .handler(new LoggingHandler(LogLevel.ERROR))
  44.                     //绑定消息收发时的编码/解码器。 和自定义的Handler绑定
  45.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  46.                         @Override
  47.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  48.                             ChannelPipeline pipeline = socketChannel.pipeline();
  50.                             //因为要传输Clinet端的消息，所以肯定是先收到消息，所以解码器放前面
  51.                             pipeline.addLast(new LineBasedFrameDecoder(2024));
  52.                             pipeline.addLast(new StringDecoder());
  53.                             pipeline.addLast(new StringEncoder());
  54.                             //自定义业务处理器
  55.                             pipeline.addLast(new ServerHandler());
  56.                         }
  57.                     });
  58.             //声明服务端绑定的端口
  59.             ChannelFuture future = bootstrap.bind(8888).sync();
  60.             System.out.println("Server 启动成功");
  61.             //当Channel调用了close()方法，并且成功关闭之后，才会调用此代码
  62.             future.channel().closeFuture().sync();
  63.         }catch (Exception e){
  64.             e.printStackTrace();
  65.         }finally {
  66.             //优雅的关闭方式
  67.             parentEventLoop.shutdownGracefully();
  68.             childEventLoop.shutdownGracefully();
  69.         }
  70.     }
  71. }

服务端业务处理器定义

  1. package com.gc.socket.server;
  3. import io.netty.channel.Channel;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.channel.group.ChannelGroup;
  7. import io.netty.channel.group.DefaultChannelGroup;
  8. import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
  9. import io.netty.util.concurrent.EventExecutor;
  10. import io.netty.util.concurrent.GlobalEventExecutor;
  12. import java.util.Date;
  13. import java.util.Random;
  14. import java.util.UUID;
  15. import java.util.concurrent.TimeUnit;
  17. public class ServerHandler extends ChannelInboundHandlerAdapter {
  20.     //消息组，聊天的人都在这个组里面。类似社交软件的群聊
  21.     private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  23.     /**
  24.      * 读取到发送的消息
  25.      * @param ctx
  26.      * @param msg
  27.      * @throws Exception
  28.      */
  29.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  30.         Channel channel = ctx.channel();
  31.         channelGroup.forEach(ch -> {
  32.             if(ch != channel)//不是自己，显示别人的地址
  33.                 ch.writeAndFlush(channel.remoteAddress()+":"+msg+"\n");
  34.             else//是自己，不需要显示
  35.                 ch.writeAndFlush(msg+"\n");
  37.         });
  38.     }
  40.     /**
  41.      * 用户上线，会触发此方法
  42.      * @param ctx
  43.      * @throws Exception
  44.      */
  45.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  46.         Channel channel = ctx.channel();
  47.         channelGroup.writeAndFlush(channel.remoteAddress()+"上线"+"\n");
  48.         channelGroup.add(channel);
  49.     }
  51.     /**
  52.      * 用户下线，会触发此犯法
  53.      * @param ctx
  54.      * @throws Exception
  55.      */
  56.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  57.         Channel channel = ctx.channel();
  58.         channelGroup.writeAndFlush(channel.remoteAddress()+"下线。当前在线人数:"+channelGroup.size()+"\n");
  59.         //当触发此方法时，会自动从channelGroup中删除。以下代码是当channel在线时，将它踢出channelGroup
  60.         //channelGroup.remove(channel);
  61.     }
  63.     //当发生Throwable异常时，会调用该方法
  64.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  65.         ctx.fireExceptionCaught(cause);
  66.         ctx.close();
  67.     }
  69. }

客户端定义

  1. package com.gc.socket.clinet;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.FixedLengthFrameDecoder;
  9. import io.netty.handler.codec.LineBasedFrameDecoder;
  10. import io.netty.handler.codec.string.StringDecoder;
  11. import io.netty.handler.codec.string.StringEncoder;
  12. import io.netty.util.CharsetUtil;
  14. import java.io.*;
  15. import java.net.SocketAddress;
  17. public class Clinet {
  19.     public static void main(String[] args) throws Exception {
  20.         //客户端处理服务端的EventLoopGroup
  21.         EventLoopGroup clint = new NioEventLoopGroup();
  22.         //Bootstrap专门用于客户端的属性设置以及关系绑定
  23.         Bootstrap bootstrap = new Bootstrap();
  24.         bootstrap.group(clint)
  25.                 //使用NIO客户端的Channel
  26.                 .channel(NioSocketChannel.class)
  27.                 //绑定编码器/解码器。以及自定义的Handler
  28.                 .handler(new ChannelInitializer<SocketChannel>(){
  29.                     @Override
  30.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
  31.                         ChannelPipeline pipeline = socketChannel.pipeline();
  32.                         //当上线后，Client端会收到Server端的提示信息，所以这里解码器放前面
  33.                         pipeline.addLast(new LineBasedFrameDecoder(2048));
  34.                         pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
  35.                         pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
  36.                         pipeline.addLast(new ClinetHandler());
  37.                     }
  38.                 });
  39.         //指定连接服务端的端口
  40.         ChannelFuture future = bootstrap.connect("127.0.0.1",8888).sync();
  41.         System.out.println("Clinet 启动成功");
  43.         Channel channel = future.channel();
  44.         InputStream in;
  45.         InputStreamReader is = new InputStreamReader(System.in , "UTF-8");
  46.         BufferedReader br = new BufferedReader(is);
  47.         while (true) {
  48.             channel.writeAndFlush(br.readLine()+"\n");
  49.         }
  50.         //这里不关闭资源，因为客户端要持续输入。
  51.     }
  52. }

客户端业务处理器定义

  1. package com.gc.socket.clinet;
  3. import io.netty.channel.ChannelHandlerContext;
  4. import io.netty.channel.ChannelInboundHandlerAdapter;
  5. import io.netty.channel.SimpleChannelInboundHandler;
  7. import java.util.UUID;
  8. import java.util.concurrent.TimeUnit;
  10. /**
  11.  * ChannelInboundHandlerAdapter 类中的 channelRead方法不会自动释放资源。所以会不断的发送消息
  12.  * SimpleChannelInboundHandler 类中的 channelRead0()方法不会自动释放资源。所以只要Clinet端触发了消息发送， 双方两段就会一直死循环发送消息。
  13.  *
  14.  */
  15. public class ClinetHandler extends SimpleChannelInboundHandler<String> {
  18.     /**
  19.      * 读取服务端发送的消息
  20.      * @param ctx
  21.      * @param msg
  22.      * @throws Exception
  23.      */
  24.     public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  25.         System.err.println("Clinet:"+msg);
  26.     }
  28.     //当发生异常，捕获并关闭
  29.     @Override
  30.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  31.         cause.printStackTrace();
  32.         ctx.close();
  33.     }
  35. }

读写空闲监测

服务端修改

Netty高级应用实战篇(上) - 图2

服务端业务处理器修改

Netty高级应用实战篇(上) - 图3

心跳机制

简介

  1. 所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还“活着”, 以确保 TCP 连接的有效性。

客户端定义

  1. package com.gc.socket.clinet;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioSocketChannel;
  8. import io.netty.handler.codec.FixedLengthFrameDecoder;
  9. import io.netty.handler.codec.LineBasedFrameDecoder;
  10. import io.netty.handler.codec.string.StringDecoder;
  11. import io.netty.handler.codec.string.StringEncoder;
  12. import io.netty.util.CharsetUtil;
  14. import java.io.*;
  15. import java.net.SocketAddress;
  17. public class Clinet {
  19.     public static void main(String[] args) throws Exception {
  20.         //客户端处理服务端的EventLoopGroup
  21.         EventLoopGroup clint = new NioEventLoopGroup();
  22.         //Bootstrap专门用于客户端的属性设置以及关系绑定
  23.         Bootstrap bootstrap = new Bootstrap();
  24.         bootstrap.group(clint)
  25.                 //使用NIO客户端的Channel
  26.                 .channel(NioSocketChannel.class)
  27.                 //绑定编码器/解码器。以及自定义的Handler
  28.                 .handler(new ChannelInitializer<SocketChannel>(){
  29.                     @Override
  30.                     protected void initChannel(SocketChannel socketChannel) throws Exception {
  31.                         ChannelPipeline pipeline = socketChannel.pipeline();
  32.                         //当上线后，Client端会收到Server端的提示信息，所以这里解码器放前面
  33.                         pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
  34.                         pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
  35.                         pipeline.addLast(new ClinetHandler(bootstrap));
  36.                     }
  37.                 });
  38.         //指定连接服务端的端口
  39.         ChannelFuture future = bootstrap.connect("127.0.0.1",8888).sync();
  40.         System.out.println("Clinet 启动成功");
  41.     }
  42. }

客户端业务处理器

  1. package com.gc.socket.clinet;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.Channel;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import io.netty.util.concurrent.GenericFutureListener;
  9. import io.netty.util.concurrent.ScheduledFuture;
  11. import java.util.Random;
  12. import java.util.concurrent.TimeUnit;
  14. /**
  15.  * ChannelInboundHandlerAdapter 类中的 channelRead方法不会自动释放资源。所以会不断的发送消息
  16.  * SimpleChannelInboundHandler 类中的 channelRead0()方法不会自动释放资源。所以只要Clinet端触发了消息发送， 双方两段就会一直死循环发送消息。
  17.  *
  18.  */
  19. public class ClinetHandler extends ChannelInboundHandlerAdapter {
  22.     //用于添加任务定时触发
  23.     private ScheduledFuture<?> scheduled;
  24.     //用于监听任务过期
  25.     private GenericFutureListener listener;
  26.     //用于客户端断链重连
  27.     private Bootstrap bootstrap;
  30.     public ClinetHandler(Bootstrap bootstrap){
  31.         this.bootstrap = bootstrap;
  32.     }
  34.     /**
  35.      * 用户上线后，触发此方法
  36.      * @param ctx
  37.      * @throws Exception
  38.      */
  39.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  40.         setRandom(ctx.channel());
  41.     }
  43.     /**
  44.      * 当用户下线后，触发此方法
  45.      * @param ctx
  46.      * @throws Exception
  47.      */
  48.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  49.         //下线后,删除监听任务。 否则数据量大了，可能会导致OOM
  50.         scheduled.removeListener(listener);
  52.         System.out.println("已断链, 准备重连。");
  53.         bootstrap.connect("localhost", 8888).sync();
  54.     }
  56.     /**
  57.      * 心跳监测时间
  58.      * @throws Exception
  59.      */
  60.     public void setRandom(Channel ctx) throws Exception {
  61.         //随机下次像Server端发送心跳的时间
  62.         int random = new Random().nextInt(5)+1;
  63.         System.out.println(random+"S后发送心跳");
  65.         scheduled = ctx.eventLoop().schedule(()->{
  66.             //当前channel处于活动状态, 说明没死，发送心跳监测
  67.             if(ctx.isActive()){
  68.                 System.out.println("向服务端发送心跳");
  69.                 ctx.writeAndFlush("ping");
  70.             }else{
  71.                 //已经死了
  72.                 System.out.println("心跳超时,连接断开");
  73.             }
  74.         },random, TimeUnit.SECONDS);
  76.         //给当前方法定义监听器
  77.         listener = future1 -> {
  78.             setRandom(ctx);
  79.         };
  81.         //添加一个定时监听
  82.         scheduled.addListener(listener);
  83.     }
  86.     //当发生异常，捕获并关闭
  87.     @Override
  88.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  89.         cause.printStackTrace();
  90.         ctx.close();
  91.     }
  93. }

服务端定义

  1. package com.gc.socket.server;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.handler.codec.string.StringDecoder;
  9. import io.netty.handler.codec.string.StringEncoder;
  10. import io.netty.handler.logging.LogLevel;
  11. import io.netty.handler.logging.LoggingHandler;
  12. import io.netty.handler.timeout.IdleStateHandler;
  14. /**
  15.  * @description:
  16.  * @author: GC
  17.  * @create: 2020-11-10 16:11
  18.  **/
  19. public class Server {
  20.     public static void main(String[] args) {
  22.         //定义用于处理客户端连接的EventLoopGroup
  23.         EventLoopGroup parentEventLoop = new NioEventLoopGroup();
  24.         //定义用于处理客户端请求的EventLoopGroup
  25.         EventLoopGroup childEventLoop = new NioEventLoopGroup();
  26.         try{
  28.             //服务端专用Bootstrap.主要用于将属绑定起来。建立关系
  29.             ServerBootstrap bootstrap = new ServerBootstrap();
  30.             //绑定处理客户端的EventLoopGroup对象
  31.             bootstrap.group(parentEventLoop, childEventLoop)
  32.                     //使用NIO的方式
  33.                     .channel(NioServerSocketChannel.class)
  34.                     .handler(new LoggingHandler(LogLevel.ERROR))
  35.                     //绑定消息收发时的编码/解码器。 和自定义的Handler绑定
  36.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  37.                         @Override
  38.                         protected void initChannel(SocketChannel socketChannel) throws Exception {
  39.                             ChannelPipeline pipeline = socketChannel.pipeline();
  40.                             pipeline.addLast(new StringDecoder());
  41.                             pipeline.addLast(new StringEncoder());
  42.                             /**
  43.                              * @prarm1：在指定的时间内，服务端没发生读操作（时间单位：S）
  44.                              * @prarm2：在指定的时间内，服务端没发生写操作（时间单位：S）
  45.                              * @prarm3：在指定的时间内，服务端即要发生读操作，也要发生写操作。如果只出现了两个操作中的一个，时间到达时，还是会被连接中断（时间单位：S）
  46.                              */
  47.                             pipeline.addLast(new IdleStateHandler(3,0,0));
  48.                             //自定义业务处理器
  49.                             pipeline.addLast(new ServerHandler());
  50.                         }
  51.                     });
  52.             //声明服务端绑定的端口
  53.             ChannelFuture future = bootstrap.bind(8888).sync();
  54.             System.out.println("Server 启动成功");
  55.             //当Channel调用了close()方法，并且成功关闭之后，才会调用此代码
  56.             future.channel().closeFuture().sync();
  57.         }catch (Exception e){
  58.             e.printStackTrace();
  59.         }finally {
  60.             //优雅的关闭方式
  61.             parentEventLoop.shutdownGracefully();
  62.             childEventLoop.shutdownGracefully();
  63.         }
  64.     }
  65. }

服务端业务处理器定义

  1. package com.gc.socket.server;
  3. import io.netty.channel.Channel;
  4. import io.netty.channel.ChannelHandlerContext;
  5. import io.netty.channel.ChannelInboundHandlerAdapter;
  6. import io.netty.handler.timeout.IdleState;
  7. import io.netty.handler.timeout.IdleStateEvent;
  9. public class ServerHandler extends ChannelInboundHandlerAdapter {
  12.     /**
  13.      * 读取客户端发送的消息
  14.      * @param ctx
  15.      * @param msg
  16.      * @throws Exception
  17.      */
  18.     @Override
  19.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  20.         System.out.println("接收到Client发送的消息：" + msg);
  21.     }
  23.     /**
  24.      * 异常捕捉
  25.      * @param ctx
  26.      * @param cause
  27.      * @throws Exception
  28.      */
  29.     @Override
  30.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  31.         ctx.close();
  32.     }
  34.     /**
  35.      * 用于捕获当前Server中的各种事件
  36.      * @param ctx
  37.      * @param evt
  38.      * @throws Exception
  39.      */
  40.     @Override
  41.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  42.         if (evt instanceof IdleStateEvent) {
  43.             IdleState state = ((IdleStateEvent) evt).state();
  44.             if (state == IdleState.READER_IDLE) {
  45.                 System.out.println("将要断开连接");
  46.                 ctx.close();
  47.             } else {
  48.                 super.userEventTriggered(ctx, evt);
  49.             }
  50.         }
  51.     }
  52. }

手写Tomcat

简介

  1. 确切地说，这里要手写的是一个 Web 容器，一个类似于 Tomcat 的容器，用于处理 HTTP请求。该Web 容器没有实现 JavaEE 的 Servlet 规范，不是一个 Servlet 容器。但其是模拟着Tomcat 来写的，这里定义了自己的请求、响应及 Servlet，分别命名为了 NettyRequest，NettyResponse 与 Servnet。
  3. 我们这里要定义一个 Tomcat，这个 Web 容器提供给用户后，用户只需要按照使用步骤就可以将其自定义的 Servnet 发布到该 Tomcat 中。我们现在给出用户对于该 Tomcat 的使用步骤：
  • 用户只需将自定义的 Servnet 放入到指定的包中。例如，com.abc.webapp 包中。
  • 用户在访问时，需要将自定义的 Servnet 的简单类名全小写后的字符串作为该 Servnet的 Name 进行访问。
  • 若没有指定的 Servnet，则访问默认的 Servnet。

思路定义

  • Servnet规范包

    1. NettyRequest
      包含了传统HttpServletRequest中的方法。如：获取请求路径，获取URL中的参数，获取方法名称，获取参数等等。

    2. NettyResponse
      包含了传统HttpServletResponse中的方法。如：读/写操作。

    3. Servnet
      和Servlet一样，有doPost()，doGet()

  • Tomcat内核包

    1. main()
      用于加载指定目录下的接口。

    2. DefaultNettyRequest，DefaultNettyResponse
      我们自定义NettyResponse，NettyResponse的具体实现。

    3. DefaultServnet
      我们自定义的Servnet 具体实现。

    4. TomcatHandler
      请求解析业务处理器，主要用于解析接收请求。

    5. TomcatServer
      当启动容器时，初始化资源到本地缓存。

  • webapp
    要访问的接口，都在该目录下。

项目结构

Netty高级应用实战篇(上) - 图4

编码实现

定义NettyRequest规范

  1. package com.gc.servnet;
  3. import com.sun.deploy.net.HttpRequest;
  5. import java.util.List;
  6. import java.util.Map;
  8. /**
  9.  * @description:
  10.  * @author: GC
  11.  * @create: 2020-11-13 15:56
  12.  **/
  13. public interface NettyRequest {
  15.     //获取URL中的参数
  16.     String getUri();
  18.     //获取方法名称
  19.     String getMethod();
  21.     //获取请求路径
  22.     String getPath();
  24.     //获取本次请求中所有的参数
  25.     Map<String, List<String>> getParameters();
  27.     //获取指定名称的参数值
  28.     List<String> getParameters(String name);
  30.     //指定参数名称获取参数值（第一个值）
  31.     String getParameter(String name);
  33. }

定义NettyResponse规范

  1. package com.gc.servnet;
  3. /**
  4.  * @description:
  5.  * @author: GC
  6.  * @create: 2020-11-13 15:56
  7.  **/
  8. public interface NettyResponse {
  10.     void write(String content) throws Exception;
  11. }

定义Servnet规范

  1. package com.gc.servnet;
  3. public abstract class Servnet {
  5.     //定义Servlet中的doGet
  6.     public abstract void doGet(NettyRequest request, NettyResponse response) throws Exception;
  8.     //定义Servlet中的doPost
  9.     public abstract void doPost(NettyRequest request, NettyResponse response) throws Exception;
  12. }

DefaultNettyRequest具体实现

  1. package com.gc.tomcat;
  3. import com.gc.servnet.NettyRequest;
  4. import io.netty.handler.codec.http.HttpRequest;
  5. import io.netty.handler.codec.http.QueryStringDecoder;
  7. import java.util.List;
  8. import java.util.Map;
  10. /**
  11.  * @description:
  12.  * @author: GC
  13.  * @create: 2020-11-13 16:25
  14.  **/
  15. public class DefaultNettyRequest implements NettyRequest {
  17.     HttpRequest request;
  19.     public DefaultNettyRequest(HttpRequest request){
  20.         this.request = request;
  21.     }
  23.     @Override
  24.     public String getUri() {
  25.         return request.uri();
  26.     }
  28.     @Override
  29.     public String getMethod() {
  30.         return request.method().name();
  31.     }
  33.     @Override
  34.     public String getPath() {
  35.         QueryStringDecoder decoder = new QueryStringDecoder(request.uri());
  36.         return decoder.path();
  37.     }
  39.     @Override
  40.     public Map<String, List<String>> getParameters() {
  41.         QueryStringDecoder decoder = new QueryStringDecoder(getUri());
  42.         return decoder.parameters();
  43.     }
  45.     private int count = 0;
  46.     @Override
  47.     public List<String> getParameters(String name) {
  48.         Map<String, List<String>> map = getParameters();
  49.         return map.get(name);
  50.     }
  52.     @Override
  53.     public String getParameter(String name) {
  54.         List<String> list = getParameters(name);
  55.         if(null == list){
  56.             System.out.println("要获取的参数不存在");
  57.             return null;
  58.         }
  59.         return list.get(0);
  60.     }
  61. }

DefaultNettyResponse具体实现

  1. package com.gc.tomcat;
  3. import com.gc.servnet.NettyResponse;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.handler.codec.http.*;
  7. import io.netty.util.internal.StringUtil;
  9. /**
  10.  * @description:
  11.  * @author: GC
  12.  * @create: 2020-11-13 16:34
  13.  **/
  14. public class DefaultNettyResponse implements NettyResponse {
  15.     private HttpRequest request;
  16.     private ChannelHandlerContext context;
  18.     public DefaultNettyResponse(HttpRequest request, ChannelHandlerContext context) {
  19.         this.request = request;
  20.         this.context = context;
  21.     }
  23.     @Override
  24.     public void write(String content) throws Exception {
  25.         // 处理content为空的情况
  26.         if (StringUtil.isNullOrEmpty(content)) {
  27.             return;
  28.         }
  29.         // 创建响应对象
  30.         FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
  31.                 HttpResponseStatus.OK,
  32.                 // 根据响应体内容大小为response对象分配存储空间
  33.                 Unpooled.wrappedBuffer(content.getBytes("UTF-8")));
  35.         // 获取响应头
  36.         HttpHeaders headers = response.headers();
  37.         // 设置响应体类型
  38.         headers.set(HttpHeaderNames.CONTENT_TYPE, "text/json");
  39.         // 设置响应体长度
  40.         headers.set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
  41.         // 设置缓存过期时间
  42.         headers.set(HttpHeaderNames.EXPIRES, 0);
  43.         // 若HTTP请求是长连接，则响应也使用长连接
  44.         if (HttpUtil.isKeepAlive(request)) {
  45.             headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
  46.         }
  47.         context.writeAndFlush(response);
  48.     }
  49. }

DefaultServnet具体实现

  1. package com.gc.tomcat;
  3. import com.gc.servnet.NettyRequest;
  4. import com.gc.servnet.NettyResponse;
  5. import com.gc.servnet.Servnet;
  7. /**
  8.  * @description:
  9.  * @author: GC
  10.  * @create: 2020-11-13 16:40
  11.  **/
  12. public class DefaultServnet extends Servnet {
  14.     //定义Servlet中的doGet
  15.     @Override
  16.     public void doGet(NettyRequest request, NettyResponse response) throws Exception {
  17.         String sernetName = request.getUri().split("/")[1];
  18.         response.write("404 - no this servnet : " + sernetName);
  19.     }
  22.     //定义Servlet中的doPost
  23.     @Override
  24.     public void doPost(NettyRequest request, NettyResponse response) throws Exception {
  25.         doGet(request, response);
  26.     }
  28. }

编写自定义TomcatHandler处理器

  1. package com.gc.tomcat;
  3. import com.gc.servnet.NettyRequest;
  4. import com.gc.servnet.NettyResponse;
  5. import com.gc.servnet.Servnet;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import io.netty.handler.codec.http.HttpRequest;
  10. import java.util.Map;
  12. /**
  13.  * @description:
  14.  * @author: GC
  15.  * @create: 2020-11-13 16:47
  16.  **/
  17. public class TomcatHandler extends ChannelInboundHandlerAdapter {
  19.     //启动时候加载到这个Map
  20.     private Map<String, Servnet> nameToServnetMap;
  22.     //如果代码被放射的方式或者其它的方式修改过，则重写加载
  23.     private Map<String, String> nameToClassNameMap;
  26.     //定义一个构造器，当Netty启动时候加载到Netty的Handler中被Netty管理
  27.     public TomcatHandler(Map<String, Servnet> nameToServnetMap, Map<String, String> nameToClassNameMap){
  28.         this.nameToServnetMap = nameToServnetMap;
  29.         this.nameToClassNameMap = nameToClassNameMap;
  30.     }
  34.     @Override
  35.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  36.         //如果是Http请求，则进入解析
  37.         if(msg instanceof HttpRequest){
  38.             HttpRequest request = (HttpRequest) msg;
  39.             //拿到Servnet名称
  40.             String serverName = request.uri().split("/")[1];
  41.             serverName = serverName.substring(0,serverName.indexOf("?"));
  42.             //获得客户端要访问的Servnet
  43.             Servnet servnet = new DefaultServnet();
  45.             //一级缓存是否有这个key
  46.             if(nameToServnetMap.containsKey(serverName)){
  47.                 //直接拿出去
  48.                 servnet = nameToServnetMap.get(serverName);
  49.             }else if(nameToClassNameMap.containsKey(serverName)){//一级缓存没有，二级缓存是否有这个key
  50.                 //双重检测，防止并发请求时，重复创建
  51.                 if(!nameToServnetMap.containsKey(serverName)) {
  52.                     synchronized (this) {
  53.                         if(!nameToServnetMap.containsKey(serverName)) {
  54.                             //有，获取到类的全路径限定路径
  55.                             String classPath = nameToClassNameMap.get(serverName);
  56.                             servnet = (Servnet) Class.forName(classPath).newInstance();
  57.                             nameToServnetMap.put(serverName, servnet);
  58.                         }
  59.                     }
  60.                 }
  61.             }
  63.             //开始根据路径处理请求
  64.             //用于获取当前请求中的参数，路径等信息
  65.             NettyRequest defultRequest = new DefaultNettyRequest(request);
  66.             //用于获取当前请求中的上下文数据
  67.             NettyResponse defultResponse = new DefaultNettyResponse(request, ctx);
  69.             // 根据不同的请求类型，调用servnet实例的不同方法
  70.             String methodName = request.method().name();
  71.             if (methodName.equalsIgnoreCase("GET")) {
  72.                 servnet.doGet(defultRequest, defultResponse);
  73.             } else if(methodName.equalsIgnoreCase("POST")) {
  74.                 servnet.doPost(defultRequest, defultResponse);
  75.             }
  77.             ctx.close();
  78.         }
  79.     }
  81.     @Override
  82.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  83.         cause.printStackTrace();
  84.         ctx.close();
  85.     }
  87. }

编写容器启动时初始化TomcatServer

  1. TomcatServerpackage com.gc.tomcat;
  3. import com.gc.servnet.Servnet;
  4. import io.netty.bootstrap.ServerBootstrap;
  5. import io.netty.channel.*;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.handler.codec.http.HttpServerCodec;
  9. import io.netty.handler.codec.string.StringDecoder;
  10. import io.netty.handler.codec.string.StringEncoder;
  12. import java.io.File;
  13. import java.io.UnsupportedEncodingException;
  14. import java.net.URL;
  15. import java.util.HashMap;
  16. import java.util.Map;
  17. import java.util.concurrent.ConcurrentHashMap;
  19. /**
  20.  * @description:
  21.  * @author: GC
  22.  * @create: 2020-11-13 17:24
  23.  **/
  24. public class TomcatServer {
  25.     // key为servnet的简单类名，value为对应servnet实例
  26.     private Map<String, Servnet> nameToServnetMap = new ConcurrentHashMap<>();
  27.     // key为servnet的简单类名，value为对应servnet类的全限定性类名
  28.     private Map<String, String> nameToClassNameMap = new HashMap<>();
  29.     //加载该路径下的类为Servnet
  30.     private String basePackage;
  32.     public TomcatServer(String basePackage) {
  33.         this.basePackage = basePackage;
  34.     }
  36.     // 启动tomcat
  37.     public void start() throws Exception {
  38.         // 加载指定包中的所有Servnet的类名
  39.         cacheClassName(basePackage);
  40.         //资源准备完毕，启动Netty
  41.         runServer();
  42.     }
  44.     private void cacheClassName(String basePackage) throws UnsupportedEncodingException {
  45.         //获取指定路径下的类
  46.         URL resource = this.getClass().getClassLoader().getResource(basePackage.replaceAll("\\.", "/"));
  47.         if(null == resource){
  48.             System.out.println("没有找到访问资源");
  49.             return;
  50.         }
  52.         //将取到的资源转成一个File
  53.         File dir = new File(resource.getPath());
  54.         //找出该路径下所有的类
  55.         for (File file : dir.listFiles()) {
  56.             if (file.isDirectory()) {
  57.                 // 若当前遍历的file为目录，则递归调用当前方法
  58.                 cacheClassName(basePackage + "." + file.getName());
  59.             } else if (file.getName().endsWith(".class")) {
  60.                 //读到了将文件后缀去掉
  61.                 String simpleClassName = file.getName().replace(".class", "").trim();
  62.                 nameToClassNameMap.put(simpleClassName.toLowerCase(), basePackage+"."+simpleClassName);
  63.             }
  64.         }
  65.         System.out.println(nameToClassNameMap);
  67.     }
  69.     private void runServer() {
  71.         EventLoopGroup parentLoopGroup = new NioEventLoopGroup();
  72.         EventLoopGroup childLoopGroup = new NioEventLoopGroup();
  73.         try {
  74.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  75.             serverBootstrap.group(parentLoopGroup, childLoopGroup)
  76.                     //当出现并发时，设置这个队列的上限长度
  77.                     .option(ChannelOption.SO_BACKLOG, 1024)
  78.                     // 指定是否启用心跳机制来检测长连接的存活性，即客户端的存活性
  79.                     .childOption(ChannelOption.SO_KEEPALIVE, true)
  80.                     .channel(NioServerSocketChannel.class)
  81.                     .childHandler(new ChannelInitializer() {
  82.                         @Override
  83.                         protected void initChannel(Channel channel) throws Exception {
  84.                             ChannelPipeline channelPipeline = channel.pipeline();
  85.                             channelPipeline.addLast(new HttpServerCodec());
  86.                             channelPipeline.addLast(new TomcatHandler(nameToServnetMap, nameToClassNameMap));
  87.                         }
  88.                     });
  90.             ChannelFuture future = serverBootstrap.bind(8888).sync();
  91.             System.out.println("启动成功");
  92.             future.channel().closeFuture().sync();
  93.         }catch (Exception e){
  94.             e.printStackTrace();
  95.         }finally {
  96.             parentLoopGroup.shutdownGracefully();
  97.             childLoopGroup.shutdownGracefully();
  98.         }
  99.     }
  102. }

编写需要访问接口（Servnet）

  1. package com.gc.webapp;
  3. import com.gc.servnet.NettyRequest;
  4. import com.gc.servnet.NettyResponse;
  5. import com.gc.servnet.Servnet;
  7. /**
  8.  * @description:
  9.  * @author: GC
  10.  * @create: 2020-11-13 17:44
  11.  **/
  12. public class GC1 extends Servnet {
  13.     @Override
  14.     public void doGet(NettyRequest request, NettyResponse response) throws Exception {
  15.         String uri = request.getUri();
  16.         String path = request.getPath();
  17.         String method = request.getMethod();
  18.         String name = request.getParameter("name");
  20.         String content = "uri = " + uri + "\n" +
  21.                 "path = " + path + "\n" +
  22.                 "method = " + method + "\n" +
  23.                 "param = " + name;
  24.         response.write(content);
  25.     }
  27.     @Override
  28.     public void doPost(NettyRequest request, NettyResponse response) throws Exception {
  29.         doGet(request, response);
  30.     }
  32. }

启动类

  1. package com.gc.tomcat;
  3. /**
  4.  * @description:
  5.  * @author: GC
  6.  * @create: 2020-11-13 16:24
  7.  **/
  8. public class RunTomcat {
  9.     public static void main(String[] args) throws Exception {
  10.         TomcatServer tomcatServer = new TomcatServer("com.gc.webapp");
  11.         tomcatServer.start();
  12.     }
  13. }

手写Tomcat总结

  1. 在写之前，我们把思路整理了一下，然后把思路落地。

Servnet规范包

  1. 在Servnet包下，我们把传统的Servlet的基本规则和常用的方法功能以接口的形式定义出来了。包括Request中的获取参数，方法名称，访问路径等基本的操作。
  3. 以及Response向客户端的写操作定义。
  5. 还有Servnet中的doGet和doPos，我们这里只定义具体的声明，把具体的业务逻辑实现交给用户自己灵活编写。

Tomcat核心包

  1. 这里面主要是重写各个接口和抽象类的方法，把它们的定义的方法变成我们自己的具体业务实现。
  1. 我们使用DefaultNettyRequest和DefaultNettyResponse实现了NettyRequest和NettyResponse，在这里我们把定义实现了具体的操作。
  2. 我们使用DefaultServnet继承了Servent，分别处理get货post请求。让用户可以在自己的Servent中灵活的做业务逻辑。

  3. 我们编写了TomcatHandler来用于接受客户端的请求解析并处理。在TomcatHandler类中我们具体做的事情：

    • 接受用户的http请求。
    • 从一级缓存中拿到用户要访问的Servnet实例，如果没有则去二级缓存拿到类的全路径限定名，使用反射机制来动态创建出一个Servnet实例提供给本次请求需要用到的实例，创建好后把实例放到一级缓存。值得一提的是，我们怕当客户端出现并发访问出现线程安全问题可能会导致Map中的Servnet实例被重复创建，所以使用了双重检查锁来预防这个问题。
    • 当Servnet实例创建好后，我们将组装Request和Response，其中Respone中我们传递了本次客户端请求的上下文，从该上下文中拿到相关信息组装成一个请求头。
    • 然后获取到请求方式，分别执行响应的方法。

  4. TomcatServer的主要作用有两点：

    • 当容器启动时，模仿真正的Tomcat加载指定目录下的类（我们也可以称为具体的接口地址），到Tomcat的二级缓存中。其中Map的Key为具体的Servnet实例名称，Value为Servent源文件的具体路径。

    • 第二件事是我们在这个类中定义了Netty的启动代码。主要定义了：

      1. 当出现并发时，队列可缓存的最大成功。
      2. 启用心跳机制来检测客户端的存活性，及时的释放链接，减轻服务器的负载压力。
      3. 定义了Http的编码/解码为一体的编解码器。以及实现我们自己的Handler。
  5. 最后，我们定义了启动方法main() 。在这里面我们指定了需要扫描加载成Servlet实例的具体包名路径。