JAVA NIO 之一

传统IO 写道

  网络传输方式问题：传统的RPC框架或者基于RMI等方式的远程服务（过程）调用采用了同步阻塞IO，当客户端的并发压力或者网络时延增大之后，同步阻塞IO会由于频繁的wait导致IO线程经常性的阻塞，由于线程无法高效的工作，IO处理能力自然下降。下面，我们通过BIO通信模型图看下BIO通信的弊端：

  采用BIO通信模型的服务端，通常由一个独立的Acceptor线程负责监听客户端的连接，接收到客户端连接之后为客户端连接创建一个新的线程处理请求消息，处理完成之后，返回应答消息给客户端，线程销毁，这就是典型的一请求一应答模型。该架构最大的问题就是不具备弹性伸缩能力，当并发访问量增加后，服务端的线程个数和并发访问数成线性正比，由于线程是JAVA虚拟机非常宝贵的系统资源，当线程数膨胀之后，系统的性能急剧下降，随着并发量的继续增加，可能会发生句柄溢出、线程堆栈溢出等问题，并导致服务器最终宕机。

NIO 写道

  在IO编程过程中，当需要同时处理多个客户端接入请求时，可以利用多线程或者IO多路复用技术进行处理。IO多路复用技术通过把多个IO的阻塞复用到同一个select的阻塞上，从而使得系统在单线程的情况下可以同时处理多个客户端请求。与传统的多线程/多进程模型比，I/O多路复用的最大优势是系统开销小，系统不需要创建新的额外进程或者线程，也不需要维护这些进程和线程的运行，降低了系统的维护工作量，节省了系统资源。
  JDK1.4提供了对非阻塞IO（NIO）的支持，JDK1.5_update10版本使用epoll替代了传统的select/poll，极大的提升了NIO通信的性能。
  JAVA NIO实现服务端与客户端简单数据传输

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
/**
* 服务端
*/
public class SocketServer {

	/**
	 * 服务器默认绑定端口
	 */
	public static final int DEFAULT_PORT = 9999;

	/**
	 * 选择器
	 */
	public Selector selector;

	public SocketServer(String ip, int port) {
		ServerSocketChannel ssc = null;
		try {
			int _port = DEFAULT_PORT;
			if (port > 0)
				_port = port;
			/* 获取通道 */
			ssc = ServerSocketChannel.open();
			/* 配置非阻塞 */
			ssc.configureBlocking(false);
			/**
			 * 配置绑定端口 ServerSocketChannel没有bind()方法，
			 * 因此有必要取出对等的ServerSocket并使用它来绑定到一
			 * 个端口以开始监听连接
			 */
			ssc.socket().bind(new InetSocketAddress(ip, _port));
			/* 获取选择器 */
			this.selector = Selector.open();
			/* 将通道注册到选择器 */
			ssc.register(this.selector, SelectionKey.OP_ACCEPT);
		}catch(ClosedChannelException e1){
			System.out.println("关闭的通道,无法注册到选择器");
			e1.printStackTrace();
		} catch (IOException e2) {
			try {
				if(ssc != null) ssc.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			System.out.println("服务器绑定端口冲突");
			e2.printStackTrace();
		}
	}
	
	/**
	 * 轮询选择器
	 * @throws Exception
	 */
	public void pollSelect() throws Exception {
		/* (阻塞)轮询选择器,直到有事件 */
		while (this.selector.select()>0) {
			/* 获取事件通知列表 */
			Iterator<SelectionKey> it = this.selector.selectedKeys().iterator();
			while (it.hasNext()) {
				SelectionKey selectKey = it.next();
				it.remove();
				try {
					process(selectKey);
				} catch (Exception e) {
					e.printStackTrace();
					continue;
				}
			}
		}
	}

	/**
	 * 事件处理
	 * @param selectKey
	 */
	public void process(SelectionKey selectKey) throws Exception{
		if (selectKey.isAcceptable()) { /* 客户端连接事件 */
			accept(selectKey);
		} else if (selectKey.isReadable()) { /* 可读事件 */
			read(selectKey);
		}
	}
	
	/**
	 * 连接事件
	 * @param selectKey
	 */
	public void accept(SelectionKey selectKey) throws Exception {
		ServerSocketChannel ssc = null;
		try {
			ssc = (ServerSocketChannel) selectKey
					.channel();
			SocketChannel sc = ssc.accept();
			sc.configureBlocking(false);
			/* 发送信息 */
			sc.write(ByteBuffer.wrap(new String("Hello World!")
					.getBytes()));
			/* 注册读事件 */
			sc.register(this.selector, SelectionKey.OP_READ);
		} catch (ClosedChannelException e) {
			if(ssc!=null) 
				ssc.close();
			throw new IOException("关闭的通道,无法注册到选择器");
		} catch (IOException e) {
			if(ssc!=null) 
				ssc.close();
			throw new IOException("连接服务或配置失败!");
		}
	}
	
	/**
	 * 可读事件
	 * @param selectKey
	 */
	public void read(SelectionKey selectKey) throws Exception{
		SocketChannel channel = null;
		try {
			// 服务器可读取消息:得到事件发生的Socket通道
			channel = (SocketChannel) selectKey.channel();
			// 创建读取的缓冲区
			ByteBuffer buffer = ByteBuffer.allocate(100);
			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);
		} catch (Exception e) {
			if(channel != null) 
				channel.close();
			throw new Exception("客户端将通道关闭,无法从通道读入缓冲或将缓冲数据写回通道!");
		}
	}
	
	
	public static void main(String[] args) {
		SocketServer ss = null;
		try {
			ss = new SocketServer("localhost", 9999);
			ss.pollSelect();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
/**
* 客户端
*/
public class SocketClient {

	public Selector selector;
	
	public SocketClient(String ip, int port){
		SocketChannel channel = null;
		try {
			//channel = SocketChannel.open(new InetSocketAddress(ip,port));
			channel = SocketChannel.open();
			// 设置通道为非阻塞  
	        channel.configureBlocking(false);
	        // 获得一个通道管理器  
	        this.selector = Selector.open();
	        // 客户端连接服务器,其实方法执行并没有实现连接 
	        channel.connect(new InetSocketAddress(ip, port)); 
	        /**while(!channel.finishConnect()){
	        	System.out.println("尝试连接....");
	        }*/
	        // 注册连接事件。  
	        channel.register(this.selector, SelectionKey.OP_CONNECT);  
		} catch(ClosedChannelException e1){
			System.out.println("关闭的通道,无法注册到选择器");
			e1.printStackTrace();
		} catch (IOException e2) {
			System.out.println("连接异常!");
			try {
				if(channel != null) channel.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
			e2.printStackTrace();
		}
	}
	
	/**
	 * 轮询选择器
	 * @throws IOException
	 */
	public void  pollSelect() throws Exception {  
		/* (阻塞)轮询选择器,直到有事件 */
		while ( this.selector.select() > 0 ) {
            /* 获取事件通知列表 */
            Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator();  
            while (ite.hasNext()) {  
                SelectionKey selectKey = (SelectionKey) ite.next();  
                // 删除已选的key,以防重复处理  
                ite.remove();  
                process(selectKey);
            }  
        }  
    }
	
	/**
	 * 处理事件
	 * @param selectKey
	 */
	public void process(SelectionKey selectKey) throws Exception{
        if (selectKey.isConnectable()) {
        	connect(selectKey);
        } else if (selectKey.isReadable()) {  
        	read(selectKey);
        }  
	}
	
	/**
	 * 连接事件
	 * @param selectKey
	 * @throws Exception
	 */
	public void connect(SelectionKey selectKey) throws Exception{
		try {
			SocketChannel channel = (SocketChannel) selectKey  
			        .channel();  
			/* 如果正在连接,则完成连接 */  
			if(channel.isConnectionPending()){
				/**
				 * connect()方法尚未被调用,调用finishConnect()方法,
				 * 那么将产生NoConnectionPendingException
				 */
			    channel.finishConnect();  
			}
			/**
			 * 在非阻塞模式下调用connect()方法之后,SocketChannel又被切换回了阻塞模式;那么如果
			 * 有必要的话，调用线程会阻塞直到连接建立完成,finishConnect()方法接着就会返回true
			 * 值。
			 */
			/* 设置成非阻塞 */  
			channel.configureBlocking(false);  
			/* 给服务端发送信息 */
			channel.write(ByteBuffer.wrap(new String("编号001客户端连接成功!").getBytes()));  
			/* 注册读事件 */  
			channel.register(this.selector, SelectionKey.OP_READ);
		} catch (ClosedChannelException e) {
			throw new IOException("关闭的通道,无法注册到选择器");
		} catch (IOException e) {
			throw new IOException("连接服务或配置失败!");
		}
	}
	
	/**
	 * 读事件
	 * @param selectKey
	 * @throws Exception
	 */
	public void read(SelectionKey selectKey) throws Exception{
		try {
			// 服务器可读通道
			SocketChannel channel = (SocketChannel) selectKey.channel(); 
			// 创建读取的缓冲区  
			ByteBuffer buffer = ByteBuffer.allocate(100);  
			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);
		} catch (Exception e) {
			throw new IOException("服务端将通道关闭,无法从通道读入缓冲或将缓冲数据写回通道!");
		}
	}
	
	public static void main(String[] args) {
		SocketClient sc = null;
		try {
			sc = new SocketClient("localhost", 9999);
			sc. pollSelect();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
}

评论

1 楼 successfultzb 2016-01-15  

  
刚好准备了解这个，讲的很详细明了，赞