探索NIO之selector的底层实现

NIO — Not Block IO。在java中大体有三种角色。

  • Channel

  • Buffer

  • Selector

Channel故名思义就是一个渠道,一条链接,实际物理意义映射着我们平时说的tcp链接。

在java的tcp服务端中,主要有两种Channel,ServerSocketChannel和 SocketChannel,工作模式都是全双工的。ServerSocketChannel是用于监听服务端端口的,假设我们的程序监听了8080端口,那么ServerSocketChannel就扮演者监听者的角色,如果有客户端握手成功了,将第一时间被这个Channel发现到。发现到以后,ServerSocketChannel会调用accept()来创建一个小弟ServerSocketChannel,对这个客户端链接独立服务。从这以后,ServerSocketChannel就不在关注这个客户端链接了。和客户端之间的数据传输,均是交由该SocketChannel去干。

由此看来。在一个服务端程序中,ServerSocketChannel就像是一个总机的角色,有客户端连进来了,由总机首次发觉,然后就交给一个新的分机去处理具体的业务数据流动。

客户端程序相比于服务端,因其是靶向连接到固定地址,所以不需要ServerSocketChannel的角色。

Buffer,服务端socket和客户端socket通讯时,存储数据的缓冲区,这个不展开讲了。网络编程中,我们常用ByteBuffer。

一个服务端程序,如果仅仅使用Channel和Buffer,我们也许会这样做:
一个线程去检测ServerSocketChannel的状态,然后每accept()一个客户端之后,起一个新的线程带着新的SocketChannel去玩,循环检测这个Channel上是否有数据传进来,读入Buffer,处理,用Buffer塞入Channel。。。
这样我们需要开启1+n个线程。当然,我们也可以使用1个线程轮寻所有Channel,但是这种方法效率低下,不被使用。为了解决这个问题,jdk引入了NIO的第三大组件 — Selector。

  • Selector 复用选择器,可以register方法将所有需要事件监听的Channel都放入其中,监听的事件主要有两种,可读和可写。 可读又进一步区分成accept和read事件。
    我们只需要开启一个线程循环调用Selector的select方法,即可一个返回发生事件的Channel迭代器,使用上更加方便了。

下面我们借用netty的example代码我们展示下如何使用Selector创建一个服务端程序 。


/*
 * Copyright 2013-2018 Lilinfeng.
 *  
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.phei.netty.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketOption;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * @author Administrator
 * @date 2014年2月16日
 * @version 1.0
 */
public class MultiplexerTimeServer implements Runnable {

    private Selector selector;

    private ServerSocketChannel servChannel;

    private volatile boolean stop;

    /**
     * 初始化多路复用器、绑定监听端口
     * 
     * @param port
     */
    public MultiplexerTimeServer(int port) {
    try {
        selector = Selector.open();
        servChannel = ServerSocketChannel.open();
        servChannel.configureBlocking(false);
        servChannel.socket().bind(new InetSocketAddress(port), 1024);
        servChannel.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("The time server is start in port : " + port);
    } catch (IOException e) {
        e.printStackTrace();
        System.exit(1);
    }
    }

    public void stop() {
    this.stop = true;
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
    while (!stop) {
        try {
        selector.select(1000);
        Set<SelectionKey> selectedKeys = selector.selectedKeys();
        Iterator<SelectionKey> it = selectedKeys.iterator();
        SelectionKey key = null;
        while (it.hasNext()) {
            key = it.next();
            it.remove();
            try {
            handleInput(key);
            } catch (Exception e) {
            if (key != null) {
                key.cancel();
                if (key.channel() != null)
                key.channel().close();
            }
            }
        }
        } catch (Throwable t) {
        t.printStackTrace();
        }
    }

    // 多路复用器关闭后,所有注册在上面的Channel和Pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
    if (selector != null)
        try {
        selector.close();
        } catch (IOException e) {
        e.printStackTrace();
        }
    }

    private void handleInput(SelectionKey key) throws IOException {

    if (key.isValid()) {
        // 处理新接入的请求消息
        if (key.isAcceptable()) {
        // Accept the new connection
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        SocketChannel sc = ssc.accept();
        sc.configureBlocking(false);
        // Add the new connection to the selector
        sc.register(selector, SelectionKey.OP_READ);
        }
        if (key.isReadable()) {
        // Read the data
        SocketChannel sc = (SocketChannel) key.channel();
        //sc.setOption(SocketOption)
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);
        int readBytes = sc.read(readBuffer);
        if (readBytes > 0) {
            readBuffer.flip();
            byte[] bytes = new byte[readBuffer.remaining()];
            readBuffer.get(bytes);
            String body = new String(bytes, "UTF-8");
            System.out.println("The time server receive order : "
                + body);
            String currentTime = "QUERY TIME ORDER"
                .equalsIgnoreCase(body) ? new java.util.Date(
                System.currentTimeMillis()).toString()
                : "BAD ORDER";
            doWrite(sc, currentTime);
        } else if (readBytes < 0) {
            // 对端链路关闭
            key.cancel();
            sc.close();
        } else
            ; // 读到0字节,忽略
        }
    }
    }

    private void doWrite(SocketChannel channel, String response)
        throws IOException {
    if (response != null && response.trim().length() > 0) {
        byte[] bytes = response.getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
        writeBuffer.put(bytes);
        writeBuffer.flip();
        channel.write(writeBuffer);
    }
    }

    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 采用默认值
            }
        }
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);
        new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
    }
}

如果有c的编程经验的话,可以看出来,java中的nio在c中都是有对应原型的,selector对象对应着c中的nio句柄(不同平台下有着不同的实现),Channel对应着socket句柄,可以调用setoption设置socket参数。而Channel应该就是socket句柄调用read等方法读取出来的数据做了缓存。下面我们来验证一下我们的猜想。

selector = Selector.open();

这里打开了一个selector对象,查看实现。

    public static Selector open() throws IOException {
        return SelectorProvider.provider().openSelector();
    }

看下provider的实现。

    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;
                        }
                    });
        }
    }

重要的是这一句”provider = sun.nio.ch.DefaultSelectorProvider.create();”,这里下层的代码,各个平台的jdk版本有了差异,我这里是mac,生成的是一个KQueueSelectorProvider对象。一步步跟进,最后调用到了这里

    KQueueArrayWrapper() {
        int var1 = SIZEOF_KEVENT * 128;
        this.keventArray = new AllocatedNativeObject(var1, true);
        this.keventArrayAddress = this.keventArray.address();
        this.kq = this.init();
    }

其中init是一个native方法,这个方法的jni层命名应该是包名+对象名+函数名,拼凑出来是这样的。”sun.nio.ch”+ “KQueueArrayWrapper”+”init”,结果是”Java_sun_nio_ch_KQueueArrayWrapper_init”找到jvm的实现代码,位于jdk/src/macosx/native/sun/nio/ch/KQueueArrayWrapper.c 。

JNIEXPORT jint JNICALL
Java_sun_nio_ch_KQueueArrayWrapper_init(JNIEnv *env, jobject this)
{
    int kq = kqueue();
    if (kq < 0) {
        JNU_ThrowIOExceptionWithLastError(env, "KQueueArrayWrapper: kqueue() failed");
    }
    return kq;
}

确实是新建了一个kqueue句柄。由此可以得到结论,selector在macos下的底层实现是kqueue。

同时附上register和event的实现。

JNIEXPORT void JNICALL
Java_sun_nio_ch_KQueueArrayWrapper_register0(JNIEnv *env, jobject this,
                                             jint kq, jint fd, jint r, jint w)
{
    struct kevent changes[2];
    struct kevent errors[2];
    struct timespec dontBlock = {0, 0};

    // if (r) then { register for read } else { unregister for read }
    // if (w) then { register for write } else { unregister for write }
    // Ignore errors - they're probably complaints about deleting non-
    //   added filters - but provide an error array anyway because
    //   kqueue behaves erratically if some of its registrations fail.
    EV_SET(&changes[0], fd, EVFILT_READ,  r ? EV_ADD : EV_DELETE, 0, 0, 0);
    EV_SET(&changes[1], fd, EVFILT_WRITE, w ? EV_ADD : EV_DELETE, 0, 0, 0);
    kevent(kq, changes, 2, errors, 2, &dontBlock);
}


JNIEXPORT jint JNICALL
Java_sun_nio_ch_KQueueArrayWrapper_kevent0(JNIEnv *env, jobject this, jint kq,
                                           jlong kevAddr, jint kevCount,
                                           jlong timeout)
{
    struct kevent *kevs = (struct kevent *)jlong_to_ptr(kevAddr);
    struct timespec ts;
    struct timespec *tsp;
    int result;

    // Java timeout is in milliseconds. Convert to struct timespec.
    // Java timeout == -1 : wait forever : timespec timeout of NULL
    // Java timeout == 0  : return immediately : timespec timeout of zero
    if (timeout >= 0) {
        ts.tv_sec = timeout / 1000;
        ts.tv_nsec = (timeout % 1000) * 1000000; //nanosec = 1 million millisec
        tsp = &ts;
    } else {
        tsp = NULL;
    }

    result = kevent(kq, NULL, 0, kevs, kevCount, tsp);
 
    if (result < 0) {
        if (errno == EINTR) {
            // ignore EINTR, pretend nothing was selected
            result = 0;
        } else {
            JNU_ThrowIOExceptionWithLastError(env, "KQueueArrayWrapper: kqueue failed");
        }
    }

    return result;
}

那么我们可以假设程序运行到这一步到时候

servChannel.register(selector, SelectionKey.OP_ACCEPT);

应该会调用到底层的函数Java_sun_nio_ch_KQueueArrayWrapper_register0。
我们来看一下:

    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;
        }
    }

    protected final SelectionKey register(AbstractSelectableChannel var1, int var2, Object var3) {
        if (!(var1 instanceof SelChImpl)) {
            throw new IllegalSelectorException();
        } else {
            SelectionKeyImpl var4 = new SelectionKeyImpl((SelChImpl)var1, this);
            var4.attach(var3);
            Set var5 = this.publicKeys;
            synchronized(this.publicKeys) {
                this.implRegister(var4);
            }

            var4.interestOps(var2);
            return var4;
        }
    }

进一步看impRegister.

    protected void implRegister(SelectionKeyImpl var1) {
        if (this.closed) {
            throw new ClosedSelectorException();
        } else {
            int var2 = IOUtil.fdVal(var1.channel.getFD());
            this.fdMap.put(var2, new KQueueSelectorImpl.MapEntry(var1));
            ++this.totalChannels;
            this.keys.add(var1);
        }
    }

发现并没有,这里仅仅是将Channel封装成SelectionKeyImpl塞入一个HashSet了,那么证明我们刚才的推断是错误的,那NIO是如何将注册的Channel和底层的kqueue关联起来呢

我们在看这一步。

selector.select(1000);

查看jdk源码。

    private int lockAndDoSelect(long var1) throws IOException {
        synchronized(this) {
            if (!this.isOpen()) {
                throw new ClosedSelectorException();
            } else {
                Set var4 = this.publicKeys;
                int var10000;
                synchronized(this.publicKeys) {
                    Set var5 = this.publicSelectedKeys;
                    synchronized(this.publicSelectedKeys) {
                        var10000 = this.doSelect(var1);
                    }
                }

                return var10000;
            }
        }
    }

看看doSelect

    protected int doSelect(long var1) throws IOException {
        boolean var3 = false;
        if (this.closed) {
            throw new ClosedSelectorException();
        } else {
            this.processDeregisterQueue();

            int var7;
            try {
                this.begin();
                var7 = this.kqueueWrapper.poll(var1);
            } finally {
                this.end();
            }

            this.processDeregisterQueue();
            return this.updateSelectedKeys(var7);
        }
    }

查看poll

    int poll(long var1) {
        this.updateRegistrations();
        int var3 = this.kevent0(this.kq, this.keventArrayAddress, 128, var1);
        return var3;
    }

    void updateRegistrations() {
        LinkedList var1 = this.updateList;
        synchronized(this.updateList) {
            KQueueArrayWrapper.Update var2 = null;

            while((var2 = (KQueueArrayWrapper.Update)this.updateList.poll()) != null) {
                SelChImpl var3 = var2.channel;
                if (var3.isOpen()) {
                    this.register0(this.kq, var3.getFDVal(), var2.events & Net.POLLIN, var2.events & Net.POLLOUT);
                }
            }

        }
    }

结果显而易见了,在调用selector.select内部,会去检查是否有取消关联的channel,如果有,先把它删去,然后将新增的channel(存储在updateList链表变量中)注册到kqueue中。最后调用native函数event0来获取selector监听的所有Channel上是否有事件发生。

综上,Selector不单单是JDK内部批量管理了Channel的状态,更是借用了底层的异步IO实现来支持高并发的客户端连接场景。

    原文作者:msrpp
    原文地址: https://www.jianshu.com/p/2cccad527509
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞