selectsocketsthreadpool.java

java nio 编程一个实例子.服务端程序

package com.ronsoft.books.nio.channels;

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

/**
 * Specialization of the SelectSockets class which uses a thread pool
 * to service channels.  The thread pool is an ad-hoc implementation
 * quicky lashed togther in a few hours for demonstration purposes.
 * It's definitely not production quality.
 *
 * Created May 2002
 * @author Ron Hitchens ([e-mail])
 * @version $Id: SelectSocketsThreadPool.java,v 1.5 2002/05/20 07:24:29 ron Exp $
 */
public class SelectSocketsThreadPool extends SelectSockets
{
	private static final int MAX_THREADS = 5;

	private ThreadPool pool = new ThreadPool (MAX_THREADS);

	// -------------------------------------------------------------

	public static void main (String [] argv)
		throws Exception
	{
		new SelectSocketsThreadPool().go (argv);
	}

	// -------------------------------------------------------------

	/**
	 * Sample data handler method for a channel with data ready to read.
	 * This method is invoked from the go() method in the parent class.
	 * This handler delegates to a worker thread in a thread pool to
	 * service the channel, then returns immediately.
	 * @param key A SelectionKey object representing a channel
	 *  determined by the selector to be ready for reading.  If the
	 *  channel returns an EOF condition, it is closed here, which
	 *  automatically invalidates the associated key.  The selector
	 *  will then de-register the channel on the next select call.
	 */
	protected void readDataFromSocket (SelectionKey key)
		throws Exception
	{
		WorkerThread worker = pool.getWorker();

		if (worker == null) {
			// No threads available, do nothing, the selection
			// loop will keep calling this method until a
			// thread becomes available.  This design could
			// be improved.
			return;	
		}

		// invoking this wakes up the worker thread then returns
		worker.serviceChannel (key);
	}

	// ---------------------------------------------------------------

	/**
	 * A very simple thread pool class.  The pool size is set at
	 * construction time and remains fixed.  Threads are cycled
	 * through a FIFO idle queue.
	 */
	private class ThreadPool
	{
		List idle = new LinkedList();

		ThreadPool (int poolSize)
		{
			// fill up the pool with worker threads
			for (int i = 0; i < poolSize; i++) {
				WorkerThread thread = new WorkerThread (this);

				// set thread name for debugging, start it
				thread.setName ("Worker" + (i + 1));
				thread.start();

				idle.add (thread);
			}
		}

		/**
		 * Find an idle worker thread, if any.  Could return null.
		 */
		WorkerThread getWorker()
		{
			WorkerThread worker = null;

			synchronized (idle) {
				if (idle.size() > 0) {
					worker = (WorkerThread) idle.remove (0);
				}
			}

			return (worker);
		}

		/**
		 * Called by the worker thread to return itself to the
		 * idle pool.
		 */
		void returnWorker (WorkerThread worker)
		{
			synchronized (idle) {
				idle.add (worker);
			}
		}
	}

	/**
	 * A worker thread class which can drain channels and echo-back
	 * the input.  Each instance is constructed with a reference to
	 * the owning thread pool object. When started, the thread loops
	 * forever waiting to be awakened to service the channel associated
	 * with a SelectionKey object.
	 * The worker is tasked by calling its serviceChannel() method
	 * with a SelectionKey object.  The serviceChannel() method stores
	 * the key reference in the thread object then calls notify()
	 * to wake it up.  When the channel has been drained, the worker
	 * thread returns itself to its parent pool.
	 */
	private class WorkerThread extends Thread
	{
		private ByteBuffer buffer = ByteBuffer.allocate (1024);
		private ThreadPool pool;
		private SelectionKey key;

		WorkerThread (ThreadPool pool)
		{
			this.pool = pool;
		}

		// loop forever waiting for work to do
		public synchronized void run()
		{
			System.out.println (this.getName() + " is ready");

			while (true) {
				try {
					// sleep and release object lock
					this.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
					// clear interrupt status
					this.interrupted();
				}

				if (key == null) {
					continue;	// just in case
				}

				System.out.println (this.getName()
					+ " has been awakened");

				try {
					drainChannel (key);
				} catch (Exception e) {
					System.out.println ("Caught '"
						+ e + "' closing channel");

					// close channel and nudge selector
					try {
						key.channel().close();
					} catch (IOException ex) {
						ex.printStackTrace();
					}

					key.selector().wakeup();
				}

				key = null;

				// done, ready for more, return to pool
				this.pool.returnWorker (this);
			}
		}

		/**
		 * Called to initiate a unit of work by this worker thread
		 * on the provided SelectionKey object.  This method is
		 * synchronized, as is the run() method, so only one key
		 * can be serviced at a given time.
		 * Before waking the worker thread, and before returning
		 * to the main selection loop, this key's interest set is
		 * updated to remove OP_READ.  This will cause the selector
		 * to ignore read-readiness for this channel while the
		 * worker thread is servicing it.
		 */
		synchronized void serviceChannel (SelectionKey key)
		{
			this.key = key;

			key.interestOps (key.interestOps() & (~SelectionKey.OP_READ));

			this.notify();		// awaken the thread
		}

		/**
		 * The actual code which drains the channel associated with
		 * the given key.  This method assumes the key has been
		 * modified prior to invocation to turn off selection
		 * interest in OP_READ.  When this method completes it
		 * re-enables OP_READ and calls wakeup() on the selector
		 * so the selector will resume watching this channel.
		 */
		void drainChannel (SelectionKey key)
			throws Exception
		{
			SocketChannel channel = (SocketChannel) key.channel();
			int count;

			buffer.clear();			// make buffer empty

			// loop while data available, channel is non-blocking
			while ((count = channel.read (buffer)) > 0) {
				buffer.flip();		// make buffer readable

				// send the data, may not go all at once
				while (buffer.hasRemaining()) {
					channel.write (buffer);
				}
				// WARNING: the above loop is evil.
				// See comments in superclass.

				buffer.clear();		// make buffer empty
			}

			if (count < 0) {
				// close channel on EOF, invalidates the key
				channel.close();
				return;
			}

			// resume interest in OP_READ
			key.interestOps (key.interestOps() | SelectionKey.OP_READ);

			// cycle the selector so this key is active again
			key.selector().wakeup();
		}
	}
}