readerhandler.java

java NIO源代码,初学者学习java NIO技术的基础代码

package org.Arios.XHandler;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import org.Arios.XFactory.ReaderThreadFactory;
import org.Arios.XWrap.Request;

public class ReaderHandler implements Runnable {
	// private static Notifier notifier = Notifier.getNotifier();

	private SelectionKey key;
	private static int BUFFER_SIZE = 1024;

	public ReaderHandler(SelectionKey key) {
		this.key = key;
	}

	public void run() {
		// 读取数据
		read(key);
	}

	/**
	 * 读取客户端发出请求数据
	 * 
	 * @param sc
	 *            套接通道
	 * 
	 */
	public byte[] readRequest(SocketChannel sc) throws IOException {
		ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
		int off = 0;
		int r = 0;
		byte[] data = new byte[BUFFER_SIZE * 10];

		while (true) {
			buffer.clear();
			r = sc.read(buffer);
			if (r == -1 || r == 0)
				break;
			if ((off + r) > data.length) {
				data = grow(data, BUFFER_SIZE * 10);
			}
			byte[] buf = buffer.array();
			System.arraycopy(buf, 0, data, off, r);
			off += r;
		}
		byte[] req = new byte[off];
		System.arraycopy(data, 0, req, 0, off);
		return req;
	}

	/**
	 * 处理连接数据读取
	 * 
	 * @param key
	 *            SelectionKey
	 */
	public void read(SelectionKey key) {
		Request request = (Request) key.attachment();
		try {
			// 读取客户端数据
			SocketChannel sc = (SocketChannel) key.channel();
			byte[] clientData = readRequest(sc);

			request.setDataInput(clientData);
			request.attach(1);

			// 触发onRead
			// notifier.callOnRead(request);

			// 提交主控线程进行写处理
			key.interestOps(SelectionKey.OP_WRITE);
		} catch (Exception e) {
			try {
				// notifier.callOnClosed(request);
			} catch (Exception e1) {
				e1.printStackTrace();
			}
		}
	}

	/**
	 * 数组扩容
	 * 
	 * @param src
	 *            byte[] 源数组数据
	 * @param size
	 *            int 扩容的增加量
	 * @return byte[] 扩容后的数组
	 */
	public static byte[] grow(byte[] src, int size) {
		byte[] tmp = new byte[src.length + size];
		System.arraycopy(src, 0, tmp, 0, src.length);
		return tmp;
	}
}