Netty权威指南之NIO入门

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

本章学习目标：

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

2、基于NIO非阻塞式编程

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

4、为什么使用NIO编程

5、为什么选择Netty

**传统BIO编程**

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

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

**BIO通信模型图**

![Center][]

**BIO演示代码**

**服务端程序代码：**

package com.nio.server;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    import com.nio.thread.ServerThread;
    
    public class TimeServer {
    
    	/**
    	 * @param args
    	 */
    	public static void main(String[] args) throws IOException {
    		// TODO Auto-generated method stub
    		//监听端口
    		int port=1666;
    		//创建ServerSocket 对象
    		ServerSocket server=null;
    		try{
    			server=new ServerSocket(port);
    			System.out.println("Serverocket启动对:"+port+"端口监听");
    			Socket socket=null;
    			while(true){
    				socket=server.accept();
    				new Thread(new ServerThread(socket)).start();
    			}
    				
    		}finally{
    			if(server!=null){
    				System.out.println("ServerSocket 服务正在关闭");
    				server.close();
    				server=null;
    			}
    		}
    
    	}
    
    }

package com.nio.thread;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    
    public class ServerThread implements Runnable {
    	// 套接字
    	private Socket socket;
    	// 构造函数
    	public ServerThread(Socket socket) {
    		super();
    		this.socket = socket;
    	}
    	public void run() {
    		// TODO Auto-generated method stub
    		BufferedReader in = null;
    		PrintWriter out = null;
    		try {
    			in = new BufferedReader(new InputStreamReader(
    					this.socket.getInputStream()));
    			out = new PrintWriter(this.socket.getOutputStream(), true);
    			String currentTime = null;
    			String body = null;
    			while (true) {
    				body = in.readLine();
    				if (body == null) {
    					break;
    				}
    				System.out.println("this ServerSocket receiver message is:"
    						+ body);
    				currentTime = "time".equalsIgnoreCase(body) ? new java.util.Date(
    						System.currentTimeMillis()).toString() : "bad time";
    				out.println("current is:" + currentTime);
    			}
    		} catch (Exception e) {
    			e.printStackTrace();
    			if (in != null) {
    				try {
    					in.close();
    					in = null;
    				} catch (IOException el) {
    					el.printStackTrace();
    				}
    			}
    			if (out != null) {
    				out.close();
    				out = null;
    			}
    			if (this.socket != null) {
    				try {
    					this.socket.close();
    				} catch (IOException el) {
    					el.printStackTrace();
    				}
    				this.socket = null;
    			}
    
    		}
    
    	}
    
    }

**客户端程序代码：**

package com.nio.client;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.io.PrintWriter;
    import java.net.Socket;
    
    public class TimeClient {
    
    	/**
    	 * @param args
    	 */
    	public static void main(String[] args) {
    		// TODO Auto-generated method stub
    		int port=1666;
    		Socket socket=null;
    		BufferedReader in = null;
    		PrintWriter out = null;
    		try{
    			socket=new Socket("127.0.0.1",port);
    			in = new BufferedReader(new InputStreamReader(
    					socket.getInputStream()));
    			out = new PrintWriter(socket.getOutputStream(), true);
    			out.println("123");
    			System.out.println("this client has send message");
    			String resp=in.readLine();
    			System.out.println("response message is:"+resp);
    		}catch(Exception e){
    			
    		}finally{
    			if (in != null) {
    				try {
    					in.close();
    					in = null;
    				} catch (IOException el) {
    					el.printStackTrace();
    				}
    			}
    			if (out != null) {
    				out.close();
    				out = null;
    			}
    			if (socket != null) {
    				try {
    					socket.close();
    				} catch (IOException el) {
    					el.printStackTrace();
    				}
    				socket = null;
    			}
    		}
    		
    
    	}
    
    }

结果展示:

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

服务端返回结果为:

![Center 1][]

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

服务端返回信息为:

![Center 2][]

**第二节：伪异步IO编程**

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

**伪异步IO模型图**

**![Center 3][]**

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

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

**伪异步IO演示代码**

**服务端代码**

package com.nio.server;
    
    import java.io.IOException;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    import com.nio.executepool.TimeServerHandlerExcutePool;
    import com.nio.thread.ServerThread;
    
    public class TimeServer {
    
    	/**
    	 * @param args
    	 */
    	public static void main(String[] args) throws IOException {
    		// TODO Auto-generated method stub
    		//监听端口
    		int port=1667;
    		//创建ServerSocket 对象
    		ServerSocket server=null;
    		try{
    			server=new ServerSocket(port);
    			System.out.println("Serverocket启动对:"+port+"端口监听");
    			Socket socket=null;
    			TimeServerHandlerExcutePool pool=new TimeServerHandlerExcutePool(50,1000);
    			while(true){
    				socket=server.accept();
    				pool.execute(new ServerThread(socket));			
    			}
    				
    		}finally{
    			if(server!=null){
    				System.out.println("ServerSocket 服务正在关闭");
    				server.close();
    				server=null;
    			}
    		}
    
    	}
    
    }

线程池

package com.nio.executepool;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class TimeServerHandlerExcutePool {
    	private ExecutorService executor;
    
    	public TimeServerHandlerExcutePool(int maxPoolSize,int queueSize) {
    		executor=new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),maxPoolSize,120L,TimeUnit.SECONDS,new ArrayBlockingQueue<java.lang.Runnable>(queueSize));
    	}
    	
    	public void execute(java.lang.Runnable command){
    		executor.execute(command);
    	}
    	
    	
    	
    	
    
    }

线程类

**客户端代码**

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

**第三节 NIO编程**

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

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

**NIO演示代码**

**服务端代码**

package com.nio.server;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    /**
     * NIO服务端
     * @author 小路
     */
    public class NIOServer {
    	//通道管理器
    	private Selector selector;
    
    	/**
    	 * 获得一个ServerSocket通道，并对该通道做一些初始化的工作
    	 * @param port  绑定的端口号
    	 * @throws IOException
    	 */
    	public void initServer(int port) throws IOException {
    		// 获得一个ServerSocket通道
    		ServerSocketChannel serverChannel = ServerSocketChannel.open();
    		// 设置通道为非阻塞
    		serverChannel.configureBlocking(false);
    		// 将该通道对应的ServerSocket绑定到port端口
    		serverChannel.socket().bind(new InetSocketAddress(port));
    		// 获得一个通道管理器
    		this.selector = Selector.open();
    		//将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后，
    		//当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。
    		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    	}
    
    	/**
    	 * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
    	 * @throws IOException
    	 */
    	@SuppressWarnings("unchecked")
    	public void listen() throws IOException {
    		System.out.println("服务端启动成功！");
    		// 轮询访问selector
    		while (true) {
    			//当注册的事件到达时，方法返回；否则,该方法会一直阻塞
    			selector.select();
    			// 获得selector中选中的项的迭代器，选中的项为注册的事件
    			Iterator ite = this.selector.selectedKeys().iterator();
    			while (ite.hasNext()) {
    				SelectionKey key = (SelectionKey) ite.next();
    				// 删除已选的key,以防重复处理
    				ite.remove();
    				// 客户端请求连接事件
    				if (key.isAcceptable()) {
    					ServerSocketChannel server = (ServerSocketChannel) key
    							.channel();
    					// 获得和客户端连接的通道
    					SocketChannel channel = server.accept();
    					// 设置成非阻塞
    					channel.configureBlocking(false);
    
    					//在这里可以给客户端发送信息哦
    					channel.write(ByteBuffer.wrap(new String("server").getBytes()));
    					//在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。
    					channel.register(this.selector, SelectionKey.OP_READ);
    					
    					// 获得了可读的事件
    				} else if (key.isReadable()) {
    						read(key);
    				}
    
    			}
    
    		}
    	}
    	/**
    	 * 处理读取客户端发来的信息 的事件
    	 * @param key
    	 * @throws IOException 
    	 */
    	public void read(SelectionKey key) throws IOException{
    		// 服务器可读取消息:得到事件发生的Socket通道
    		SocketChannel channel = (SocketChannel) key.channel();
    		// 创建读取的缓冲区
    		ByteBuffer buffer = ByteBuffer.allocate(10);
    		channel.read(buffer);
    		byte[] data = buffer.array();
    		String msg = new String(data).trim();
    		System.out.println("服务端收到信息："+msg);
    		ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
    		channel.write(outBuffer);// 将消息回送给客户端
    	}
    	
    	/**
    	 * 启动服务端测试
    	 * @throws IOException 
    	 */
    	public static void main(String[] args) throws IOException {
    		NIOServer server = new NIOServer();
    		server.initServer(8000);
    		server.listen();
    	}
    
    }

**客户端代码**

package com.nio.client;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    /**
     * NIO客户端
     * @author 小路
     */
    public class NIOClient {
    	//通道管理器
    	private Selector selector;
    
    	/**
    	 * 获得一个Socket通道，并对该通道做一些初始化的工作
    	 * @param ip 连接的服务器的ip
    	 * @param port  连接的服务器的端口号         
    	 * @throws IOException
    	 */
    	public void initClient(String ip,int port) throws IOException {
    		// 获得一个Socket通道
    		SocketChannel channel = SocketChannel.open();
    		// 设置通道为非阻塞
    		channel.configureBlocking(false);
    		// 获得一个通道管理器
    		this.selector = Selector.open();
    		
    		// 客户端连接服务器,其实方法执行并没有实现连接，需要在listen（）方法中调
    		//用channel.finishConnect();才能完成连接
    		channel.connect(new InetSocketAddress(ip,port));
    		//将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_CONNECT事件。
    		channel.register(selector, SelectionKey.OP_CONNECT);
    	}
    
    	/**
    	 * 采用轮询的方式监听selector上是否有需要处理的事件，如果有，则进行处理
    	 * @throws IOException
    	 */
    	@SuppressWarnings("unchecked")
    	public void listen() throws IOException {
    		// 轮询访问selector
    		while (true) {
    			selector.select();
    			// 获得selector中选中的项的迭代器
    			Iterator ite = this.selector.selectedKeys().iterator();
    			while (ite.hasNext()) {
    				SelectionKey key = (SelectionKey) ite.next();
    				// 删除已选的key,以防重复处理
    				ite.remove();
    				// 连接事件发生
    				if (key.isConnectable()) {
    					SocketChannel channel = (SocketChannel) key
    							.channel();
    					// 如果正在连接，则完成连接
    					if(channel.isConnectionPending()){
    						channel.finishConnect();
    						
    					}
    					// 设置成非阻塞
    					channel.configureBlocking(false);
    
    					//在这里可以给服务端发送信息哦
    					channel.write(ByteBuffer.wrap(new String("client").getBytes()));
    					//在和服务端连接成功之后，为了可以接收到服务端的信息，需要给通道设置读的权限。
    					channel.register(this.selector, SelectionKey.OP_READ);
    					
    					// 获得了可读的事件
    				} else if (key.isReadable()) {
    						read(key);
    				}
    
    			}
    
    		}
    	}
    	/**
    	 * 处理读取服务端发来的信息 的事件
    	 * @param key
    	 * @throws IOException 
    	 */
    	public void read(SelectionKey key) throws IOException{		
    		// 客户端可读取消息:得到事件发生的Socket通道
    				SocketChannel channel = (SocketChannel) key.channel();
    				// 创建读取的缓冲区
    				ByteBuffer buffer = ByteBuffer.allocate(10);
    				channel.read(buffer);
    				byte[] data = buffer.array();
    				String msg = new String(data).trim();
    				System.out.println("客户端收到信息："+msg);
    				ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
    				channel.write(outBuffer);// 将消息回送给服务器
    	}
    	
    	
    	/**
    	 * 启动客户端测试
    	 * @throws IOException 
    	 */
    	public static void main(String[] args) throws IOException {
    		NIOClient client = new NIOClient();
    		client.initClient("localhost",8000);
    		client.listen();
    	}
    
    }

[Center]: https://img-blog.csdn.net/20150618195051658?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvemhvdXpoaXdlbmdhbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 1]: https://img-blog.csdn.net/20150618202925844?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvemhvdXpoaXdlbmdhbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 2]: https://img-blog.csdn.net/20150618203113571?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvemhvdXpoaXdlbmdhbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 3]: https://img-blog.csdn.net/20150618204053814?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvemhvdXpoaXdlbmdhbmc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center