Java nio

本文转载自http://www.iteye.com/topic/834447

1. 基本概念
    IO 是主存和外部设备 ( 硬盘、终端和网络等 ) 拷贝数据的过程。 IO 是操作系统的底层功能实现，底层通过 I/O 指令进行完成。
    所有语言运行时系统提供执行 I/O 较高级别的工具。 (c 的 printf scanf,java 的面向对象封装 )
2.Java标准io回顾
    Java 标准 IO 类库是 io 面向对象的一种抽象。基于本地方法的底层实现，我们无须关注底层实现。 InputStream\OutputStream( 字节流 ) ：一次传送一个字节。 Reader\Writer( 字符流 ) ：一次一个字符。
3.nio 简介
    nio 是 java New IO 的简称，在 jdk1.4 里提供的新 api 。 Sun 官方标榜的特性如下：
–为所有的原始类型提供 (Buffer) 缓存支持。
–字符集编码解码解决方案。
–Channel ：一个新的原始 I/O 抽象。
–支持锁和内存映射文件的文件访问接口。
–提供多路 (non-bloking) 非阻塞式的高伸缩性网络 I/O 。
4.Buffer&Chanel
    Channel 和 buffer 是 NIO 是两个最基本的数据类型抽象。
Buffer:
–是一块连续的内存块。
–是 NIO 数据读或写的中转地。
Channel:
–数据的源头或者数据的目的地
–用于向 buffer 提供数据或者读取 buffer 数据 ,buffer 对象的唯一接口。
–异步 I/O 支持
    下面代码使用nio实现文件的拷贝

package com.ajita;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

public class CopyFile {
	public static void main(String[] args) throws Exception {
		String infile = "C:\\log.txt";
		String outfile = "D:\\log.txt";
		// 获取源文件和目标文件的输入输出流
		FileInputStream fin = new FileInputStream(infile);
		FileOutputStream fout = new FileOutputStream(outfile);
		// 获取输入输出通道
		FileChannel fcin = fin.getChannel();
		FileChannel fcout = fout.getChannel();
		// 创建缓冲区
		ByteBuffer buffer = ByteBuffer.allocate(1024);
		while (true) {
			// clear方法重设缓冲区，使它可以接受读入的数据
			buffer.clear();
			// 从输入通道中将数据读到缓冲区
			int r = fcin.read(buffer);
			// read方法返回读取的字节数，可能为零，如果该通道已到达流的末尾，则返回-1
			if (r == -1) {
				break;
			}
			// flip方法让缓冲区可以将新读入的数据写入另一个通道
			buffer.flip();
			// 从输出通道中将数据写入缓冲区
			fcout.write(buffer);
		}
	}
}

    一个 buffer 主要由 position,limit,capacity 三个变量来控制读写的过程。此三个变量的含义见如下表格：

参数	写模式	读模式
position	当前写入的单位数据数量。	当前读取的单位数据位置。
limit	代表最多能写多少单位数据和容量是一样的。	代表最多能读多少单位数据，和之前写入的单位数据量一致。
capacity	buffer 容量	buffer 容量

Buffer 常见方法：
flip(): 写模式转换成读模式
rewind() ：将 position 重置为 0 ，一般用于重复读。
clear() ：清空 buffer ，准备再次被写入 (position 变成 0 ， limit 变成 capacity) 。
compact(): 将未读取的数据拷贝到 buffer 的头部位。
mark() 、 reset():mark 可以标记一个位置， reset 可以重置到该位置。

    Buffer 常见类型： ByteBuffer 、 MappedByteBuffer 、 CharBuffer 、 DoubleBuffer 、 FloatBuffer 、 IntBuffer 、 LongBuffer 、 ShortBuffer 。

    channel 常见类型 :FileChannel 、 DatagramChannel(UDP) 、 SocketChannel(TCP) 、 ServerSocketChannel(TCP)
5.nio.charset
    字符编码解码 : 字节码本身只是一些数字，放到正确的上下文中被正确被解析。向 ByteBuffer 中存放数据时需要考虑字符集的编码方式，读取展示 ByteBuffer 数据时涉及对字符集解码。
    Java.nio.charset 提供了编码解码一套解决方案。
以我们最常见的 http 请求为例，在请求的时候必须对请求进行正确的编码。在得到响应时必须对响应进行正确的解码。

以下代码向 baidu 发一次请求，并获取结果进行显示。例子演示到了 charset 的使用。

package com.ajita;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class BaiduReader {
	private Charset charset = Charset.forName("GBK");// 创建GBK字符集
	private SocketChannel channel;

	public void readHTMLContent() {
		try {
			InetSocketAddress socketAddress = new InetSocketAddress(
					"www.baidu.com", 80);
			// step1:打开连接
			channel = SocketChannel.open(socketAddress);
			// step2:发送请求，使用GBK编码
			channel.write(charset.encode("GET " + "/ HTTP/1.1" + "\r\n\r\n"));
			// step3:读取数据
			ByteBuffer buffer = ByteBuffer.allocate(1024);// 创建1024字节的缓冲
			while (channel.read(buffer) != -1) {
				buffer.flip();// flip方法在读缓冲区字节操作之前调用。
				System.out.println(charset.decode(buffer));
				// 使用Charset.decode方法将字节转换为字符串
				buffer.clear();// 清空缓冲
			}
		} catch (IOException e) {
			System.err.println(e.toString());
		} finally {
			if (channel != null) {
				try {
					channel.close();
				} catch (IOException e) {
				}
			}
		}
	}

	public static void main(String[] args) {
		new BaiduReader().readHTMLContent();
	}
}

6.非阻塞 IO
非阻塞的原理
    把整个过程切换成小的任务，通过任务间协作完成。
    由一个专门的线程来处理所有的 IO 事件，并负责分发。
    事件驱动机制：事件到的时候触发，而不是同步的去监视事件。
    线程通讯：线程之间通过 wait,notify 等方式通讯。保证每次上下文切换都是有意义的。减少无谓的进程切换。

异步 IO 核心 API
Selector
    异步 IO 的核心类，它能检测一个或多个通道 (channel) 上的事件，并将事件分发出去。
    使用一个 select 线程就能监听多个通道上的事件，并基于事件驱动触发相应的响应。而不需要为每个 channel 去分配一个线程。
SelectionKey
    包含了事件的状态信息和时间对应的通道的绑定。
例子 单线程实现监听两个端口。 ( 见 nio.asyn 包下面的例子。)

package com.ajita.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.nio.channels.spi.SelectorProvider;
import java.util.Iterator;

public class OperationServer implements Runnable {
	private int port1 = 8090;
	private int port2 = 9090;
	private Selector selector;
	private ServerSocketChannel serverChannel1;
	private ByteBuffer readBuffer = ByteBuffer.allocate(8192);
	private ServerSocketChannel serverChannel2;
	private SocketChannel socketChannel1;
	private SocketChannel socketChannel2;
	private AddProcessor client1Processor = new AddProcessor();
	private MultiProcessor client2Processor = new MultiProcessor();

	public OperationServer() {
		initSelector();
	}

	public void run() {
		while (true) {
			try {
				this.selector.select();
				Iterator<SelectionKey> selectedKeys = this.selector.selectedKeys().iterator();
				while (selectedKeys.hasNext()) {
					SelectionKey key = (SelectionKey) selectedKeys.next();
					selectedKeys.remove();

					if (!key.isValid()) {
						continue;
					}

					if (key.isAcceptable()) {
						this.accept(key);
					} else if (key.isReadable()) {
						this.read(key);
					} else if (key.isWritable()) {
						this.write(key);
					}
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}

	public void accept(SelectionKey key) throws IOException {
		ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key
				.channel();
		if (serverSocketChannel.equals(serverChannel1)) {
			socketChannel1 = serverSocketChannel.accept();
			socketChannel1.configureBlocking(false);
			socketChannel1.register(this.selector, SelectionKey.OP_READ);
		} else {
			socketChannel2 = serverSocketChannel.accept();
			socketChannel2.configureBlocking(false);
			socketChannel2.register(this.selector, SelectionKey.OP_READ);
		}

	}

	public void read(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();

		this.readBuffer.clear();

		// Attempt to read off the channel
		int numRead;
		try {
			numRead = socketChannel.read(this.readBuffer);
		} catch (IOException e) {
			// The remote forcibly closed the connection, cancel
			// the selection key and close the channel.
			key.cancel();
			socketChannel.close();
			return;
		}

		if (numRead == -1) {
			// Remote entity shut the socket down cleanly. Do the
			// same from our end and cancel the channel.
			key.channel().close();
			key.cancel();
			return;
		}
		String input = new String(readBuffer.array()).trim();
		if (socketChannel.equals(socketChannel1)) {
			client1Processor.process(input);
		} else {
			client2Processor.process(input);
		}
	}

	public void write(SelectionKey key) {

	}

	/**
	 * 注册事件到selector；
	 */
	public void initSelector() {
		try {
			selector = SelectorProvider.provider().openSelector();
			this.serverChannel1 = ServerSocketChannel.open();
			serverChannel1.configureBlocking(false);
			InetSocketAddress isa = new InetSocketAddress("localhost",
					this.port1);
			serverChannel1.socket().bind(isa);
			serverChannel1.register(selector, SelectionKey.OP_ACCEPT);

			this.serverChannel2 = ServerSocketChannel.open();
			serverChannel2.configureBlocking(false);
			InetSocketAddress isa2 = new InetSocketAddress("localhost",
					this.port2);
			serverChannel2.socket().bind(isa2);
			serverChannel2.register(selector, SelectionKey.OP_ACCEPT);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		OperationServer server = new OperationServer();
		Thread t = new Thread(server);
		t.start();
	}
}

package com.ajita.nio;

public class AddProcessor {
	public void process(String input) {
		if (input != null) {
			String[] elements = input.split(",");
			if (elements.length != 2) {
				System.out
						.println("sorry, input expression error! right format:A+B");
				return;
			}
			double A = Double.parseDouble(elements[0]);
			double B = Double.parseDouble(elements[1]);

			System.out.println(A + "+" + B + "=" + (A + B));
		} else {
			System.out.println("no input");
		}

	}
}

package com.ajita.nio;

public class MultiProcessor {
	public void process(String input) {
		if (input != null) {
			String[] elements = input.split(",");
			if (elements.length != 2) {
				System.out
						.println("sorry, input expression error! right format:A*B");
				return;
			}
			double A = Double.parseDouble(elements[0]);
			double B = Double.parseDouble(elements[1]);

			System.out.println(A + "*" + B + "=" + (A * B));
		} else {
			System.out.println("no input");
		}

	}
}

package com.ajita.nio;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class OperationClient {
	// Charset and decoder for US-ASCII
	private static Charset charset = Charset.forName("US-ASCII");

	// Direct byte buffer for reading
	private static ByteBuffer dbuf = ByteBuffer.allocateDirect(1024);

	// Ask the given host what time it is
	//
	private static void query(String host, int port) throws IOException {
		byte inBuffer[] = new byte[100];
		InetSocketAddress isa = new InetSocketAddress(
				InetAddress.getByName(host), port);
		SocketChannel sc = null;
		while (true) {
			try {
				System.in.read(inBuffer);
				sc = SocketChannel.open();
				sc.connect(isa);
				dbuf.clear();
				dbuf.put(inBuffer);
				dbuf.flip();
				sc.write(dbuf);
				dbuf.clear();

			} finally {
				// Make sure we close the channel (and hence the socket)
				if (sc != null)
					sc.close();
			}
		}
	}

	public static void main(String[] args) throws IOException {
		query("localhost", 8090);// A+B
		// query("localhost", 9090);//A*B
	}
}