java NIO中的Selector SelectableChannel SelectionKey

推荐参考：
http://rox-xmlrpc.sourceforge.net/niotut/index.html

下面是我写的一个小demo:

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.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.text.MessageFormat;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author chega
 * 
 *         2012-12-3下午6:41:21
 * 
 * 为每个客户端分配一个线程来执行，执行完后客户端连接并不关闭（即长连接）
 */
public class MyServer
{
	Selector							selector;
	Map<SocketChannel, StringBuilder>	channelMap;
	CharsetDecoder						decoder		= Charset.forName("utf-8").newDecoder();
	ExecutorService						executors	= Executors.newCachedThreadPool();
	private MessageFormat				format		= new MessageFormat("{0, time, medium}, {1}");

	public static void main(String... args)
	{
		try
		{
			new MyServer().init();
		}
		catch (IOException e)
		{
			e.printStackTrace();
		}
	}

	private void init() throws IOException
	{
		this.channelMap = Collections.synchronizedMap(new HashMap<SocketChannel, StringBuilder>());

		selector = Selector.open();
		ServerSocketChannel channel = ServerSocketChannel.open();
		channel.configureBlocking(false);
		channel.socket().bind(new InetSocketAddress("localhost", 80));
		print("ServerSocket绑定在本机80端口，等待客户端连接");
		channel.register(selector, SelectionKey.OP_ACCEPT);
		while (true)
		{
			int count = this.selector.select();
			if (count == 0)
			{
				continue;
			}
			Iterator<SelectionKey> it = this.selector.selectedKeys().iterator();
			while (it.hasNext())
			{
				final SelectionKey key = it.next();
				it.remove();
				if (!key.isValid())
				{
					continue;
				}
				if (key.isAcceptable())
				{
					this.accept(key);
				}
				else if (key.isReadable())
				{
					this.read(key);
//					Server端用一个线程来处理一个客户端的请求，在read完之后立即向客户端写数据，因此不需要再注册SeletionKey.OP_WRITE
//					由于线程是在线程池中来运行的，因此一定要把当然key的SelectionKey.OP_READ事件给取消掉，否则会多次执行read方法
					key.interestOps(0);
				}
			}
		}
	}

	/**
	 * @param key
	 */
	private void accept(SelectionKey key)
	{
		try
		{
			SocketChannel c = ((ServerSocketChannel) key.channel()).accept();
			c.configureBlocking(false);
			c.register(selector, SelectionKey.OP_READ);
			this.print(c + "连接成功，并注册READ事件");
		}
		catch (IOException e)
		{
			e.printStackTrace();
		}
	}

	/**
	 * @param key
	 */
	private void read(final SelectionKey key)
	{
		this.executors.execute(new Runnable() {
			@Override
			public void run()
			{
				try
				{
					int num = -1;
					SocketChannel c = (SocketChannel) key.channel();
					ByteBuffer buffer = ByteBuffer.allocate(1024);
					while (c.isOpen())
					{
						num = c.read(buffer);
						if (num < 1)
						{
							break;
						}
						buffer.flip();
						MyServer.this.put2Map(c, buffer);
						buffer.clear();
					}
					if (num == -1)
					{
						c.close();
						key.channel();
						print(c + "has been closed");
						MyServer.this.channelMap.remove(c);
						return;
					}
					print("从" + c + "读取数据完毕");

					if (c.isOpen() && MyServer.this.channelMap.get(c) != null && MyServer.this.channelMap.get(c).length() != 0)
					{
						MyServer.this.channelMap.get(c).trimToSize();
						String str = "echo: " + MyServer.this.channelMap.get(c);
						c.write(ByteBuffer.wrap(str.getBytes()));
						print("向" + c + "写数据完毕");
						MyServer.this.put2Map(c, null);
//						由于Selector线程中已经取消了OP_READ事件，因此这里再加上OP_READ
						key.interestOps(SelectionKey.OP_READ);
						key.selector().wakeup();
					}
				}
				catch (IOException e)
				{
					e.printStackTrace();
					key.channel();
					try
					{
						key.channel().close();
					}
					catch (IOException e1)
					{
						e1.printStackTrace();
					}
				}
			}
		});
	}

	private void put2Map(SocketChannel channel, ByteBuffer buffer)
	{
		if (this.channelMap.get(channel) == null)
		{
			this.channelMap.put(channel, new StringBuilder());
		}
		if (buffer == null)
		{
			this.channelMap.get(channel).setLength(0);
			return;
		}
		try
		{
			this.channelMap.get(channel).append(decoder.decode(buffer).toString());
		}
		catch (CharacterCodingException e)
		{
			e.printStackTrace();
		}
	}

	private void print(String msg)
	{
		System.out.println(this.format.format(new Object[] { new Date(), msg }));
	}
}

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.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.text.MessageFormat;
import java.util.Date;
import java.util.Iterator;

/**
 * chega
 * 2012-11-21下午3:19:52
 */

/**
 * @author chega
 * 
 *         2012-11-21下午3:19:52
 * 
 */
public class MyClient
{
	Selector				selector;
	InetSocketAddress		address;
	ByteBuffer				buffer;
	CharsetDecoder			decoder;
	private MessageFormat	format	= new MessageFormat("{0, time, medium}, {1}");
	final String			TEST	= "ntp";

	public static void main(String[] args) throws IOException
	{
		MyClient t = new MyClient();
		t.selector = Selector.open();
		t.address = new InetSocketAddress("localhost", 80);
		t.buffer = ByteBuffer.allocate(1024);
		t.decoder = Charset.forName("utf-8").newDecoder();
		t.run();
	}

	private void run() throws IOException
	{
		this.createNewConnection();
		while (true)
		{
			try
			{
				int selectedCount = selector.select();
				if (selectedCount == 0)
				{
					continue;
				}
				Iterator<SelectionKey> it = selector.selectedKeys().iterator();
				while (it.hasNext())
				{
					SelectionKey key = it.next();
					it.remove();// 从己选择键中删除，否则该键一直存在
					if (!key.isValid())
					{
						continue;
					}
					if (key.isConnectable())
					{
						this.connect(key);
					}
					else if (key.isReadable())
					{
						this.read(key);
					}
					else if (key.isWritable())
					{
						this.write(key);
					}
				}
			}
			catch (IOException e)
			{
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param key
	 * @throws IOException
	 * 
	 */
	private void connect(SelectionKey key)
	{
		try
		{
			while (!((SocketChannel) key.channel()).finishConnect())
			{
				try
				{
					Thread.sleep(100);
				}
				catch (InterruptedException e)
				{
					e.printStackTrace();
				}
				continue;
			}
			key.interestOps(SelectionKey.OP_WRITE);
			print(key.channel() + "连接服务器成功，并注册WRITE事件");
		}
		catch (IOException e)
		{
			e.printStackTrace();
			print(key.channel() + "连接失败");
			System.out.println(((SocketChannel) key.channel()).socket().getLocalPort());
			System.exit(-1);
		}
	}

	/**
	 * @param key
	 */
	private void write(SelectionKey key)
	{
		try
		{
			// ((SocketChannel) key.channel()).write(ByteBuffer.wrap("GET / HTTP/1.0\r\n\r\n".getBytes()));
			((SocketChannel) key.channel()).write(ByteBuffer.wrap(TEST.getBytes()));
			key.interestOps(SelectionKey.OP_READ);
			print(key.channel() + "向服务器完数据完毕，并注册READ事件");
		}
		catch (IOException e)
		{
			e.printStackTrace();
			print(key.channel() + "写数据发生错误");
			try
			{
				key.channel().close();
			}
			catch (IOException e1)
			{
				e1.printStackTrace();
			}
		}
	}

	/**
	 * @param key
	 */
	private void read(SelectionKey key)
	{
		try
		{
			int num = -1;
			buffer.clear();
			while (key.channel().isOpen())
			{
				num = ((SocketChannel) key.channel()).read(buffer);
				if (num < 1)
				{
					break;
				}
				buffer.flip();
				String result = this.decoder.decode(buffer).toString();
				System.out.println(result);
				buffer.clear();
			}
//			sleep 5s and write again
			Thread.sleep(5000);
			key.interestOps(SelectionKey.OP_WRITE);
		}
		catch (IOException e)
		{
			e.printStackTrace();
			print(key.channel() + "读取数据发生错误");
			try
			{
				key.channel().close();
			}
			catch (IOException e1)
			{
				e1.printStackTrace();
			}
		}
		catch (InterruptedException e)
		{
			e.printStackTrace();
		}
	}

	private void createNewConnection() throws IOException
	{
		SocketChannel channel = SocketChannel.open();
		channel.configureBlocking(false);
		channel.connect(address);
		channel.register(selector, SelectionKey.OP_CONNECT);
	}

	private void print(String msg)
	{
		System.out.println(this.format.format(new Object[] { new Date(), msg }));
	}
}