【从入门到放弃-Java】并发编程-NIO-Selector

2021-07-20 05:23:31

前言

前两篇【从入门到放弃-Java】并发编程-NIO-Channel和【从入门到放弃-Java】并发编程-NIO-Buffer中我们学习了NIO中两个重要的概念Channel和Buffer。
今天我们来看下另一个重要的内容 Selector

简介

Selector是多路复用器，会不断轮询已经注册了的Channel。当有注册的channel产生连接、读、写等事件时，就会被Selector发现，从而可以进行相关后续操作。

Selector的好处是，可以通过一个线程来管理多个通道，减少了创建线程的资源占用及线程切换带来的消耗

Selector

SelectableChannel可以通过SelectionKey(记录channel和selector的注册关系)注册到Selector上。Selector维护了三个SelectionKey集合：

  • key set：存放了Selector上已经注册了的Channel的key。可以通过keys()方法获取。
  • selected-key set：当之前注册感兴趣的事件到达时，set中的keys会被更新或添加，set中维护了当前至少有一个可以操作的事件的channel key的集合。是key set的子集。可以使用selectedKeys()获取。
  • cancelled-key：存放已经调用cancel方法取消，等待下次操作时会调用deregister取消注册的channel，调用deregister后，所有的set中都没有这个channel的key了。

open

/**
 * Opens a selector.
 *
 * <p> The new selector is created by invoking the {@link
 * java.nio.channels.spi.SelectorProvider#openSelector openSelector} method
 * of the system-wide default {@link
 * java.nio.channels.spi.SelectorProvider} object.  </p>
 *
 * @return  A new selector
 *
 * @throws  IOException
 *          If an I/O error occurs
 */
public static Selector open() throws IOException {
    return SelectorProvider.provider().openSelector();
}

开启selector，具体的实现会根据操作系统类型创建不同的实现类，如macOS下实际上是new了一个KQueueSelectorProvider实例，低层基于操作系统的kqueue实现。

register

protected final SelectionKey register(AbstractSelectableChannel ch,
                                          int ops,
                                          Object attachment)
{
    if (!(ch instanceof SelChImpl))
        throw new IllegalSelectorException();
    //新建一个SelectionKey，记录channel与selector之间的注册关系
    SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);
    k.attach(attachment);

    //前置操作，这里主要是判断下selector是否还处于open状态
    // register (if needed) before adding to key set
    implRegister(k);

    // 添加selectionKey至key set
    // add to the selector's key set, removing it immediately if the selector
    // is closed. The key is not in the channel's key set at this point but
    // it may be observed by a thread iterating over the selector's key set.
    keys.add(k);
    try {
        // 更新注册的事件码
        k.interestOps(ops);
    } catch (ClosedSelectorException e) {
        assert ch.keyFor(this) == null;
        keys.remove(k);
        k.cancel();
        throw e;
    }
    return k;
}

注册selector和channel之间的事件关系。

select

// timeout超时
@Override
public final int select(long timeout) throws IOException {
    if (timeout < 0)
        throw new IllegalArgumentException("Negative timeout");
    return lockAndDoSelect(null, (timeout == 0) ? -1 : timeout);
}

@Override
public final int select() throws IOException {
    return lockAndDoSelect(null, -1);
}

// 不阻塞
@Override
public final int selectNow() throws IOException {
    return lockAndDoSelect(null, 0);
}

private int lockAndDoSelect(Consumer<SelectionKey> action, long timeout)
        throws IOException
{
    synchronized (this) {
        ensureOpen();
        if (inSelect)
            throw new IllegalStateException("select in progress");
        inSelect = true;
        try {
            synchronized (publicSelectedKeys) {
                return doSelect(action, timeout);
            }
        } finally {
            inSelect = false;
        }
    }
}

protected int doSelect(Consumer<SelectionKey> action, long timeout)
        throws IOException
{
    assert Thread.holdsLock(this);

    // 如果timeout = 0时，不阻塞
    long to = Math.min(timeout, Integer.MAX_VALUE);  // max kqueue timeout
    boolean blocking = (to != 0);
    boolean timedPoll = (to > 0);

    int numEntries;
    processUpdateQueue();
    processDeregisterQueue();
    try {
        // 设置interrupt 可以处理中断信号 防止线程一直阻塞
        begin(blocking);

        // 轮询的监听，直到有注册的事件发生或超时。
        do {
            long startTime = timedPoll ? System.nanoTime() : 0;
            numEntries = KQueue.poll(kqfd, pollArrayAddress, MAX_KEVENTS, to);
            if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
                // timed poll interrupted so need to adjust timeout
                long adjust = System.nanoTime() - startTime;
                to -= TimeUnit.MILLISECONDS.convert(adjust, TimeUnit.NANOSECONDS);
                if (to <= 0) {
                    // timeout expired so no retry
                    numEntries = 0;
                }
            }
        } while (numEntries == IOStatus.INTERRUPTED);
        assert IOStatus.check(numEntries);

    } finally {
        end(blocking);
    }
    processDeregisterQueue();
    return processEvents(numEntries, action);
}

selectedKeys

public final Set<SelectionKey> selectedKeys() {
    ensureOpen();
    return publicSelectedKeys;
}

获取被事件唤醒的key
注意：当被遍历处理selectedKeys时，key被处理完需要手动remove掉，防止下次被重复消费，selectedKeys不会帮你删除已处理过的key。

close

public final void close() throws IOException {
    boolean open = selectorOpen.getAndSet(false);
    if (!open)
        return;
    implCloseSelector();
}


public final void implCloseSelector() throws IOException {
    //通知处于阻塞的select方法立即返回
    wakeup();
    synchronized (this) {
        implClose();
        synchronized (publicSelectedKeys) {
            // 遍历所有的SelectionKey，取消注册
            // Deregister channels
            Iterator<SelectionKey> i = keys.iterator();
            while (i.hasNext()) {
                SelectionKeyImpl ski = (SelectionKeyImpl)i.next();
                deregister(ski);
                SelectableChannel selch = ski.channel();
                if (!selch.isOpen() && !selch.isRegistered())
                    ((SelChImpl)selch).kill();
                selectedKeys.remove(ski);
                i.remove();
            }
            assert selectedKeys.isEmpty() && keys.isEmpty();
        }
    }
}

SelectionKey

SelectionKey在channel register时创建。用来记录channel和selector之间的注册事件关系。
事件主要有：

  • OP_READ
  • OP_WRITE
  • OP_CONNECT
  • OP_ACCEPT

每个SelectionKey有两个由整数表示的操作集合，用来标识channel支持的操作类型。

interest set：是在创建SelectionKey时定义的，当集合中的操作发生时，将会把channel置为ready状态
ready set：检测到selector中已经就绪的操作类型集合

channel

public SelectableChannel channel() {
    return (SelectableChannel)channel;
}

获取SelectionKey中的channel

selector

public Selector selector() {
    return selector;
}

获取SelectionKey中的selector

isReadable

public final boolean isReadable() {
    return (readyOps() & OP_READ) != 0;
}

根据readyOps(readySet)判断channel是否是可读状态

isWritable

public final boolean isWritable() {
    return (readyOps() & OP_WRITE) != 0;
}

根据readyOps(readySet)判断channel是否是可写状态

isConnectable

public final boolean isConnectable() {
    return (readyOps() & OP_CONNECT) != 0;
}

根据readyOps(readySet)判断channel是否是connect状态，通常是客户端使用，判断连接是否建立

isReadable

public final boolean isAcceptable() {
    return (readyOps() & OP_ACCEPT) != 0;
}

根据readyOps(readySet)判断channel是否是accept状态，通常是服务端使用，判断是否有客户端请求建立连接

总结

通过使用selector，可以使用一个线程来管理多个连接。需要注意的一点是，通常读、写操作都是比较耗时的，为了提高服务端的性能应该把Selector::select和read、write的具体处理逻辑在不同的线程中处理。
即：使用一个线程来进行select，只做分发。在获取到就绪的SelectionKey后，通过线程池在不同的线程中处理读写操作。

通过学习完NIO相关的知识，我们可以很清楚的回答下面这个问题

  • 问：基于BIO实现的server端，当建立100个连接时，需要多少个线程？基于NIO实现的呢？
  • 答：基于BIO实现的server端，通常需要由一个线程accept，并为每个新建立的连接创建一个线程去处理IO操作，因此需要 1个accept线程+100个IO线程
    基于NIO实现的server端，使用Selector多路复用机制，由一个线程进行select，为了提高并发可以使用线程池来处理IO操作，通常为了发挥CPU的性能会创建(cpu核数 x 2)个线程来处理IO操作。因此需要 1个select线程 + cpu核数 x 2 个IO线程

更多文章见：https://nc2era.com

上一篇：当使用代理服务器连接互联网时，如何安装Ubuntu桌面版


下一篇：<转>SQL 左外连接，右外连接，全连接，内连接