Netty权威指南之NIO入门

水深无声 2023-10-16 21:48 89阅读 0赞

本章学习目标：

1、传统的同步阻塞式IO编程

2、基于NIO非阻塞式编程

3、基于NIO2.0异步非阻塞式编程

4、为什么使用NIO编程

5、为什么选择Netty

传统BIO编程

网络编程的基本模式是C/S模式，也就是两个进程之间的相互通信，其中服务端提供位置信息（绑定IP地址和监听端口），客户端通过连接操作向服务端监听的地址发起连接请求，通过三次握手建立连接，如果连接成功，双方通过网络套接字（Socket）进行通信。

在基于传统同步阻塞开发模式中，ServerSocket负责绑定IP地址，启动监听端口：Socket负责发起连接操作。连接成功之后，双方通过输入和输出流进行同步阻塞式通信。

BIO通信模型图

Center

BIO演示代码

服务端程序代码：

  1. package com.nio.server;
  3. import java.io.IOException;
  4. import java.net.ServerSocket;
  5. import java.net.Socket;
  7. import com.nio.thread.ServerThread;
  9. public class TimeServer {
  11.     /**
  12.      * @param args
  13.      */
  14.     public static void main(String[] args) throws IOException {
  15.         // TODO Auto-generated method stub
  16.         //监听端口
  17.         int port=1666;
  18.         //创建ServerSocket 对象
  19.         ServerSocket server=null;
  20.         try{
  21.             server=new ServerSocket(port);
  22.             System.out.println("Serverocket启动对:"+port+"端口监听");
  23.             Socket socket=null;
  24.             while(true){
  25.                 socket=server.accept();
  26.                 new Thread(new ServerThread(socket)).start();
  27.             }
  29.         }finally{
  30.             if(server!=null){
  31.                 System.out.println("ServerSocket 服务正在关闭");
  32.                 server.close();
  33.                 server=null;
  34.             }
  35.         }
  37.     }
  39. }
  41. package com.nio.thread;
  43. import java.io.BufferedReader;
  44. import java.io.IOException;
  45. import java.io.InputStreamReader;
  46. import java.io.PrintWriter;
  47. import java.net.Socket;
  49. public class ServerThread implements Runnable {
  50.     // 套接字
  51.     private Socket socket;
  52.     // 构造函数
  53.     public ServerThread(Socket socket) {
  54.         super();
  55.         this.socket = socket;
  56.     }
  57.     public void run() {
  58.         // TODO Auto-generated method stub
  59.         BufferedReader in = null;
  60.         PrintWriter out = null;
  61.         try {
  62.             in = new BufferedReader(new InputStreamReader(
  63.                     this.socket.getInputStream()));
  64.             out = new PrintWriter(this.socket.getOutputStream(), true);
  65.             String currentTime = null;
  66.             String body = null;
  67.             while (true) {
  68.                 body = in.readLine();
  69.                 if (body == null) {
  70.                     break;
  71.                 }
  72.                 System.out.println("this ServerSocket receiver message is:"
  73.                         + body);
  74.                 currentTime = "time".equalsIgnoreCase(body) ? new java.util.Date(
  75.                         System.currentTimeMillis()).toString() : "bad time";
  76.                 out.println("current is:" + currentTime);
  77.             }
  78.         } catch (Exception e) {
  79.             e.printStackTrace();
  80.             if (in != null) {
  81.                 try {
  82.                     in.close();
  83.                     in = null;
  84.                 } catch (IOException el) {
  85.                     el.printStackTrace();
  86.                 }
  87.             }
  88.             if (out != null) {
  89.                 out.close();
  90.                 out = null;
  91.             }
  92.             if (this.socket != null) {
  93.                 try {
  94.                     this.socket.close();
  95.                 } catch (IOException el) {
  96.                     el.printStackTrace();
  97.                 }
  98.                 this.socket = null;
  99.             }
  101.         }
  103.     }
  105. }

客户端程序代码：

  1. package com.nio.client;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  6. import java.io.PrintWriter;
  7. import java.net.Socket;
  9. public class TimeClient {
  11.     /**
  12.      * @param args
  13.      */
  14.     public static void main(String[] args) {
  15.         // TODO Auto-generated method stub
  16.         int port=1666;
  17.         Socket socket=null;
  18.         BufferedReader in = null;
  19.         PrintWriter out = null;
  20.         try{
  21.             socket=new Socket("127.0.0.1",port);
  22.             in = new BufferedReader(new InputStreamReader(
  23.                     socket.getInputStream()));
  24.             out = new PrintWriter(socket.getOutputStream(), true);
  25.             out.println("123");
  26.             System.out.println("this client has send message");
  27.             String resp=in.readLine();
  28.             System.out.println("response message is:"+resp);
  29.         }catch(Exception e){
  31.         }finally{
  32.             if (in != null) {
  33.                 try {
  34.                     in.close();
  35.                     in = null;
  36.                 } catch (IOException el) {
  37.                     el.printStackTrace();
  38.                 }
  39.             }
  40.             if (out != null) {
  41.                 out.close();
  42.                 out = null;
  43.             }
  44.             if (socket != null) {
  45.                 try {
  46.                     socket.close();
  47.                 } catch (IOException el) {
  48.                     el.printStackTrace();
  49.                 }
  50.                 socket = null;
  51.             }
  52.         }
  55.     }
  57. }

结果展示:

客户端发送信息为:”123”

服务端返回结果为:

Center 1

客户端发送信息为:”time”

服务端返回信息为:

Center 2

第二节：伪异步IO编程

为了解决同步阻塞IO面临的一个链路需要一个线程处理的问题，后来有人对它的线程模式进行了优化，后端通过一个线程池来处理多个客户端的请求接入，形成客户端个数M:线程池最大线程数N的比例关系，其中M可以远远大于N，通过线程池可以灵活的调配线程资源，设置线程的最大值，防止由于海量并发导致线程耗尽。

伪异步IO模型图

Center 3

采用线程池和任务队列可以实现一种叫做伪异步的IO通信框架，它的模型如上图。

当有新的客户端接入的时候，将客户端的Socket封装成一个Task（）该任务实现java.lang.Runnable 接口投递到后端的线程池进行处理，JDK维护一个消息队列和N个活跃的线程对消息队列中的任务进行处理。由于线程池可以设置消息队列的大小和最大线程数，因此，它的资源是可控的，无论多少个客户端并发访问，都不会导致资源耗尽和宕机。

伪异步IO演示代码

服务端代码

  1. package com.nio.server;
  3. import java.io.IOException;
  4. import java.net.ServerSocket;
  5. import java.net.Socket;
  7. import com.nio.executepool.TimeServerHandlerExcutePool;
  8. import com.nio.thread.ServerThread;
  10. public class TimeServer {
  12.     /**
  13.      * @param args
  14.      */
  15.     public static void main(String[] args) throws IOException {
  16.         // TODO Auto-generated method stub
  17.         //监听端口
  18.         int port=1667;
  19.         //创建ServerSocket 对象
  20.         ServerSocket server=null;
  21.         try{
  22.             server=new ServerSocket(port);
  23.             System.out.println("Serverocket启动对:"+port+"端口监听");
  24.             Socket socket=null;
  25.             TimeServerHandlerExcutePool pool=new TimeServerHandlerExcutePool(50,1000);
  26.             while(true){
  27.                 socket=server.accept();
  28.                 pool.execute(new ServerThread(socket));            
  29.             }
  31.         }finally{
  32.             if(server!=null){
  33.                 System.out.println("ServerSocket 服务正在关闭");
  34.                 server.close();
  35.                 server=null;
  36.             }
  37.         }
  39.     }
  41. }

线程池

  1. package com.nio.executepool;
  3. import java.util.concurrent.ArrayBlockingQueue;
  4. import java.util.concurrent.ExecutorService;
  5. import java.util.concurrent.ThreadPoolExecutor;
  6. import java.util.concurrent.TimeUnit;
  8. public class TimeServerHandlerExcutePool {
  9.     private ExecutorService executor;
  11.     public TimeServerHandlerExcutePool(int maxPoolSize,int queueSize) {
  12.         executor=new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),maxPoolSize,120L,TimeUnit.SECONDS,new ArrayBlockingQueue<java.lang.Runnable>(queueSize));
  13.     }
  15.     public void execute(java.lang.Runnable command){
  16.         executor.execute(command);
  17.     }
  23. }

线程类

  1. package com.nio.thread;
  3. import java.io.BufferedReader;
  4. import java.io.IOException;
  5. import java.io.InputStreamReader;
  6. import java.io.PrintWriter;
  7. import java.net.Socket;
  9. public class ServerThread implements Runnable {
  10.     // 套接字
  11.     private Socket socket;
  12.     // 构造函数
  13.     public ServerThread(Socket socket) {
  14.         super();
  15.         this.socket = socket;
  16.     }
  17.     public void run() {
  18.         // TODO Auto-generated method stub
  19.         BufferedReader in = null;
  20.         PrintWriter out = null;
  21.         try {
  22.             in = new BufferedReader(new InputStreamReader(
  23.                     this.socket.getInputStream()));
  24.             out = new PrintWriter(this.socket.getOutputStream(), true);
  25.             String currentTime = null;
  26.             String body = null;
  27.             while (true) {
  28.                 body = in.readLine();
  29.                 if (body == null) {
  30.                     break;
  31.                 }
  32.                 System.out.println("this ServerSocket receiver message is:"
  33.                         + body);
  34.                 currentTime = "time".equalsIgnoreCase(body) ? new java.util.Date(
  35.                         System.currentTimeMillis()).toString() : "bad time";
  36.                 out.println("current is:" + currentTime);
  37.             }
  38.         } catch (Exception e) {
  39.             e.printStackTrace();
  40.             if (in != null) {
  41.                 try {
  42.                     in.close();
  43.                     in = null;
  44.                 } catch (IOException el) {
  45.                     el.printStackTrace();
  46.                 }
  47.             }
  48.             if (out != null) {
  49.                 out.close();
  50.                 out = null;
  51.             }
  52.             if (this.socket != null) {
  53.                 try {
  54.                     this.socket.close();
  55.                 } catch (IOException el) {
  56.                     el.printStackTrace();
  57.                 }
  58.                 this.socket = null;
  59.             }
  61.         }
  63.     }
  65. }

客户端代码

客户端代码未做任何修改。

第三节 NIO编程

在介绍NIO编程之前，我们首先澄清一个概念：NIO到底是什么的简称？有人称之为New IO，因为它是相对之前的IO库新增的，所以称之为New IO。这是它的官方叫法。但是，由于之前的IO类库是阻塞IO，New IO类库的目标就是让IO支持非阻塞，所以，人们更喜欢称之为非阻塞IO，由于非阻塞IO更能体现NIO的特点。

与Socket和ServerSocket类相对应，NIO也提供了SocketChannel和ServerSocketChannel两种不同的套接字通道实现。这两种新增的通道都支持阻塞和非阻塞两种模式。阻塞模式使用非常简单，但是性能和可靠性都不好，非阻塞模式正好相反。一般来说低负载、低并发的应用程序可以选择阻塞式IO来降低开发的编程复杂度，但是对于高负载、高并发的网络应用，需要使用NIO非阻塞模式进行开发。

NIO演示代码

服务端代码

  1. package com.nio.server;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.ServerSocketChannel;
  9. import java.nio.channels.SocketChannel;
  10. import java.util.Iterator;
  12. /**
  13.  * NIO服务端
  14.  * @author 小路
  15.  */
  16. public class NIOServer {
  17.     //通道管理器
  18.     private Selector selector;
  20.     /**
  21.      * 获得一个ServerSocket通道，并对该通道做一些初始化的工作
  22.      * @param port  绑定的端口号
  23.      * @throws IOException
  24.      */
  25.     public void initServer(int port) throws IOException {
  26.         // 获得一个ServerSocket通道
  27.         ServerSocketChannel serverChannel = ServerSocketChannel.open();
  28.         // 设置通道为非阻塞
  29.         serverChannel.configureBlocking(false);
  30.         // 将该通道对应的ServerSocket绑定到port端口
  31.         serverChannel.socket().bind(new InetSocketAddress(port));
  32.         // 获得一个通道管理器
  33.         this.selector = Selector.open();
  34.         //将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后，
  35.         //当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。
  36.         serverChannel.register(selector, SelectionKey.OP_ACCEPT);
  37.     }
  39.     /**
  40.      * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
  41.      * @throws IOException
  42.      */
  43.     @SuppressWarnings("unchecked")
  44.     public void listen() throws IOException {
  45.         System.out.println("服务端启动成功！");
  46.         // 轮询访问selector
  47.         while (true) {
  48.             //当注册的事件到达时，方法返回；否则,该方法会一直阻塞
  49.             selector.select();
  50.             // 获得selector中选中的项的迭代器，选中的项为注册的事件
  51.             Iterator ite = this.selector.selectedKeys().iterator();
  52.             while (ite.hasNext()) {
  53.                 SelectionKey key = (SelectionKey) ite.next();
  54.                 // 删除已选的key,以防重复处理
  55.                 ite.remove();
  56.                 // 客户端请求连接事件
  57.                 if (key.isAcceptable()) {
  58.                     ServerSocketChannel server = (ServerSocketChannel) key
  59.                             .channel();
  60.                     // 获得和客户端连接的通道
  61.                     SocketChannel channel = server.accept();
  62.                     // 设置成非阻塞
  63.                     channel.configureBlocking(false);
  65.                     //在这里可以给客户端发送信息哦
  66.                     channel.write(ByteBuffer.wrap(new String("server").getBytes()));
  67.                     //在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。
  68.                     channel.register(this.selector, SelectionKey.OP_READ);
  70.                     // 获得了可读的事件
  71.                 } else if (key.isReadable()) {
  72.                         read(key);
  73.                 }
  75.             }
  77.         }
  78.     }
  79.     /**
  80.      * 处理读取客户端发来的信息 的事件
  81.      * @param key
  82.      * @throws IOException 
  83.      */
  84.     public void read(SelectionKey key) throws IOException{
  85.         // 服务器可读取消息:得到事件发生的Socket通道
  86.         SocketChannel channel = (SocketChannel) key.channel();
  87.         // 创建读取的缓冲区
  88.         ByteBuffer buffer = ByteBuffer.allocate(10);
  89.         channel.read(buffer);
  90.         byte[] data = buffer.array();
  91.         String msg = new String(data).trim();
  92.         System.out.println("服务端收到信息："+msg);
  93.         ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
  94.         channel.write(outBuffer);// 将消息回送给客户端
  95.     }
  97.     /**
  98.      * 启动服务端测试
  99.      * @throws IOException 
  100.      */
  101.     public static void main(String[] args) throws IOException {
  102.         NIOServer server = new NIOServer();
  103.         server.initServer(8000);
  104.         server.listen();
  105.     }
  107. }

客户端代码

  1. package com.nio.client;
  3. import java.io.IOException;
  4. import java.net.InetSocketAddress;
  5. import java.nio.ByteBuffer;
  6. import java.nio.channels.SelectionKey;
  7. import java.nio.channels.Selector;
  8. import java.nio.channels.SocketChannel;
  9. import java.util.Iterator;
  11. /**
  12.  * NIO客户端
  13.  * @author 小路
  14.  */
  15. public class NIOClient {
  16.     //通道管理器
  17.     private Selector selector;
  19.     /**
  20.      * 获得一个Socket通道，并对该通道做一些初始化的工作
  21.      * @param ip 连接的服务器的ip
  22.      * @param port  连接的服务器的端口号         
  23.      * @throws IOException
  24.      */
  25.     public void initClient(String ip,int port) throws IOException {
  26.         // 获得一个Socket通道
  27.         SocketChannel channel = SocketChannel.open();
  28.         // 设置通道为非阻塞
  29.         channel.configureBlocking(false);
  30.         // 获得一个通道管理器
  31.         this.selector = Selector.open();
  33.         // 客户端连接服务器,其实方法执行并没有实现连接，需要在listen（）方法中调
  34.         //用channel.finishConnect();才能完成连接
  35.         channel.connect(new InetSocketAddress(ip,port));
  36.         //将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_CONNECT事件。
  37.         channel.register(selector, SelectionKey.OP_CONNECT);
  38.     }
  40.     /**
  41.      * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
  42.      * @throws IOException
  43.      */
  44.     @SuppressWarnings("unchecked")
  45.     public void listen() throws IOException {
  46.         // 轮询访问selector
  47.         while (true) {
  48.             selector.select();
  49.             // 获得selector中选中的项的迭代器
  50.             Iterator ite = this.selector.selectedKeys().iterator();
  51.             while (ite.hasNext()) {
  52.                 SelectionKey key = (SelectionKey) ite.next();
  53.                 // 删除已选的key,以防重复处理
  54.                 ite.remove();
  55.                 // 连接事件发生
  56.                 if (key.isConnectable()) {
  57.                     SocketChannel channel = (SocketChannel) key
  58.                             .channel();
  59.                     // 如果正在连接，则完成连接
  60.                     if(channel.isConnectionPending()){
  61.                         channel.finishConnect();
  63.                     }
  64.                     // 设置成非阻塞
  65.                     channel.configureBlocking(false);
  67.                     //在这里可以给服务端发送信息哦
  68.                     channel.write(ByteBuffer.wrap(new String("client").getBytes()));
  69.                     //在和服务端连接成功之后，为了可以接收到服务端的信息，需要给通道设置读的权限。
  70.                     channel.register(this.selector, SelectionKey.OP_READ);
  72.                     // 获得了可读的事件
  73.                 } else if (key.isReadable()) {
  74.                         read(key);
  75.                 }
  77.             }
  79.         }
  80.     }
  81.     /**
  82.      * 处理读取服务端发来的信息 的事件
  83.      * @param key
  84.      * @throws IOException 
  85.      */
  86.     public void read(SelectionKey key) throws IOException{        
  87.         // 客户端可读取消息:得到事件发生的Socket通道
  88.                 SocketChannel channel = (SocketChannel) key.channel();
  89.                 // 创建读取的缓冲区
  90.                 ByteBuffer buffer = ByteBuffer.allocate(10);
  91.                 channel.read(buffer);
  92.                 byte[] data = buffer.array();
  93.                 String msg = new String(data).trim();
  94.                 System.out.println("客户端收到信息："+msg);
  95.                 ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
  96.                 channel.write(outBuffer);// 将消息回送给服务器
  97.     }
  100.     /**
  101.      * 启动客户端测试
  102.      * @throws IOException 
  103.      */
  104.     public static void main(String[] args) throws IOException {
  105.         NIOClient client = new NIOClient();
  106.         client.initClient("localhost",8000);
  107.         client.listen();
  108.     }
  110. }

发表评论

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

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

相关阅读

    相关 Netty权威指南之文件传输

    本章相关知识点： 文件是最常见的数据源之一，在程序经常需要将数据存储到文件中，比如：图片文件、声音文件等数据文件。在实际使用中，文件都包含一个特定的格式，这个格式需要程序员根

    Dear 丶Dear 丶/ 0 赞/ 71 阅读

    相关 Netty权威指南之AIO编程

    由JDK1.7提供的NIO2.0新增了异步的套接字通道，它是真正的异步I/O，在异步I/O操作的时候可以传递信号变量，当操作完成后会回调相关的方法，异步I/o也被称为AIO，对

    比眉伴天荒比眉伴天荒/ 0 赞/ 574 阅读