java NIO 运行原理介绍
开篇 回想研究生期间在H3C做项目的时候第一次接触epoll的异步事件，心血来潮看了下java的NIO的实现，希望同样感兴趣的人一起看看。Netty是java NIO的集大成者，一定要看看。 java NIO server demo socket server端工作标准流程 创建socket： 创建ServerSocketChannel，通过ServerSocketChannel.open()方法。 绑定socket：ServerSocketChannel绑定端口，通过serverSocketChannel.bind()方法。 前置准备： 创建selector对象，通过Selector.open()方法。 前置准备： 注册Channel到selector并绑定事件，通过serverSocketChannel.register()。 监听端口号： 通过listen()方法开始进入监听。 处理事件： while循环中等待select操作返回区分连接还是数据进行不同处理。 public class NIOServer { private Selector selector; public void initServer(int port) throws IOException { // 获得一个ServerSocketChannel通道 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); // 设置通道为非阻塞 serverSocketChannel.configureBlocking(false); // 将该通道对应的ServerSocket绑定到port端口 serverSocketChannel.bind(new InetSocketAddress(port)); // 获得一个通道管理器 this.selector = Selector.open(); // 将通道管理器和该通道绑定，并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后， // 当该事件到达时，selector.select()会返回，如果该事件没到达selector.select()会一直阻塞。 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); } public void listen() throws IOException { System.out.println("服务端启动成功！"); // 轮询访问selector while (true) { // 当注册的事件到达时，方法返回；否则,该方法会一直阻塞 selector.select(); // 获得selector中选中的项的迭代器，选中的项为注册的事件 Iterator<SelectionKey> ite = this.selector.selectedKeys().iterator(); while (ite.hasNext()) { SelectionKey key = (SelectionKey) ite.next(); // 删除已选的key,以防重复处理 ite.remove(); if (key.isAcceptable()) {// 客户端请求连接事件 ServerSocketChannel server = (ServerSocketChannel) key.channel(); // 获得和客户端连接的通道 SocketChannel channel = server.accept(); // 设置成非阻塞 channel.configureBlocking(false); // 在这里可以给客户端发送信息哦 channel.write(ByteBuffer.wrap(new String("向客户端发送了一条信息") .getBytes("utf-8"))); // 在和客户端连接成功之后，为了可以接收到客户端的信息，需要给通道设置读的权限。 channel.register(this.selector, SelectionKey.OP_READ); } else if (key.isReadable()) {// 获得了可读的事件 read(key); } } } } public void read(SelectionKey key) throws IOException { // 服务器可读取消息:得到事件发生的Socket通道 SocketChannel channel = (SocketChannel) key.channel(); // 创建读取的缓冲区 ByteBuffer buffer = ByteBuffer.allocate(512); channel.read(buffer); byte[] data = buffer.array(); String msg = new String(data).trim(); System.out.println("服务端收到信息：" + msg); ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes("utf-8")); channel.write(outBuffer);// 将消息回送给客户端 } public static void main(String[] args) throws IOException { NIOServer server = new NIOServer(); server.initServer(8000); server.listen(); } } ServerSocketChannel和Selector初始化过程 在java NIO Server的标准过程中，有两个核心的操作需要深入分析一下，分别是ServerSocketChannel.open() 和 Selector.open()两个过程，这里针对这两个对象的初始化流程进行下细致的分解。 // ServerSocketChannel的初始化过程 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); // Selector的初始化过程 selector = Selector.open(); 通用逻辑抽取 ServerSocketChannel.open()=SelectorProvider.provider().openServerSocketChannel() Selector.open()=SelectorProvider.provider().openSelector() 两者有共同点在于都调用了SelectorProvider.provider()方法，所以先把相同部分进行分析。 public abstract class ServerSocketChannel extends AbstractSelectableChannel implements NetworkChannel { protected ServerSocketChannel(SelectorProvider provider) { super(provider); } public static ServerSocketChannel open() throws IOException { return SelectorProvider.provider().openServerSocketChannel(); } } public abstract class Selector implements Closeable { protected Selector() { } public static Selector open() throws IOException { return SelectorProvider.provider().openSelector(); } } SelectorProvider对象创建 SelectorProvider.provider()方法会在内部创建唯一的SelectorProvider对象，通过锁来保证创建唯一对象。 SelectorProvider对象通过DefaultSelectorProvider.create()方法进行创建。 DefaultSelectorProvider.create()方法内部根据实际系统创建不同的对象，以linux环境中EPollSelectorProvider对象为例继续分析。 public abstract class SelectorProvider { private static final Object lock = new Object(); private static SelectorProvider provider = null; public static SelectorProvider provider() { synchronized (lock) { if (provider != null) return provider; return AccessController.doPrivileged( new PrivilegedAction<SelectorProvider>() { public SelectorProvider run() { if (loadProviderFromProperty()) return provider; if (loadProviderAsService()) return provider; provider = sun.nio.ch.DefaultSelectorProvider.create(); return provider; } }); } } } public class DefaultSelectorProvider { public static SelectorProvider create() { String osname = AccessController.doPrivileged( new GetPropertyAction("os.name")); if ("SunOS".equals(osname)) { return new sun.nio.ch.DevPollSelectorProvider(); } // use EPollSelectorProvider for Linux kernels >= 2.6 if ("Linux".equals(osname)) { String osversion = AccessController.doPrivileged( new GetPropertyAction("os.version")); String[] vers = osversion.split("\\.", 0); if (vers.length >= 2) { try { int major = Integer.parseInt(vers[0]); int minor = Integer.parseInt(vers[1]); if (major > 2 || (major == 2 && minor >= 6)) { return new sun.nio.ch.EPollSelectorProvider(); } } catch (NumberFormatException x) { // format not recognized } } } return new sun.nio.ch.PollSelectorProvider(); } } EPollSelectorProvider的操作过程 SelectorProvider.provider().openServerSocketChannel()调用EPollSelectorProvider的openServerSocketChannel()方法返回EPollSelectorImpl对象。 SelectorProvider.provider().openSelector()调用EPollSelectorProvider的openSelector()方法返回ServerSocketChannelImpl对象。 继续分析ServerSocketChannelImpl对象和EPollSelectorImpl对象。 public class EPollSelectorProvider extends SelectorProviderImpl { public AbstractSelector openSelector() throws IOException { return new EPollSelectorImpl(this); } public Channel inheritedChannel() throws IOException { return InheritedChannel.getChannel(); } } public abstract class SelectorProviderImpl extends SelectorProvider { public DatagramChannel openDatagramChannel() throws IOException { return new DatagramChannelImpl(this); } public DatagramChannel openDatagramChannel(ProtocolFamily family) throws IOException { return new DatagramChannelImpl(this, family); } public Pipe openPipe() throws IOException { return new PipeImpl(this); } public abstract AbstractSelector openSelector() throws IOException; public ServerSocketChannel openServerSocketChannel() throws IOException { return new ServerSocketChannelImpl(this); } public SocketChannel openSocketChannel() throws IOException { return new SocketChannelImpl(this); } } EPollSelectorImpl对象 EPollSelectorImpl构造函数创建内部通信的socket对IOUtil.makePipe(false)。 EPollSelectorImpl的doSelect方法负责返回事件到来的fds。 EPollSelectorImpl的fdToKey的map保存fd和SelectionKey的映射。 class EPollSelectorImpl extends SelectorImpl { // File descriptors used for interrupt protected int fd0; protected int fd1; // The poll object EPollArrayWrapper pollWrapper; // Maps from file descriptors to keys private Map<Integer,SelectionKeyImpl> fdToKey; // True if this Selector has been closed private volatile boolean closed = false; // Lock for interrupt triggering and clearing private Object interruptLock = new Object(); private boolean interruptTriggered = false; EPollSelectorImpl(SelectorProvider sp) { super(sp); long pipeFds = IOUtil.makePipe(false); fd0 = (int) (pipeFds >>> 32); fd1 = (int) pipeFds; pollWrapper = new EPollArrayWrapper(); pollWrapper.initInterrupt(fd0, fd1); fdToKey = new HashMap<Integer,SelectionKeyImpl>(); } protected int doSelect(long timeout) throws IOException { if (closed) throw new ClosedSelectorException(); processDeregisterQueue(); try { begin(); // 等待事件到来，收集事件到来的socket的fd并用来处理 pollWrapper.poll(timeout); } finally { end(); } processDeregisterQueue(); // 更新需要写入的keys 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; } 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保存ski也就是事件到的socket连接 // ski的对象数据结构需要好好研究一下 selectedKeys.add(ski); numKeysUpdated++; } } } } return numKeysUpdated; } } ServerSocketChannelImpl对象 ServerSocketChannelImpl extends ServerSocketChannel ServerSocketChannel extends AbstractSelectableChannel ServerSocketChannelImpl对象提供bind()&accept()方法 ServerSocketChannelImpl的accept方法内部创建新连接的SocketChannelImpl对象返回 class ServerSocketChannelImpl extends ServerSocketChannel implements SelChImpl { private final Object stateLock = new Object(); private SocketAddress localAddress; ServerSocket socket; ServerSocketChannelImpl(SelectorProvider sp) throws IOException { super(sp); this.fd = Net.serverSocket(true); this.fdVal = IOUtil.fdVal(fd); this.state = ST_INUSE; } ServerSocketChannelImpl(SelectorProvider sp, FileDescriptor fd, boolean bound) throws IOException { super(sp); this.fd = fd; this.fdVal = IOUtil.fdVal(fd); this.state = ST_INUSE; if (bound) localAddress = Net.localAddress(fd); } @Override public ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException { // 省略相关代码 } public SocketChannel accept() throws IOException { // 省略相关代码 } } public abstract class AbstractSelectableChannel extends SelectableChannel { protected AbstractSelectableChannel(SelectorProvider provider) { this.provider = provider; } public final SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException { synchronized (regLock) { if (!isOpen()) throw new ClosedChannelException(); if ((ops & ~validOps()) != 0) throw new IllegalArgumentException(); if (blocking) throw new IllegalBlockingModeException(); SelectionKey k = findKey(sel); if (k != null) { k.interestOps(ops); k.attach(att); } if (k == null) { // New registration synchronized (keyLock) { if (!isOpen()) throw new ClosedChannelException(); k = ((AbstractSelector)sel).register(this, ops, att); addKey(k); } } return k; } } } select过程 select执行过程 执行selector.select()操作时实际是调用了子类实现的doSelect()方法。 进一步跟进子类的doSelect()方法。 abstract class SelectorImpl extends AbstractSelector { // 保存事件到来的keys protected Set<SelectionKey> selectedKeys; protected HashSet<SelectionKey> keys; private Set<SelectionKey> publicKeys; // Immutable private Set<SelectionKey> publicSelectedKeys; // Removal allowed, but not addition protected abstract int doSelect(long timeout) throws IOException; private int lockAndDoSelect(long timeout) throws IOException { synchronized (this) { if (!isOpen()) throw new ClosedSelectorException(); synchronized (publicKeys) { synchronized (publicSelectedKeys) { return doSelect(timeout); } } } } public int select(long timeout) throws IOException { if (timeout < 0) throw new IllegalArgumentException("Negative timeout"); return lockAndDoSelect((timeout == 0) ? -1 : timeout); } public int select() throws IOException { return select(0); } public Set<SelectionKey> selectedKeys() { if (!isOpen() && !Util.atBugLevel("1.4")) throw new ClosedSelectorException(); return publicSelectedKeys; } } pollWrapper.poll(timeout)以超时等待的形式等待epoll的消息通知。 通过updateSelectedKeys方法收集有事件到达的fds保存到selectedKeys。 class EPollSelectorImpl extends SelectorImpl { protected int doSelect(long timeout) throws IOException { if (closed) throw new ClosedSelectorException(); processDeregisterQueue(); try { begin(); // 等待事件到来，收集事件到来的socket的fd并用来处理 pollWrapper.poll(timeout); } finally { end(); } processDeregisterQueue(); // 更新需要写入的keys 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; } 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保存ski也就是事件到的socket连接 // ski的对象数据结构需要好好研究一下 selectedKeys.add(ski); numKeysUpdated++; } } } } return numKeysUpdated; } } accept过程 accept的过程很简单就是accept新socket并创建SocketChannelImpl返回即可。 SocketChannelImpl对象后面需要注册到Selector当中所以需要进一步分析。 public SocketChannel accept() throws IOException { // 省略相关代码 try { // 省略相关代码 // 新accept的socket放在newfd当中 n = accept0(this.fd, newfd, isaa); } } IOUtil.configureBlocking(newfd, true); InetSocketAddress isa = isaa[0]; // 通过SocketChannelImpl包装newfd对象 sc = new SocketChannelImpl(provider(), newfd, isa); // 省略相关代码 return sc; } } SocketChannelImpl对象 SocketChannelImpl可以理解为普通Socket的封装，包括read/write等方法 SocketChannelImpl extends SocketChannel extends AbstractSelectableChannel AbstractSelectableChannel提供register到selector对象的方法 class SocketChannelImpl extends SocketChannel implements SelChImpl { SocketChannelImpl(SelectorProvider sp) throws IOException { super(sp); this.fd = Net.socket(true); this.fdVal = IOUtil.fdVal(fd); this.state = ST_UNCONNECTED; } SocketChannelImpl(SelectorProvider sp, FileDescriptor fd, boolean bound) throws IOException { super(sp); this.fd = fd; this.fdVal = IOUtil.fdVal(fd); this.state = ST_UNCONNECTED; if (bound) this.localAddress = Net.localAddress(fd); } SocketChannelImpl(SelectorProvider sp, FileDescriptor fd, InetSocketAddress remote) throws IOException { super(sp); this.fd = fd; this.fdVal = IOUtil.fdVal(fd); this.state = ST_CONNECTED; this.localAddress = Net.localAddress(fd); this.remoteAddress = remote; } public long read(ByteBuffer[] dsts, int offset, int length) throws IOException { // 读数据的逻辑 } public int write(ByteBuffer buf) throws IOException { // 写数据的逻辑 } } register过程 register过程并没有调用epollCtl方法添加fd到selector当中 register过程真正是保存fd到待绑定的列表当中 在SelectorImpl中执行pollWrapper.poll(timeout)方法先把fd列表执行epollCtl添加selector当中，在通过epollWait获取事件到来 public abstract class AbstractSelectableChannel extends SelectableChannel { public final SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException { synchronized (regLock) { if (!isOpen()) throw new ClosedChannelException(); if ((ops & ~validOps()) != 0) throw new IllegalArgumentException(); if (blocking) throw new IllegalBlockingModeException(); SelectionKey k = findKey(sel); if (k != null) { k.interestOps(ops); k.attach(att); } if (k == null) { // New registration synchronized (keyLock) { if (!isOpen()) throw new ClosedChannelException(); k = ((AbstractSelector)sel).register(this, ops, att); addKey(k); } } return k; } } } abstract class SelectorImpl extends AbstractSelector { protected final SelectionKey register(AbstractSelectableChannel ch, int ops, Object attachment) { if (!(ch instanceof SelChImpl)) throw new IllegalSelectorException(); SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this); k.attach(attachment); synchronized (publicKeys) { implRegister(k); } k.interestOps(ops); return k; } } abstract class AbstractPollSelectorImpl extends SelectorImpl { protected void implRegister(SelectionKeyImpl ski) { synchronized (closeLock) { if (closed) throw new ClosedSelectorException(); // Check to see if the array is large enough if (channelArray.length == totalChannels) { // Make a larger array int newSize = pollWrapper.totalChannels * 2; SelectionKeyImpl temp[] = new SelectionKeyImpl[newSize]; // Copy over for (int i=channelOffset; i<totalChannels; i++) temp[i] = channelArray[i]; channelArray = temp; // Grow the NativeObject poll array pollWrapper.grow(newSize); } channelArray[totalChannels] = ski; ski.setIndex(totalChannels); // 核心的将channel添加到pollWrapper当中 pollWrapper.addEntry(ski.channel); totalChannels++; keys.add(ski); } } } class EPollArrayWrapper { int poll(long timeout) throws IOException { updateRegistrations(); 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; } void updateRegistrations() { synchronized (updateList) { Updator u = null; while ((u = updateList.poll()) != null) { SelChImpl ch = u.channel; if (!ch.isOpen()) continue; // if the events are 0 then file descriptor is put into "idle // set" to prevent it being polled if (u.events == 0) { boolean added = idleSet.add(u.channel); // if added to idle set then remove from epoll if registered if (added && (u.opcode == EPOLL_CTL_MOD)) epollCtl(epfd, EPOLL_CTL_DEL, ch.getFDVal(), 0); } else { // events are specified. If file descriptor was in idle set // it must be re-registered (by converting opcode to ADD) boolean idle = false; if (!idleSet.isEmpty()) idle = idleSet.remove(u.channel); int opcode = (idle) ? EPOLL_CTL_ADD : u.opcode; epollCtl(epfd, opcode, ch.getFDVal(), u.events); } } } } }
