Java进阶之高并发核心Selector详解

 // Selector.java public static Selector open() throws IOException { // 首先找到provider,然后再打开Selector return SelectorProvider.provider().openSelector(); } 

 // java.nio.channels.spi.SelectorProvider public static SelectorProvider provider() { synchronized (lock) { if (provider != null) return provider; return AccessController.doPrivileged( new PrivilegedAction() { public SelectorProvider run() { if (loadProviderFromProperty()) return provider; if (loadProviderAsService()) return provider; // 这里就是打开Selector的真正方法 provider = sun.nio.ch.DefaultSelectorProvider.create(); return provider; } }); } } 

 /** * Returns the default SelectorProvider. */ public static SelectorProvider cre<a style="color:transparent">来源gao($daima.com搞@代@#码网</a>ate() { // 获取OS名称 String osname = AccessController .doPrivileged(new GetPropertyAction("os.name")); // 根据名称来创建不同的Selctor if (osname.equals("SunOS")) return createProvider("sun.nio.ch.DevPollSelectorProvider"); if (osname.equals("Linux")) return createProvider("sun.nio.ch.EPollSelectorProvider"); return new sun.nio.ch.PollSelectorProvider(); } 

 public AbstractSelector openSelector() throws IOException { return new EPollSelectorImpl(this); }

 EPollSelectorImpl(SelectorProvider sp) throws IOException { super(sp); // makePipe返回管道的2个文件描述符，编码在一个long类型的变量中 // 高32位代表读 低32位代表写 // 使用pipe为了实现Selector的wakeup逻辑 long pipeFds = IOUtil.makePipe(false); fd0 = (int) (pipeFds >>> 32); fd1 = (int) pipeFds; // 新建一个EPollArrayWrapper pollWrapper = new EPollArrayWrapper(); pollWrapper.initInterrupt(fd0, fd1); fdToKey = new HashMap(); } 

 EPollArrayWrapper() throws IOException { // creates the epoll file descriptor // 创建epoll fd epfd = epollCreate(); // the epoll_event array passed to epoll_wait int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT; pollArray = new AllocatedNativeObject(allocationSize, true); pollArrayAddress = pollArray.address(); // eventHigh needed when using file descriptors > 64k if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE) eventsHigh = new HashMap(); } private native int epollCreate(); 

 EPollArrayWrapper() throws IOException { // creates the epoll file descriptor // 创建epoll fd epfd = epollCreate(); // the epoll_event array passed to epoll_wait int allocationSize = NUM_EPOLLEVENTS * SIZE_EPOLLEVENT; pollArray = new AllocatedNativeObject(allocationSize, true); pollArrayAddress = pollArray.address(); // eventHigh needed when using file descriptors > 64k if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE) eventsHigh = new HashMap(); } private native int epollCreate(); 

 protected int doSelect(long timeout) throws IOException { if (closed) throw new ClosedSelectorException(); processDeregisterQueue(); try { begin(); // 真正的实现是这行 pollWrapper.poll(timeout); } finally { end(); } processDeregisterQueue(); int numKeysUpdated = updateSelectedKeys(); // 以下基本都是异常处理 if (pollWrapper.interrupted()) { // Clear the wakeup pipe pollWrapper.putEventOps(pollWrapper.interruptedIndex(), 0); synchronized (interruptLock) { pollWrapper.clearInterrupted(); IOUtil.drain(fd0); interruptTriggered = false; } } return numKeysUpdated; } 

 int poll(long timeout) throws IOException { updateRegistrations(); // 这个epollWait是不是有点熟悉呢？ updated = epollWait(pollArrayAddress, NUM_EPOLLEVENTS, timeout, epfd); for (int i=0; i<updated; i++) { if (getDescriptor(i) == incomingInterruptFD) { interruptedIndex = i; interrupted = true; break; } } return updated; } private native int epollWait(long pollAddress, int numfds, long timeout, int epfd) throws IOException; 

 JNIEXPORT jint JNICALL Java_sun_nio_ch_EPollArrayWrapper_epollWait(JNIEnv *env, jobject this, jlong address, jint numfds, jlong timeout, jint epfd) { struct epoll_event *events = jlong_to_ptr(address); int res; if (timeout <= 0) {           /* Indefinite or no wait */ // 发起epoll_wait系统调用等待内核事件 RESTARTABLE(epoll_wait(epfd, events, numfds, timeout), res); } else {                      /* Bounded wait; bounded restarts */ res = iepoll(epfd, events, numfds, timeout); } if (res <0) { JNU_ThrowIOExceptionWithLastError(env, "epoll_wait failed"); } return res; } =

 // events for file descriptors with registration changes pending, indexed // by file descriptor and stored as bytes for efficiency reasons. For // file descriptors higher than MAX_UPDATE_ARRAY_SIZE (unlimited case at // least) then the update is stored in a map. // 使用数组保存事件变更, 数组的最大长度是MAX_UPDATE_ARRAY_SIZE, 最大64*1024 private final byte[] eventsLow = new byte[MAX_UPDATE_ARRAY_SIZE]; // 超过数组长度的事件会缓存到这个map中，等待下次处理 private Map eventsHigh; /** * Sets the pending update events for the given file descriptor. This * method has no effect if the update events is already set to KILLED, * unless {@code force} is {@code true}. */ private void setUpdateEvents(int fd, byte events, boolean force) { // 判断fd和数组长度 if (fd <MAX_UPDATE_ARRAY_SIZE) { if ((eventsLow[fd] != KILLED) || force) { eventsLow[fd] = events; } } else { Integer key = Integer.valueOf(fd); if (!isEventsHighKilled(key) || force) { eventsHigh.put(key, Byte.valueOf(events)); } } } 

 /** * Returns the pending update events for the given file descriptor. */ private byte getUpdateEvents(int fd) { if (fd <MAX_UPDATE_ARRAY_SIZE) { return eventsLow[fd]; } else { Byte result = eventsHigh.get(Integer.valueOf(fd)); // result should never be null return result.byteValue(); } } /** * Update the pending registrations. */ private void updateRegistrations() { synchronized (updateLock) { int j = 0; while (j <updateCount) { int fd = updateDescriptors[j]; // 从保存的eventsLow和eventsHigh里取出事件 short events = getUpdateEvents(fd); boolean isRegistered = registered.get(fd); int opcode = 0; if (events != KILLED) { // 判断操作类型以传给epoll_ctl // 没有指定EPOLLET事件类型 if (isRegistered) { opcode = (events != 0) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL; } else { opcode = (events != 0) ? EPOLL_CTL_ADD : 0; } if (opcode != 0) { // 熟悉的epoll_ctl epollCtl(epfd, opcode, fd, events); if (opcode == EPOLL_CTL_ADD) { registered.set(fd); } else if (opcode == EPOLL_CTL_DEL) { registered.clear(fd); } } } j++; } updateCount = 0; } } private native void epollCtl(int epfd, int opcode, int fd, int events); 

 JNIEXPORT void JNICALL Java_sun_nio_ch_EPollArrayWrapper_epollCtl(JNIEnv *env, jobject this, jint epfd, jint opcode, jint fd, jint events) { struct epoll_event event; int res; event.events = events; event.data.fd = fd; // 发起epoll_ctl调用来进行IO事件的管理 RESTARTABLE(epoll_ctl(epfd, (int)opcode, (int)fd, &event), res); /* * A channel may be registered with several Selectors. When each Selector * is polled a EPOLL_CTL_DEL op will be inserted into its pending update * list to remove the file descriptor from epoll. The "last" Selector will * close the file descriptor which automatically unregisters it from each * epoll descriptor. To avoid costly synchronization between Selectors we * allow pending updates to be processed, ignoring errors. The errors are * harmless as the last update for the file descriptor is guaranteed to * be EPOLL_CTL_DEL. */ if (res <0 && errno !=EBADF enoent eperm) { jnu_throwioexceptionwithlasterror(env, "epoll_ctl failed"); } <pre></div><p>原来还是我们的老朋友epoll_ctl.<br /> 有个小细节是<code>jdk没有指定ET(边缘触发)还是LT(水平触发),所以默认会用LT:)</code></p><p>在<code>AbstractSelectorImpl中有3个set保存事件</code></p><div class="gaodaimacode"><pre class="prettyprint linenums"> // Public views of the key sets // 注册的所有事件 private Set publicKeys;             // Immutable // 内核返回的IO事件封装，表示哪些fd有数据可读可写 private Set publicSelectedKeys;     // Removal allowed, but not addition // 取消的事件 private final Set cancelledKeys = new HashSet(); 

 private int updateSelectedKeys() { int entries = pollWrapper.updated; int numKeysUpdated = 0; for (int i=0; i<entries; i++) { int nextFD = pollWrapper.getDescriptor(i); SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD)); // ski is null in the case of an interrupt if (ski != null) { int rOps = pollWrapper.getEventOps(i); if (selectedKeys.contains(ski)) { if (ski.channel.translateAndSetReadyOps(rOps, ski)) { numKeysUpdated++; } } else { ski.channel.translateAndSetReadyOps(rOps, ski); if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) { selectedKeys.add(ski); numKeysUpdated++; } } } } return numKeysUpdated; 

 public class NIOServer { public static void main(String[] args) throws Exception{ //创建ServerSocketChannel -> ServerSocket ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); //得到一个Selecor对象 Selector selector = Selector.open(); //绑定一个端口6666, 在服务器端监听 serverSocketChannel.socket().bind(new InetSocketAddress(6666)); //设置为非阻塞 serverSocketChannel.configureBlocking(false); //把 serverSocketChannel 注册到  selector 关心 事件为 OP_ACCEPT serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); System.out.println("注册后的selectionkey 数量=" + selector.keys().size()); // 1 //循环等待客户端连接 while (true) { //这里我们等待1秒，如果没有事件发生, 返回 if(selector.select(1000) == 0) { //没有事件发生 System.out.println("服务器等待了1秒，无连接"); continue; } //如果返回的>0, 就获取到相关的 selectionKey集合 //1.如果返回的>0， 表示已经获取到关注的事件 //2. selector.selectedKeys() 返回关注事件的集合 //   通过 selectionKeys 反向获取通道 Set selectionKeys = selector.selectedKeys(); System.out.println("selectionKeys 数量 = " + selectionKeys.size()); //遍历 Set, 使用迭代器遍历 Iterator keyIterator = selectionKeys.iterator(); while (keyIterator.hasNext()) { //获取到SelectionKey SelectionKey key = keyIterator.next(); //根据key 对应的通道发生的事件做相应处理 if(key.isAcceptable()) { //如果是 OP_ACCEPT, 有新的客户端连接 //该该客户端生成一个 SocketChannel SocketChannel socketChannel = serverSocketChannel.accept(); System.out.println("客户端连接成功 生成了一个 socketChannel " + socketChannel.hashCode()); //将  SocketChannel 设置为非阻塞 socketChannel.configureBlocking(false); //将socketChannel 注册到selector, 关注事件为 OP_READ， 同时给socketChannel //关联一个Buffer socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024)); System.out.println("客户端连接后 ，注册的selectionkey 数量=" + selector.keys().size()); //2,3,4.. } if(key.isReadable()) {  //发生 OP_READ //通过key 反向获取到对应channel SocketChannel channel = (SocketChannel)key.channel(); //获取到该channel关联的buffer ByteBuffer buffer = (ByteBuffer)key.attachment(); channel.read(buffer); System.out.println("form 客户端 " + new String(buffer.array())); } //手动从集合中移动当前的selectionKey, 防止重复操作 keyIterator.remove(); } } } } 

 public class NIOClient { public static void main(String[] args) throws Exception{ //得到一个网络通道 SocketChannel socketChannel = SocketChannel.open(); //设置非阻塞 socketChannel.configureBlocking(false); //提供服务器端的ip 和 端口 InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 6666); //连接服务器 if (!socketChannel.connect(inetSocketAddress)) { while (!socketChannel.finishConnect()) { System.out.println("因为连接需要时间，客户端不会阻塞，可以做其它工作.."); } } //...如果连接成功，就发送数据 String str = "hello, 尚硅谷~"; //Wraps a byte array into a buffer ByteBuffer buffer = ByteBuffer.wrap(str.getBytes()); //发送数据，将 buffer 数据写入 channel socketChannel.write(buffer); System.in.read(); } }