使用Selector的原因

Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。

仅用单个线程来处理多个Channels的好处是，只需要更少的线程来处理通道。事实上，可以只用一个线程处理所有的通道。对于操作系统来说，线程之间上下文切换的开销很大，而且每个线程都要占用系统的一些资源（如内存）。因此，使用的线程越少越好。

但是，需要记住，现代的操作系统和CPU在多任务方面表现的越来越好，所以多线程的开销随着时间的推移，变得越来越小了。实际上，如果一个CPU有多个内核，不使用多任务可能是在浪费CPU能力。不管怎么说，关于那种设计的讨论应该放在另一篇不同的文章中。在这里，只要知道使用Selector能够处理多个通道就足够了。

下面是单线程使用一个Selector处理3个channel的示例图：

范例

说明一下，实际上，对数据的处理应该放在一个专门的线程中来执行，但是我为了使代码更简洁，看起来更容易理解，因此并没有这么做！！！

BaseSocketChannelSelector.java

package com.demo.channel; import java.io.IOException; import java.nio.channels.SelectableChannel; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.util.Iterator; import java.util.Set; public abstract class BaseSocketChannelSelector { private Selector selector; protected BaseSocketChannelSelector() throws IOException { selector = Selector.open(); } protected void registerChannel(SelectableChannel channel, int ops) { try { // 与Selector一起使用时，Channel必须处于非阻塞模式下。这意味着不能将FileChannel与Selector一起使用，因为FileChannel不能切换到非阻塞模式。而套接字通道都可以。 channel.configureBlocking(false); // 可以监听以下四种不同类型的事件 SelectionKey.OP_ACCEPT(server) | SelectionKey.OP_CONNECT | SelectionKey.OP_READ | SelectionKey.OP_WRITE SelectionKey key = channel.register(selector, ops); } catch (IOException e) { e.printStackTrace(); } } public void select() { while (true) { try { /* * 每次使用select时必须判断下selector是否已经被关闭，否则会报如下错误 * Exception in thread "pool-1-thread-1" java.nio.channels.ClosedSelectorException at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:83) */ if (!selector.isOpen()) { System.out.println("selector is closed"); break; } // This method performs a blocking selection operation.It returns only after at least one channel is selected, this selector's wakeup method is invoked, or the current thread is interrupted, whichever comes first int readyChannels = selector.select(); if (readyChannels == 0) { continue; } if (!selector.isOpen()) { System.out.println("selector is closed"); break; } /* * select()阻塞到至少有一个通道在你注册的事件上就绪了。 * select(long timeout)和select()一样，除了最长会阻塞timeout毫秒(参数)。 * selectNow()不会阻塞，不管什么通道就绪都立刻返回（译者注：此方法执行非阻塞的选择操作。如果自从前一次选择操作后，没有通道变成可选择的，则此方法直接返回零。）。 * select()方法返回的int值表示有多少通道已经就绪。亦即，自上次调用select()方法后有多少通道变成就绪状态。 * 如果调用select()方法，因为有一个通道变成就绪状态，返回了1，若再次调用select()方法，如果另一个通道就绪了，它会再次返回1。 * 如果对第一个就绪的channel没有做任何操作，现在就有两个就绪的通道，但在每次select()方法调用之间，只有一个通道就绪了。 * 调用Selector.wakeup()方法即可。阻塞在select()方法上的线程会立马返回。 */ Set<SelectionKey> selectedKeys = selector.selectedKeys(); Iterator<SelectionKey> iterator = selectedKeys.iterator(); while (iterator.hasNext()) { SelectionKey key = iterator.next(); if (key.isValid()) { doOnSelectionKey(key); /* * if (key.isAcceptable()) { * // a connection was accepted by a ServerSocketChannel. * } else if (key.isConnectable()) { * // a connection was established with a remote server. * } else if (key.isReadable()) { * // a channel is ready for reading * } else if (key.isWritable()) { * // a channel is ready for writing * } */ } // 处理完该事件就移除该对象 iterator.remove(); } } catch (IOException e) { e.printStackTrace(); } } } protected abstract void doOnSelectionKey(SelectionKey key); protected void closeSelector() { try { // 闭该Selector，且使注册到该Selector上的所有SelectionKey实例无效。通道本身并不会关闭。 selector.close(); } catch (IOException e) { e.printStackTrace(); } } }

TcpClientChannelSelector.java

package com.demo.channel; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; public class TcpClientChannelSelector extends BaseSocketChannelSelector { private static final String HOSTNAME = "127.0.0.1"; private static final int PORT = 5555; private static final int BUFF_SIZE = 255; private SocketChannel socketChannel; private ByteBuffer readBuffer; private ByteBuffer writeBuffer; private boolean isConnected; public TcpClientChannelSelector() throws IOException { super(); socketChannel = SocketChannel.open(); socketChannel.configureBlocking(false); socketChannel.connect(new InetSocketAddress(HOSTNAME, PORT)); readBuffer = ByteBuffer.allocate(BUFF_SIZE); writeBuffer = ByteBuffer.allocate(BUFF_SIZE); isConnected = false; registerChannel(socketChannel, SelectionKey.OP_CONNECT | SelectionKey.OP_READ); } public boolean isConnected() { return isConnected; } @Override public void doOnSelectionKey(SelectionKey key) { if (key.isAcceptable()) { System.out.println("client is accepted by server"); } else if (key.isConnectable()) { SocketChannel channel = (SocketChannel) key.channel(); if (channel.isConnectionPending()) { try { /* * 人为将程序中止，等待连接创建完成,否则会报如下错误 * java.nio.channels.NotYetConnectedException at sun.nio.ch.SocketChannelImpl.ensureWriteOpen(SocketChannelImpl.java:274) */ if (channel.finishConnect()) { System.out.println("client connect server succ"); isConnected = true; } } catch (IOException e) { e.printStackTrace(); } } } else if (key.isReadable()) { try { int len = socketChannel.read(readBuffer); if (len == -1) { System.out.println("client read : len=-1"); // 说明连接已经断开 close(); } else { readBuffer.flip(); byte[] buffer = new byte[len]; readBuffer.get(buffer); readBuffer.clear(); System.out.println("client read : len=" + len + ", str=" + new String(buffer)); } } catch (IOException e) { e.printStackTrace(); } } } public void write(String str) { byte[] buffer = str.getBytes(); writeBuffer.put(buffer); writeBuffer.flip(); try { System.out.println("client write : len=" + buffer.length + ", str=" + str); socketChannel.write(writeBuffer); } catch (IOException e) { e.printStackTrace(); } writeBuffer.clear(); } public void close() { System.out.println("close client"); isConnected = false; super.closeSelector(); try { socketChannel.close(); } catch (IOException e) { e.printStackTrace(); } } }

TcpServerChannelSelector.java

package com.demo.channel; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.ArrayList; public class TcpServerChannelSelector extends BaseSocketChannelSelector { private static final String HOSTNAME = "127.0.0.1"; private static final int PORT = 5555; private static final int BUFF_SIZE = 255; private ServerSocketChannel serverSocketChannel; private ArrayList<SocketChannel> socketChannelList; private ByteBuffer readBuffer; private ByteBuffer writeBuffer; public TcpServerChannelSelector() throws IOException { super(); serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.configureBlocking(false); serverSocketChannel.bind(new InetSocketAddress(HOSTNAME, PORT)); socketChannelList = new ArrayList<SocketChannel>(); readBuffer = ByteBuffer.allocate(BUFF_SIZE); writeBuffer = ByteBuffer.allocate(BUFF_SIZE); /* * 因此这里只能先accept，然后再read，否则会报如下错误 * Exception in thread "main" java.lang.IllegalArgumentException at java.nio.channels.spi.AbstractSelectableChannel.register(AbstractSelectableChannel.java:199) */ registerChannel(serverSocketChannel, SelectionKey.OP_ACCEPT); } @Override protected void doOnSelectionKey(SelectionKey key) { if (key.isAcceptable()) { try { SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept(); socketChannelList.add(socketChannel); System.out.println("server find client : " + socketChannel.getRemoteAddress()); registerChannel(socketChannel, SelectionKey.OP_READ); } catch (IOException e) { e.printStackTrace(); } } else if (key.isReadable()) { try { SocketChannel socketChannel = (SocketChannel) key.channel(); int len = socketChannel.read(readBuffer); if (len == -1) { System.out.println("client read : len=-1"); // 说明连接已经断开 socketChannel.close(); socketChannelList.remove(socketChannel); } else { readBuffer.flip(); byte[] buffer = new byte[len]; readBuffer.get(buffer, 0, len); readBuffer.clear(); String str = new String(buffer); System.out.println("server read : len=" + len + ", str=" + str); write(str); } } catch (IOException e) { e.printStackTrace(); } } } public void write(String str) { byte[] buffer = str.getBytes(); writeBuffer.put(buffer); writeBuffer.flip(); for (SocketChannel channel : socketChannelList) { try { System.out.println("server write : len=" + buffer.length + ", str=" + str); channel.write(writeBuffer); } catch (IOException e) { e.printStackTrace(); } writeBuffer.rewind(); } writeBuffer.clear(); } public void close() { System.out.println("close server"); super.closeSelector(); for (SocketChannel channel : socketChannelList) { try { channel.close(); } catch (IOException e) { e.printStackTrace(); } } socketChannelList.clear(); try { serverSocketChannel.close(); } catch (IOException e) { e.printStackTrace(); } } }

SelectDemo.java

package com.demo.channel; import java.io.IOException; import java.util.Date; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class SelectDemo { private static TcpServerChannelSelector serverChannelSelector; private static TcpClientChannelSelector clientChannelSelector; public static void main(String[] args) { int nThreads = 2; try { serverChannelSelector = new TcpServerChannelSelector(); clientChannelSelector = new TcpClientChannelSelector(); ExecutorService executorService = Executors.newFixedThreadPool(nThreads); executorService.execute(new Runnable() { @Override public void run() { serverChannelSelector.select(); } }); executorService.execute(new Runnable() { @Override public void run() { clientChannelSelector.select(); } }); while (!clientChannelSelector.isConnected()) { try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } for (int i = 0; i < 5; i++) { String str = "(" + i + ")" + new Date(); clientChannelSelector.write(str); } try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } serverChannelSelector.close(); clientChannelSelector.close(); } catch (IOException e) { e.printStackTrace(); } } }

执行结果

client connect server succ server find client : /127.0.0.1:59651 client write : len=31, str=(0)Mon Dec 05 19:12:46 CST 2016 client write : len=31, str=(1)Mon Dec 05 19:12:46 CST 2016 server read : len=31, str=(0)Mon Dec 05 19:12:46 CST 2016 client write : len=31, str=(2)Mon Dec 05 19:12:46 CST 2016 server write : len=31, str=(0)Mon Dec 05 19:12:46 CST 2016 client write : len=31, str=(3)Mon Dec 05 19:12:46 CST 2016 server read : len=62, str=(1)Mon Dec 05 19:12:46 CST 2016(2)Mon Dec 05 19:12:46 CST 2016 server write : len=62, str=(1)Mon Dec 05 19:12:46 CST 2016(2)Mon Dec 05 19:12:46 CST 2016 client write : len=31, str=(4)Mon Dec 05 19:12:46 CST 2016 server read : len=31, str=(3)Mon Dec 05 19:12:46 CST 2016 server write : len=31, str=(3)Mon Dec 05 19:12:46 CST 2016 server read : len=31, str=(4)Mon Dec 05 19:12:46 CST 2016 server write : len=31, str=(4)Mon Dec 05 19:12:46 CST 2016 client read : len=155, str=(0)Mon Dec 05 19:12:46 CST 2016(1)Mon Dec 05 19:12:46 CST 2016(2)Mon Dec 05 19:12:46 CST 2016(3)Mon Dec 05 19:12:46 CST 2016(4)Mon Dec 05 19:12:46 CST 2016 close server selector is closed close client Exception in thread "pool-1-thread-2" java.nio.channels.ClosedSelectorException at sun.nio.ch.SelectorImpl.lockAndDoSelect(SelectorImpl.java:83) at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:97) at sun.nio.ch.SelectorImpl.select(SelectorImpl.java:101) at com.demo.channel.BaseSocketChannelSelector.select(BaseSocketChannelSelector.java:40) at com.demo.channel.SelectDemo$2.run(SelectDemo.java:29) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745)

参考

http://ifeve.com/selectors/http://blog.csdn.net/windsunmoon/article/details/45373457