nioutil.java

sea是一个基于seda模式的实现。这个设计模式将系统分为很多stage。每个stage分布不同的任务（基于线程池）。通过任务流的方式提高系统的效率。

/*
 * Copyright (c) 2003, The Regents of the University of California, through
 * Lawrence Berkeley National Laboratory (subject to receipt of any required
 * approvals from the U.S. Dept. of Energy). All rights reserved.
 */
package gov.lbl.dsd.sea.nio.util;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.InvalidMarkException;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.NonReadableChannelException;
import java.nio.channels.NonWritableChannelException;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.WritableByteChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

/**
 * Various utilities related to the {@link java.nio} package.
 * 
 * @author [e-mail]
 * @author $Author: hoschek3 $
 * @version $Revision: 1.8 $, $Date: 2004/08/17 20:17:22 $
 */
public class NioUtil {

	private static final org.apache.commons.logging.Log log = org.apache.commons.logging.LogFactory.getLog(NioUtil.class);

	private NioUtil() {} // not instantiable
	
	/**
	 * Creates and returns a non-blocking TCP server channel bound to the given
	 * network port.
	 * 
	 * @param port
	 *            the port to bind to.
	 * @return the channel
	 * @throws IOException if the bind operation fails, or if the socket
	 *			   is already bound or if some other I/O error occurs
	 */
	public static ServerSocketChannel createServerChannel(int port) throws IOException {
		return createServerChannel(new InetSocketAddress(port), 0);
	}
	
	/**
	 * Creates and returns a non-blocking TCP server channel bound to the given
	 * network address.
	 * <p>
	 * If a connection indication arrives when the backlog queue is full, the
	 * connection is refused. The <code>backlog</code> argument must be a
	 * positive value greater than 0. If the value passed if equal or less than
	 * 0, then the default value will be assumed.
	 * 
	 * @param address
	 *            the address to bind to.
	 * @param backlog
	 *            The listen backlog length.
	 * @return the channel
	 * @throws IOException if the bind operation fails, or if the socket
	 *			   is already bound or if some other I/O error occurs
	 */ 
	public static ServerSocketChannel createServerChannel(SocketAddress address, int backlog) throws IOException {
		ServerSocketChannel channel = ServerSocketChannel.open();
		channel.configureBlocking(false);
		channel.socket().bind(address, backlog);
		return channel;
	}
	
	/**
	 * Creates and returns a non-blocking TCP server channel bound to the given
	 * network host and port.
	 * 
	 * @param hostName
	 *            the host to bind to.
	 * @param port
	 *            the port to bind to.
	 * @return the channel
	 * @throws  IOException If an I/O error occurs
	 */
	public static SocketChannel createClientChannel(String hostName, int port) throws IOException {
		return createClientChannel(new InetSocketAddress(hostName, port));
	}
	
	/**
	 * Creates and returns a non-blocking TCP client channel bound to the given
	 * network address.
	 * 
	 * @param address
	 *            the address to bind to.
	 * @return the channel
	 * @throws  IOException If an I/O error occurs
	 */ 
	public static SocketChannel createClientChannel(SocketAddress address) throws IOException {
		SocketChannel channel = SocketChannel.open();
		channel.configureBlocking(false);
		long start = System.currentTimeMillis();
		channel.connect(address); 
		long end = System.currentTimeMillis();
		log.debug("connect took [ms]=" + (end-start));
		return channel;
	}
	
	/**
	 * Creates and returns a non-blocking UDP datagram channel bound to the given network
	 * port.
	 * 
	 * @param port
	 *            the port to bind to.
	 * @return the channel
	 * @throws  IOException If an I/O error occurs
	 */ 
	public static DatagramChannel createDatagramChannel(int port) throws IOException {
		return createDatagramChannel(new InetSocketAddress(port));
	}
	
	/**
	 * Creates and returns a non-blocking UDP datagram channel bound to the given network
	 * address.
	 * 
	 * @param address
	 *            the address to bind to.
	 * @return the channel
	 * @throws  IOException If an I/O error occurs
	 */ 
	public static DatagramChannel createDatagramChannel(SocketAddress address) throws IOException {
		DatagramChannel channel = DatagramChannel.open();
		channel.configureBlocking(false);
		channel.socket().bind(address);
		return channel;
	}

	/**
	 * Returns whether or not a given buffer has the mark defined.
	 * @param buffer the buffer to check
	 * @return true if it has a mark
	 */
	public static boolean hasMark(Buffer buffer) {
		// unfortunately this seems to be the only way to figure it out :-(
		boolean hasMark = true;
		int pos = buffer.position();
		try {
			buffer.reset();
			buffer.position(pos);
		}
		catch (InvalidMarkException e) {
			hasMark = false;
		}
		return hasMark;
	}

	/**
	 * Efficiently reads (without ever blocking) as many bytes as possible from
	 * the given non-blocking channel into the given buffer. Returns the total
	 * number of bytes that have been read. The algorithm strongly improves
	 * throughput at the expense of somewhat increased latency. This method
	 * simply propagates exceptions thrown in the underlying
	 * {@link ReadableByteChannel#read(ByteBuffer)}NIO code - it never causes
	 * an exception itself.
	 * 
	 * @param channel -
	 *            the channel to read from
	 * @param buffer -
	 *            the buffer to read into
	 * @return the total number of bytes read, possibly zero. If end-of-stream
	 *         (EOS) is encountered then the number returned is precisely
	 *         <code>-(total+1)</code>. Hence, if EOS is encountered the
	 *         returned number is always &lt; 0, and else &gt;= 0. In the EOS
	 *         case, the caller can determine the total number of read bytes
	 *         with <code>-(readMany(...)+1)</code>. (The same trick is used
	 *         in {@link java.util.Arrays#binarySearch(long[], long)}to
	 *         effectively return two return parameters within one parameter).
	 * 
	 * <p>
	 * Example usage:
	 * 
	 * <pre>
	 * 
	 * 
	 * int total = readMany(channel, buffer);
	 * boolean eos = false;
	 * if (total &lt; 0) {
	 * 	eos = true;
	 * 	total = -(total + 1);
	 * }
	 * 
	 * System.out.println(total + &quot;bytes read&quot;);
	 * 
	 * // some meaningful processing of the read bytes goes here
	 * 
	 * if (eos) {
	 * 	System.out.println(total + &quot;bytes read, EOS reached, now closing channel&quot;);
	 * 	channel.close();
	 * 	System.exit(0);
	 * }
	 * 
	 * 
	 * </pre>
	 * 
	 * @throws NonReadableChannelException
	 *             If this channel was not opened for reading
	 * 
	 * @throws ClosedChannelException
	 *             If this channel is closed
	 * 
	 * @throws AsynchronousCloseException
	 *             If another thread closes this channel while the read
	 *             operation is in progress
	 * 
	 * @throws ClosedByInterruptException
	 *             If another thread interrupts the current thread while the
	 *             read operation is in progress, thereby closing the channel
	 *             and setting the current thread's interrupt status
	 * 
	 * @throws IOException
	 *             If some other I/O error occurs
	 */
	public static int readMany(ReadableByteChannel channel, ByteBuffer buffer)
			throws IOException {
		// total  -->   -(total+1)   -->   -(newtotal+1) 
		//
		//   0    -->      -1        -->        0
		//   1    -->      -2        -->        1
		//   2    -->      -3        -->        2
		
		/*
		 * For efficiency try to iterate more than once even when zero or only
		 * very few bytes are currently read-ready. The way java.nio is
		 * implemented, there is a high chance that almost immediately after the
		 * first few bytes become ready, a much larger block becomes ready. We
		 * try to wait for larger block in a short-lived, tight spin loop,