javaNIO编程

Posted 2021-03-13 石

Unblocking IO（New IO）： 同步非阻塞的编程方式

NIO 本身是基于事件驱动思想来完成的，其主要想解决的是 BIO 的大并发问题，NIO 基 于 Reactor，当 socket 有流可读或可写入 socket 时，操作系统会相应的通知引用程序进行处 理，应用再将流读取到缓冲区或写入操作系统。也就是说，这个时候，已经不是一个连接就 要对应一个处理线程了，而是有效的请求，对应一个线程，当连接没有数据时，是没有工作 线程来处理的。 

NIO 的最重要的地方是当一个连接创建后，不需要对应一个线程，这个连接会被注册到 多路复用器上面，所以所有的连接只需要一个线程就可以搞定，当这个线程中的多路复用器 进行轮询的时候，发现连接上有请求的话，才开启一个线程进行处理，也就是一个请求一个 线程模式

在 NIO 的处理方式中，当一个请求来的话，开启线程进行处理，可能会等待后端应用的 资源(JDBC 连接等)，其实这个线程就被阻塞了，当并发上来的话，还是会有 BIO 一样的问题。

 

 同步非阻塞，服务器实现模式为一个请求一个通道，即客户端发送的连接请求都会注册 到多路复用器上，多路复用器轮询到连接有 I/O 请求时才启动一个线程进行处理。 NIO 方式适用于连接数目多且连接比较短（轻操作）的架构，比如聊天服务器，并发局 限于应用中，编程复杂，JDK1.4 开始支持。 

Buffer:ByteBuffer,CharBuffer,ShortBuffer,IntBuffer,LongBuffer,FloatBuffer,DoubleBuffer

Channel:SocketChannel,ServerSocketChannel 。
Selector:Selector,AbstractSelector SelectionKey:OP_READ,OP_WRITE,OP_CONNECT,OP_ACCEPT

 

 

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.Scanner;

public class NIOClient {
	public static void main(String[] args) {
		// 远程地址创建
		InetSocketAddress remote =new InetSocketAddress("localhost",  9999);
		SocketChannel channel = null;
		
		// 定义缓存
		ByteBuffer buffer =ByteBuffer.allocate(1024);
		
		try {
			// 开启通道
			channel=SocketChannel.open();
			// 连接远程服务器。
			channel.connect(remote);
			Scanner reader = new Scanner(System.in);
			while(true){
				System.out.print("put message for send to server > ");
				String line = reader.nextLine();
				if(line.equals("exit")){
					break;
				}
				// 将控制台输入的数据写入到缓存。
				 buffer.put(line.getBytes("UTF-8"));
				// 重置缓存游标
				 buffer.flip();
				// 将数据发送给服务器
				channel.write(buffer);
				// 清空缓存数据。
				buffer.clear();
				
				// 读取服务器返回的数据
				int readLength=channel.read(buffer);
				if(readLength==-1) break;
				// 重置缓存游标
				buffer.flip();
				byte[] datas=new byte[buffer.remaining()];
				// 读取数据到字节数组。
				buffer.get(datas);
				System.out.println("from server : " + new String(datas, "UTF-8"));
				// 清空缓存。
				buffer.clear();
			}
		} catch (IOException  e) {
			e.printStackTrace();
		}finally{
			if(null != channel){
				try {
					channel.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
	}
}

　　

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;

public class Nioservice implements Runnable {
	
	// 多路复用器， 选择器。 用于注册通道的。
	private Selector selector;
	// 定义了两个缓存。分别用于读和写。 初始化空间大小单位为字节。
	private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
	private ByteBuffer writeBuffer = ByteBuffer.allocate(1024);

	public static void main(String[] args) {
        new Thread(new NioService(9999)).start();
	}
	
	public NioService(int port){
		init(port);
	}
	
	private void init(int port){
		try {
			System.out.println("server starting at port " + port + " ...");
			// 开启多路复用器
			this.selector=Selector.open();
			// 开启服务通道
			ServerSocketChannel serverChannel =ServerSocketChannel.open();
			// 非阻塞， 如果传递参数true，为阻塞模式。
			serverChannel.configureBlocking(false);
			// 绑定端口
			serverChannel.bind(new InetSocketAddress(port));
			
			// 注册，并标记当前服务通道状态
			
			/*
			 * register(Selector, int)
			 * int - 状态编码
			 *  OP_ACCEPT ： 连接成功的标记位。
			 *  OP_READ ： 可以读取数据的标记
			 *  OP_WRITE ： 可以写入数据的标记
			 *  OP_CONNECT ： 连接建立后的标记
			 */
			serverChannel.register(this.selector, SelectionKey.OP_ACCEPT);
			System.out.println("server started.");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void run(){
		while(true){
			try {
				// 阻塞方法，当至少一个通道被选中，此方法返回。
				// 通道是否选择，由注册到多路复用器中的通道标记决定。
				this.selector.select();
				// 返回以选中的通道标记集合， 集合中保存的是通道的标记。相当于是通道的ID。
				Iterator<SelectionKey> keys = this.selector.selectedKeys().iterator();
				while(keys.hasNext()){
					SelectionKey key = keys.next();
					// 将本次要处理的通道从集合中删除，下次循环根据新的通道列表再次执行必要的业务逻辑
					keys.remove();
					// 通道是否有效
					if(key.isValid()){
						// 阻塞状态
						try{
							if(key.isAcceptable()){
								accept(key);
							}
						}catch(CancelledKeyException cke){
							// 断开连接。 出现异常。
							key.cancel();
						}
						// 可读状态
						try{
							if(key.isReadable()){
								read(key);
							}
						}catch(CancelledKeyException cke){
							key.cancel();
						}
						// 可写状态
						try{
							if(key.isWritable()){
								write(key);
							}
						}catch(CancelledKeyException cke){
							key.cancel();
						}
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			
		}
	}

	private void write(SelectionKey key){
		this.writeBuffer.clear();
		SocketChannel  channel =(SocketChannel) key.channel();
		Scanner reader=new Scanner(System.in);
		try {
			System.out.print("put message for send to client > ");
			String line=reader.nextLine();
			// 将控制台输入的字符串写入Buffer中。 写入的数据是一个字节数组。
			writeBuffer.put(line.getBytes("UTF-8"));
			writeBuffer.flip();
			channel.write(writeBuffer);
			
			channel.register(this.selector, SelectionKey.OP_READ);
		} catch (Exception e) {
			// TODO: handle exception
		}
	}

    private void read(SelectionKey key){
    	try {
    		// 清空读缓存。
    		this.readBuffer.clear();
    		//获取通道
    		SocketChannel  channel =(SocketChannel)key.channel();
    		// 将通道中的数据读取到缓存中。通道中的数据，就是客户端发送给服务器的数据。
    		int readLength =channel.read(readBuffer);
    		// 检查客户端是否写入数据。
    		if(readLength==-1){
    			// 关闭通道
    			key.channel().close();
    			// 关闭连接
    			key.cancel();
    			return;
    		}
    		/*
			 * flip， NIO中最复杂的操作就是Buffer的控制。
			 * Buffer中有一个游标。游标信息在操作后不会归零，如果直接访问Buffer的话，数据有不一致的可能。
			 * flip是重置游标的方法。NIO编程中，flip方法是常用方法。
			 */
    		this.readBuffer.flip();
    		// 字节数组，保存具体数据的。 Buffer.remaining() -> 是获取Buffer中有效数据长度的方法。
    		byte[] datas=new byte[readBuffer.remaining()];
    		// 是将Buffer中的有效数据保存到字节数组中。
    		readBuffer.get(datas);
    		System.out.println("from " + channel.getRemoteAddress() + " client : " + new String(datas, "UTF-8"));
    		
    		// 注册通道， 标记为写操作。
    		channel.register(this.selector, SelectionKey.OP_WRITE);
		} catch (IOException  e) {
			e.printStackTrace();
			try {
				key.channel().close();
				key.cancel();
			} catch (IOException e1) {
				e1.printStackTrace();
			}
		}
    }
 
    private void accept(SelectionKey key){
    	try {
    		// 此通道为init方法中注册到Selector上的ServerSocketChannel
    		ServerSocketChannel serverChannel =(ServerSocketChannel)key.channel();
    		// 阻塞方法，当客户端发起请求后返回。 此通道和客户端一一对应。
    		SocketChannel channel = serverChannel.accept();
    		channel.configureBlocking(false);
    		// 设置对应客户端的通道标记状态，此通道为读取数据使用的。
    		channel.register(this.selector, SelectionKey.OP_READ);	
		} catch (IOException  e) {
			e.printStackTrace();
		}
    }

}

　　

/**
 * 
 * Buffer的应用固定逻辑
 * 写操作顺序
 * 1. clear()
 * 2. put() -> 写操作
 * 3. flip() -> 重置游标
 * 4. SocketChannel.write(buffer); -> 将缓存数据发送到网络的另一端
 * 5. clear()
 * 
 * 读操作顺序
 * 1. clear()
 * 2. SocketChannel.read(buffer); -> 从网络中读取数据
 * 3. buffer.flip() -> 重置游标
 * 4. buffer.get() -> 读取数据
 * 5. buffer.clear()
 *
 */
public class TestBuffer {
	public static void main(String[] args) throws Exception {
		
		ByteBuffer buffer = ByteBuffer.allocate(8);
		
		byte[] temp = new byte[]{3,2,1};
		
		// 写入数据之前 ： java.nio.HeapByteBuffer[pos=0 lim=8 cap=8]
		// pos - 游标位置， lim - 限制数量， cap - 最大容量
		System.out.println("写入数据之前 ： " + buffer);
		
		// 写入字节数组到缓存
		buffer.put(temp);
		
		// 写入数据之后 ： java.nio.HeapByteBuffer[pos=3 lim=8 cap=8]
		// 游标为3， 限制为8， 容量为8
		System.out.println("写入数据之后 ： " + buffer);
		
		// 重置游标 ， lim = pos ;  pos = 0;
		buffer.flip();
		
		// 重置游标之后 ： java.nio.HeapByteBuffer[pos=0 lim=3 cap=8]
		// 游标为0， 限制为3， cap为8
		System.out.println("重置游标之后 ： " + buffer);

		// 清空Buffer， pos = 0; lim = cap;
		// buffer.clear();
		
		// get() -> 获取当前游标指向的位置的数据。
		// System.out.println(buffer.get());
		
		/*for(int i = 0; i < buffer.remaining(); i++){
			// get(int index) -> 获取指定位置的数据。
			int data = buffer.get(i);
			System.out.println(i + " - " + data);
		}*/
	}
}