[Java] Java NIO 详解

关于JavaNIO的API,可以看我上篇博文有详细描述：http://blog.csdn.net/mango_song/article/details/8561427

下面是大牛逸情公子中关于NIO的解读，结合小例子更容易理解NIO原理，原文地址http://weixiaolu.iteye.com/blog/1477774

Java NIO原理图文分析及代码实现

前言:

最近在分析hadoop的RPC(Remote Procedure Call Protocol ，远程过程调用协议，它是一种通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。可以参考：http://baike.baidu.com/view/32726.htm ）机制时，发现hadoop的RPC机制的实现主要用到了两个技术：动态代理（动态代理可以参考博客：http://blog.csdn.net/mango_song/article/details/8554007 ）和java NIO。为了能够正确地分析hadoop的RPC源码，我觉得很有必要先研究一下java NIO的原理和具体实现。

这篇博客我主要从两个方向来分析java NIO

一．java NIO 和阻塞I/O的区别

1. 阻塞I/O通信模型

2. java NIO原理及通信模型

二．java NIO服务端和客户端代码实现

具体分析：

一．java NIO 和阻塞I/O的区别

1. 阻塞I/O通信模型

假如现在你对阻塞I/O已有了一定了解，我们知道阻塞I/O在调用InputStream.read()方法时是阻塞的，它会一直等到数据到来时（或超时）才会返回；同样，在调用ServerSocket.accept()方法时，也会一直阻塞到有客户端连接才会返回，每个客户端连接过来后，服务端都会启动一个线程去处理该客户端的请求。阻塞I/O的通信模型示意图如下：

[Java] Java NIO 详解

如果你细细分析，一定会发现阻塞I/O存在一些缺点。根据阻塞I/O通信模型，我总结了它的两点缺点：

1. 当客户端多时，会创建大量的处理线程。且每个线程都要占用栈空间和一些CPU时间

2. 阻塞可能带来频繁的上下文切换，且大部分上下文切换可能是无意义的。

在这种情况下非阻塞式I/O就有了它的应用前景。

2. java NIO原理及通信模型

Java NIO是在jdk1.4开始使用的，它既可以说成“新I/O”，也可以说成非阻塞式I/O。下面是java NIO的工作原理：

1. 由一个专门的线程来处理所有的 IO 事件，并负责分发。

2. 事件驱动机制：事件到的时候触发，而不是同步的去监视事件。

3. 线程通讯：线程之间通过 wait,notify 等方式通讯。保证每次上下文切换都是有意义的。减少无谓的线程切换。

阅读过一些资料之后，下面贴出我理解的java NIO的工作原理图：

[Java] Java NIO 详解

（注：每个线程的处理流程大概都是读取数据、解码、计算处理、编码、发送响应。）

Java NIO的服务端只需启动一个专门的线程来处理所有的 IO 事件，这种通信模型是怎么实现的呢？呵呵，我们一起来探究它的奥秘吧。java NIO采用了双向通道（channel）进行数据传输，而不是单向的流（stream），在通道上可以注册我们感兴趣的事件。一共有以下四种事件：

事件名	对应值
服务端接收客户端连接事件	SelectionKey.OP_ACCEPT(16)
客户端连接服务端事件	SelectionKey.OP_CONNECT(8)
读事件	SelectionKey.OP_READ(1)
写事件	SelectionKey.OP_WRITE(4)

服务端和客户端各自维护一个管理通道的对象，我们称之为selector，该对象能检测一个或多个通道 (channel) 上的事件。我们以服务端为例，如果服务端的selector上注册了读事件，某时刻客户端给服务端发送了一些数据，阻塞I/O这时会调用read()方法阻塞地读取数据，而NIO的服务端会在selector中添加一个读事件。服务端的处理线程会轮询地访问selector，如果访问selector时发现有感兴趣的事件到达，则处理这些事件，如果没有感兴趣的事件到达，则处理线程会一直阻塞直到感兴趣的事件到达为止。下面是我理解的java NIO的通信模型示意图：

[Java] Java NIO 详解

二．java NIO服务端和客户端代码实现

为了更好地理解java NIO,下面贴出服务端和客户端的简单代码实现。Server和Client每隔5秒钟相互发送一下自己的时间戳

Server端代码

package rpc.nio;
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;
import java.util.Set;

public class NIOServer {
	
	private Selector selector;
	
	public NIOServer() throws IOException{
		//ͨ通道管理器
		this.selector=Selector.open();
		//创建一个ServerSocketChannel
		ServerSocketChannel serverChannel=ServerSocketChannel.open();
		//设置成non-block模式
		serverChannel.configureBlocking(false);
		//将serverChannel对应的server socket绑定IP 和 端口
		serverChannel.socket().bind(new InetSocketAddress("localhost",8001));
		//将serverChannel注册到selector	
		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
	}
	
	public void listen() throws IOException{
		while(true){
			
			int keysAdded=0;
			//用select()函数来监控注册在selector上的SelectableChannel
			//返回值代表了有多少个channel可以进行IO操作
			keysAdded=selector.select();
			System.out.println("keysAdded="+keysAdded);
			Set<SelectionKey> readyKeys=selector.selectedKeys();
			Iterator<SelectionKey> iter=readyKeys.iterator();
			while(iter.hasNext()){
				//获取一个SelectionKey
				SelectionKey key=iter.next();
				//将当前key从readyKeys集合中删除,避免后续重复操作
				iter.remove();
				//
				if(key.isAcceptable()){//如果有ACCEPT_OP事件发生，即有连接接入
					//从对key的到对应的channel
					ServerSocketChannel myServerChannel=
							(ServerSocketChannel) key.channel();
					
					//接受新的TCP连接,将当前时间写入新的TCP连接中
					SocketChannel myChannel=myServerChannel.accept();
					//将要发给客户端的数据写入通道
					myChannel.write(ByteBuffer.wrap(
							new String("Message From Server: connection has been connected!").getBytes()));
					
					//为myChannel注册读事件
					myChannel.configureBlocking(false);
					myChannel.register(selector,SelectionKey.OP_READ);
			
				}else if(key.isReadable()){
					//读取
					read(key);
				}
			}//while
		}//while
	}

	

	private void read(SelectionKey key) throws IOException {
		 
		if(!(key.isReadable())){
			return;
		}
		// 服务器可读取消息:得到事件发生的Socket通道 
		SocketChannel myChannel=(SocketChannel) key.channel();
		
		//创建读取数据的缓冲区
		ByteBuffer buf=ByteBuffer.allocate(1024);
		//读取数据
		myChannel.read(buf);
		
		String msg=new String(buf.array());
		System.out.println("server receive message:msg="+msg);

		//向client端写数据
		myChannel.write(ByteBuffer.wrap(
				new String("from server:"+"hello"+System.currentTimeMillis()).getBytes()));
	}

	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		// TODO Auto-generated method stub

		System.out.println("begin");

		NIOServer server =new NIOServer();
		server.listen();
		
		System.out.println("end");
		
	}
}

Client端代码:

package rpc.nio;
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;
import java.util.Set;

public class NIOServer {
	
	private Selector selector;
	
	public NIOServer() throws IOException{
		//ͨ通道管理器
		this.selector=Selector.open();
		//创建一个ServerSocketChannel
		ServerSocketChannel serverChannel=ServerSocketChannel.open();
		//设置成non-block模式
		serverChannel.configureBlocking(false);
		//将serverChannel对应的server socket绑定IP 和 端口
		serverChannel.socket().bind(new InetSocketAddress("localhost",8001));
		//将serverChannel注册到selector	
		serverChannel.register(selector, SelectionKey.OP_ACCEPT);
	}
	
	public void listen() throws IOException{
		while(true){
			
			int keysAdded=0;
			//用select()函数来监控注册在selector上的SelectableChannel
			//返回值代表了有多少个channel可以进行IO操作
			keysAdded=selector.select();
			System.out.println("keysAdded="+keysAdded);
			Set<SelectionKey> readyKeys=selector.selectedKeys();
			Iterator<SelectionKey> iter=readyKeys.iterator();
			while(iter.hasNext()){
				//获取一个SelectionKey
				SelectionKey key=iter.next();
				//将当前key从readyKeys集合中删除,避免后续重复操作
				iter.remove();
				//
				if(key.isAcceptable()){//如果有ACCEPT_OP事件发生，即有连接接入
					//从对key的到对应的channel
					ServerSocketChannel myServerChannel=
							(ServerSocketChannel) key.channel();
					
					//接受新的TCP连接,将当前时间写入新的TCP连接中
					SocketChannel myChannel=myServerChannel.accept();
					//将要发给客户端的数据写入通道
					myChannel.write(ByteBuffer.wrap(
							new String("Message From Server: connection has been connected!").getBytes()));
					
					//为myChannel注册读事件
					myChannel.configureBlocking(false);
					myChannel.register(selector,SelectionKey.OP_READ);
			
				}else if(key.isReadable()){
					//读取
					read(key);
				}
			}//while
		}//while
	}

	

	private void read(SelectionKey key) throws IOException {
		 
		if(!(key.isReadable())){
			return;
		}
		// 服务器可读取消息:得到事件发生的Socket通道 
		SocketChannel myChannel=(SocketChannel) key.channel();
		
		//创建读取数据的缓冲区
		ByteBuffer buf=ByteBuffer.allocate(1024);
		//读取数据
		myChannel.read(buf);
		
		String msg=new String(buf.array());
		System.out.println("server receive message:msg="+msg);

		//向client端写数据
		myChannel.write(ByteBuffer.wrap(
				new String("from server:"+"hello"+System.currentTimeMillis()).getBytes()));
	}

	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		// TODO Auto-generated method stub

		System.out.println("begin");

		NIOServer server =new NIOServer();
		server.listen();
		
		System.out.println("end");
		
	}
}

[Java] Java NIO 详解

继续阅读

Java小案例——随机数猜测随机数猜测

nginx location中斜线的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的简单使用

neo4j之cypher使用文档

Ambari介绍和架构原理

GitHub连夜封杀！这份阿里 10W 字内部 Java 字面试手册到底有多强？

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

mybatis_入门程序Mybatis入门

AOP编程_Android优雅权限框架(1)概念基础，2021金三银四前言正文大纲正文

Effective Java 8:通用程序设计

OOM三种类型

工厂模式-三种类型

【递归】高效率求2的n次幂

win10本地scala和spark安装安装scala安装spark

scala (3) Function 和 Method