开篇

回想研究生期间在H3C做项目的时候第一次接触epoll的异步事件，心血来潮看了下java的NIO的实现，希望同样感兴趣的人一起看看。Netty是java NIO的集大成者，一定要看看。

java NIO server demo

• 创建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);
                }
            }
        }
    }
}