nio与netty

﹏ヽ暗。殇╰゛Y 2022-04-15 05:28 342阅读 0赞

IO分为三种

  1. BIO(传统IO) 同步阻塞IO
  2. NIO同步非阻塞IO
  3. AIO异步非阻塞IO（涉及linux网络底层，太难，这里不讨论）

阻塞和非阻塞区别：
传统IO是阻塞的，因为read(),write()方法就是阻塞的，此时必须等读完了或者写完了才能进行后面的操作。NIO是非阻塞的，才用多路复用技术，把read()和write()单独给了一个线程，由操作系统完成，所以操作系统读完了或者写完了只需要给我们代码发个消息，代码直接去拿结果，中间不用等待，可以做自己的事。

原生NIO代码

  1. package cn.nio;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectableChannel;
  7. import java.nio.channels.SelectionKey;
  8. import java.nio.channels.Selector;
  9. import java.nio.channels.ServerSocketChannel;
  10. import java.nio.channels.SocketChannel;
  11. import java.util.Iterator;
  13. public class NIOService {
  15.     private Selector selector;
  17.     public void initService() throws IOException{
  18.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  19.         serverSocketChannel.configureBlocking(false);
  20.         serverSocketChannel.bind(new InetSocketAddress(12345));
  21.         this.selector = Selector.open();
  22.         serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
  23.         System.out.println("【星际联邦】：开启服务成功");
  24.     }
  26.     public void listenService() throws IOException{
  27.         while(true){
  28.             selector.select();
  29.             Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
  30.             while(iterator.hasNext()){
  31.                 SelectionKey selectionKey = iterator.next();
  32.                 iterator.remove();
  33.                 if (selectionKey.isAcceptable()) {
  34.                     ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();
  35.                     SocketChannel channel = server.accept();
  36.                     channel.configureBlocking(false);
  37.                     channel.write(ByteBuffer.wrap("【星际联邦】：您已成功连接星际联邦。".getBytes("UTF-8")));
  38.                     channel.register(this.selector, SelectionKey.OP_READ);
  39.                 }else if (selectionKey.isReadable()) {
  40.                     SocketChannel channel =(SocketChannel) selectionKey.channel();
  41.                     ByteBuffer byteBuffer = ByteBuffer.allocate(4096);
  42.                     channel.read(byteBuffer);
  43.                     byte[] data = byteBuffer.array();
  44.                     String message = new String(data).trim();
  45.                     System.out.println("【星际联邦】：接受到消息"+message);
  46.                     ByteBuffer responseBuffer = ByteBuffer.wrap(new String("【星际联邦】：星际联邦已派企业号前往").getBytes("UTF-8"));
  47.                     channel.write(responseBuffer);
  48.                 }
  49.             }
  50.         }
  51.     }
  53. }
  55. public class StartService {
  57.     public static void main(String[] args) {
  58.         NIOService service = new NIOService();
  59.         try {
  60.             service.initService();
  61.             service.listenService();
  62.         } catch (IOException e) {
  63.             e.printStackTrace();
  64.         }
  65.     }
  67. }
  69. package cn.nio;
  71. import java.io.IOException;
  72. import java.net.InetSocketAddress;
  73. import java.nio.ByteBuffer;
  74. import java.nio.channels.SelectionKey;
  75. import java.nio.channels.Selector;
  76. import java.nio.channels.SocketChannel;
  77. import java.util.Iterator;
  79. public class NIOClient {
  81.     private Selector selector;
  83.     public void initClient() throws IOException{
  84.         SocketChannel channel = SocketChannel.open();
  85.         channel.configureBlocking(false);
  86.         selector = Selector.open();
  87.         channel.connect(new InetSocketAddress("127.0.0.1", 12345));
  88.         channel.register(selector,SelectionKey.OP_CONNECT);
  89.         System.out.println("客户端启动成功");
  90.     }
  92.     public void listenClient() throws IOException{
  93.         while(true){
  94.             selector.select();
  95.             Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  96.             while(iterator.hasNext()){
  97.                 SelectionKey key = iterator.next();
  98.                 iterator.remove();
  99.                 if (key.isConnectable()) {
  100.                     SocketChannel socketChannel = (SocketChannel) key.channel();
  101.                     if (socketChannel.isConnectionPending()) {
  102.                         socketChannel.finishConnect();
  103.                     }
  104.                     socketChannel.configureBlocking(false);
  105.                     socketChannel.write(ByteBuffer.wrap(new String("hello word").getBytes("UTF-8")));
  106.                     socketChannel.register(selector, SelectionKey.OP_READ);
  107.                 }else if (key.isReadable()) {
  108.                     SocketChannel socketChannel = (SocketChannel) key.channel();
  109.                     ByteBuffer byteBuffer = ByteBuffer.allocate(4096);
  110.                     socketChannel.read(byteBuffer);
  111.                     byte[] date = byteBuffer.array();
  112.                     String message = new String(date).trim();
  113.                     System.out.println("【客户端收到的消息为】:"+message);
  114.                     //ByteBuffer outBuffer = ByteBuffer.wrap(message.getBytes("utf-8"));
  115.                     //socketChannel.write(outBuffer);// 将消息回送给客户端
  116.                 }
  117.             }
  118.         }
  119.     }
  120. }
  122. package cn.nio;
  124. import java.io.IOException;
  126. public class StartClinet {
  128.     public static void main(String[] args) {
  129.         NIOClient client = new NIOClient();
  130.         try {
  131.             client.initClient();
  132.             client.listenClient();
  133.         } catch (IOException e) {
  134.             // TODO Auto-generated catch block
  135.             e.printStackTrace();
  136.         }
  138.     }
  140. }

由于每次需要写这么多代码，于是一般使用第三方框架netty
netty注重点不在连接，而在我们的处理逻辑，逻辑一般写在handler

  1. package day01.netty;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.Channel;
  5. import io.netty.channel.ChannelFuture;
  6. import io.netty.channel.ChannelInitializer;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.nio.NioServerSocketChannel;
  10. public class EchoService {
  12.     public void initService() {
  13.         NioEventLoopGroup nioEventLoopGroup = null;
  14.         try {
  15.             //server端引导类
  16.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  17.             //连接池数据,netty自动弄的连接池
  18.             nioEventLoopGroup = new NioEventLoopGroup();
  19.             //装配bootstap
  20.             serverBootstrap.group(nioEventLoopGroup)
  21.             //通道类型为NioServerSocketChannel
  22.             .channel(NioServerSocketChannel.class)
  23.             //监听端口号
  24.             .localAddress(12345)
  25.                     .childHandler(new ChannelInitializer<Channel>() {
  26.                         @Override
  27.                         protected void initChannel(Channel channel) throws Exception {
  28.                             //全是模板代码，主要是这里绑定handler
  29.                             channel.pipeline().addLast(new EchoServiceHandler());
  30.                         }
  31.                     });
  32.             // 最后绑定服务器等待直到绑定完成，调用sync()方法会阻塞直到服务器完成绑定,然后服务器等待通道关闭，因为使用sync()，所以关闭操作也会被阻塞。
  33.             ChannelFuture channelFuture = serverBootstrap.bind().sync();
  34.             System.out.println("netty服务端启动成功");
  35.             channelFuture.channel().closeFuture().sync();
  36.         } catch (Exception e) {
  37.             e.printStackTrace();
  38.         }finally {
  39.             try {
  40.                 nioEventLoopGroup.shutdownGracefully().sync();
  41.             } catch (InterruptedException e) {
  42.                 // TODO Auto-generated catch block
  43.                 e.printStackTrace();
  44.             }
  45.         }
  46.     }
  49.     public static void main(String[] args) {
  50.         EchoService echoService = new EchoService();
  51.         echoService.initService();
  52.     }
  53. }
  55. package day01.netty;
  57. import io.netty.buffer.ByteBuf;
  58. import io.netty.buffer.Unpooled;
  59. import io.netty.channel.ChannelHandlerContext;
  60. import io.netty.channel.ChannelInboundHandlerAdapter;
  62. public class EchoServiceHandler extends ChannelInboundHandlerAdapter {
  64.     /**
  65.      * 服务端获取请求的时候被调用
  66.      */
  67.     @Override
  68.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  69.         ByteBuf buf = (ByteBuf)msg;
  70.         byte[] req = new byte[buf.readableBytes()];
  71.         buf.readBytes(req);
  72.         String content = new String(req,"UTF-8");
  73.         System.out.println("netty服务端接受到的消息是："+content);
  74.         String rsp = "我是netty服务端，我已收到"+content;//响应内容
  75.         ByteBuf rspBuf = Unpooled.copiedBuffer(rsp.getBytes());
  76.         ctx.write(rspBuf);//响应给客户端
  77.     }
  79.     /**
  80.      * channelRead方法完成之后调用
  81.      */
  82.     @Override
  83.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  84.         ctx.flush();//flush之后才是真的发送了
  85.         System.out.println("netty服务端响应成功");
  86.     }
  88.     /**
  89.      * 发生异常时候被调用
  90.      */
  91.     @Override
  92.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  93.         cause.printStackTrace();
  94.         ctx.close();
  95.     }
  96. }
  98. package day01.netty;
  100. import java.net.InetSocketAddress;
  102. import io.netty.bootstrap.Bootstrap;
  103. import io.netty.channel.ChannelFuture;
  104. import io.netty.channel.ChannelInitializer;
  105. import io.netty.channel.nio.NioEventLoopGroup;
  106. import io.netty.channel.socket.SocketChannel;
  107. import io.netty.channel.socket.nio.NioSocketChannel;
  109. public class EchoClient {
  111.     private String ip;
  112.     private Integer port;
  114.     public EchoClient() {
  115.         super();
  116.     }
  118.     public EchoClient(String ip, Integer port) {
  119.         super();
  120.         this.ip = ip;
  121.         this.port = port;
  122.     }
  124.     public String getIp() {
  125.         return ip;
  126.     }
  128.     public void setIp(String ip) {
  129.         this.ip = ip;
  130.     }
  132.     public Integer getPort() {
  133.         return port;
  134.     }
  136.     public void setPort(Integer port) {
  137.         this.port = port;
  138.     }
  140.     public void initClient() throws Exception {
  141.         Bootstrap bootstrap = new Bootstrap();
  142.         NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup();
  143.         bootstrap.group(nioEventLoopGroup).channel(NioSocketChannel.class).remoteAddress(new InetSocketAddress(ip, port)).handler(new ChannelInitializer<SocketChannel>() {//业务处理类
  144.             @Override
  145.             protected void initChannel(SocketChannel ch)
  146.                     throws Exception {
  147.                 ch.pipeline().addLast(new EchoClientHandler());//注册handler
  148.             }
  149.         });
  150.         ChannelFuture channelFuture = bootstrap.connect().sync();
  151.         channelFuture.channel().closeFuture().sync();
  152.         nioEventLoopGroup.shutdownGracefully().sync();
  153.     }
  156.     public static void main(String[] args) throws Exception {
  157.         EchoClient client = new EchoClient("127.0.0.1", 12345);
  158.         client.initClient();
  159.     }
  160. }
  162. package day01.netty;
  164. import io.netty.buffer.ByteBuf;
  165. import io.netty.buffer.Unpooled;
  166. import io.netty.channel.ChannelHandlerContext;
  167. import io.netty.channel.SimpleChannelInboundHandler;
  169. public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf>{
  171.     /**
  172.      * 客户端连接服务器后被调用
  173.      * 发送请求
  174.      */
  175.     @Override
  176.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  177.         byte[] req = "hello word".getBytes();//消息
  178.         ByteBuf buf = Unpooled.buffer(req.length);//把消息放入缓冲里面
  179.         buf.writeBytes(req);//发送消息
  180.         ctx.writeAndFlush(buf);//flush之后才是发送过去了
  181.         System.out.println("netty客户端发送请求成功hello word");
  182.     }
  184.     /**
  185.      * 服务端给客户端发消息时被调用
  186.      * 获得响应
  187.      */
  188.     @Override
  189.     protected void channelRead0(ChannelHandlerContext chx, ByteBuf msg) throws Exception {
  190.         ByteBuf buf = (ByteBuf) msg;
  191.         byte[] req = new byte[buf.readableBytes()];
  192.         buf.readBytes(req);
  193.         String rsp = new String(req, "UTF-8");
  194.         System.out.println("netty客户端获得响应，内容是："+rsp);
  195.     }
  197.     /**
  198.      * 发生异常后被调用
  199.      */
  200.     @Override
  201.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  202.         cause.printStackTrace();
  203.         ctx.close();
  204.     }
  206. }

发表评论

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

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

相关阅读