NIO系列1：框架拆解

最近一年用NIO写了很多网络程序，也研究了一些开源NIO网络框架netty、mina等，总结了一下NIO的架构特色。

不管是netty仍是mina它们都在java原生NIO的基础上进行了完善的封装，虽然细节有所不一样，但整体架构思路一致，都大概划分出了如下几个组成部分：

- - transport：传输层的抽象

- - protocol： 协议codec的抽象

- - event model：统一事件模型

- - buffer：底层buffer封装

在彻底屏蔽底层API的同时，对上层应用提供了自身的统一API接口。

框架进行黑盒封装的同时，再进行通用化的接口开放，带来的好处是统一化，但坏处是程序的透明度下降，抽象度提升，增长理解难度和实现难度。

 

下面说说每一个部分的一些设计考量：

transport传输层抽象都是对java原生NIO API的封装，在这一层封装的程度在于框架的实现目标。例如mina立足于通用的网络框架，所以彻底屏蔽了原生的API，提供了自身的统一接口，由于它不只须要封装NIO的API还有一系列其余类型的IO操做的API，提供统一API接口。为了通用兼顾各种传输通道所以可能不得不暴露多余的API接口，使用方需甄别传输通道的不一样，增长了理解难度。

protocol封装各种经常使用协议的codec操做，但目前这些网络框架的codec实现都与自身的API紧密绑定，下降了可重用性。

event model 事件模型的设计一般不能彻底独立，例如java NIO自己的模型是事件驱动的，但传统阻塞型IO并不是事件驱动，要兼顾两者一般要付出额外的代价和开销。

有一种说法是让异步IO同步化使用（由于同步化使用更简单，异步致使了业务处理的碎片化）到底对不对值得商榷？模型阻抗致使的代价和开销屏蔽在了黑盒中，也容易误导应用程序员对本该采用同步化处理的业务却滥用了异步化机制，并不会带来什么好处。

buffer 一般都用来配合底层IO数据流和协议codec使用，自己是否适合暴露给应用方取决于框架是否整合codec，由于codec自己带有业务性质，而纯粹的IO数据流处理使用的buffer则彻底无需暴露给应用方。

 

以上简单说了下NIO框架各部分的设计考量，能够看出目前流行的NIO框架（netty和mina）都在走一条相似“瑞士军刀”的路线，集各类功能与一身（多种IO封装、协议封装），但你又很难把瑞士军刀上的某个刀片拆下来单独使用。

在实践中感受，考虑从单一性、简洁性、重用性、组合性、透明性几个方面去设计原子化的IO组件也许更可取，更像是一种“工具箱”路线。

 

典型的事件驱动模型NIO框架组件交互图以下：

Acceptor：  负责监听链接事件负责接入

Processor：负责IO读写事件处理

EventDispatcher：负责事件派发

Handler：业务处理器

后面将经过一个系列文章来讨论一个原子化的NIO组件实现的细节及设计考量。

 

NIO系列2：TCP监听绑定

分类： 踏莎行·术2012-04-01 08:09 1241人阅读 评论(0) 收藏 举报

bind nio NIO tcp TCP

注：本文适合对象需对java NIO API的使用及异步事件模型（Reactor模式）有必定程度的了解，主要讲述使用java原生NIO实现一个TCP监听绑定的过程及细节设计。

 

咱们一开始设计了一个TCP接入服务类，这个类提供了一个API方法提供对本地一系列地址（端口）的监听绑定，类初始化后完成Selector的open操做以下：

 

selector = Selector.open();

提供的绑定API，其方法签名以下：

 

/**
 * Binds to the specified local addresses and start to accept incoming connections. If any address binding failed then
 * rollback the already binding addresses. Bind is fail fast, if encounter the first bind exception then throw it immediately.
 *
 * @param firstLocalAddress
 * @param otherLocalAddresses
 * @throws throw if bind failed.
 */
synchronized public void bind(SocketAddress firstLocalAddress, SocketAddress... otherLocalAddresses) throws IOException；

为什么须要同步？由于咱们不但愿多个线程同时调用该方法，致使地址绑定异常。

 

参数中能够传递多个本地地址（端口）同时进行监听绑定。

在NIO的绑定过程当中需进行事件注册（对OP_ACCEPT感兴趣），以下：

 

ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
ServerSocket ss = ssc.socket();
ss.setReuseAddress(config.isReuseAddress());
ss.bind(address, config.getBacklog());
ssc.register(selector, SelectionKey.OP_ACCEPT);

 

 

因为注册过程当中除了涉及锁竞争还可能产生死锁，因此通常的作法都是将绑定地址放在队列中进行异步注册由reactor线程进行处理，例如：

 

bindAddresses.addAll(localAddresses);
if (!bindAddresses.isEmpty()) {
    synchronized (lock) {
        // wake up for unblocking the select() to process binding addresses
        selector.wakeup();

        // wait for bind result
        wait0();
    }
}

从同步注册变为异步注册后就存在一个问题，实际注册绑定时可能存在端口已绑定的异常，在异步状况下就须要线程间通讯来通知异常消息，并向调用方反馈。

 

如上面代码片断中的wait0()方法就是等待绑定结果，若出现绑定异常则抛出

 

private void wait0() throws IOException {
while (!this.endFlag) {
    try {
        lock.wait();
    } catch (InterruptedException e) {
        throw new IOException(e);
    }
}

// reset end flag
this.endFlag = false;

if (this.exception != null) {
    IOException e = exception;
    this.exception = null;
    throw e;
}

以上代码也说明了，NIO异步模型转化为同步API致使的模型阻抗付出了额外的代价和开销 --- 线程间通讯。

 

至此，完成了TCP服务监听过程，下文将进一步讲述服务接入和数据传输相关设计细节。

 

NIO系列3：TCP服务接入

分类： 踏莎行·术2012-04-01 08:10 1181人阅读 评论(0) 收藏 举报

nio NIO tcp TCP

注：本文适合对象需对java NIO API的使用及异步事件模型（Reactor模式）有必定程度的了解，主要讲述使用java原生NIO实现一个TCP服务的过程及细节设计。

 

前文讲述了NIO TCP服务绑定过程的实现机制，如今能够开始讲述服务监听启动后如何和处理接入和数据传输相关的细节设计。

在NIO的接入类中有一个Reactor线程，用于处理OP_ACCEPT事件通知，以下:

 

private class AcceptThread extends Thread {
public void run() {
    while (selectable) {
        try {
            int selected = selector.select();

            if (selected > 0) {
                accept();
            }

            // bind addresses to listen
            bind0();

            // unbind canceled addresses
            unbind0();
        } catch (Exception e) {
            LOG.error("Unexpected exception caught while accept", e);
        }
    }

    // if selectable == false, shutdown the acceptor
    try {
        shutdown0();
    } catch (Exception e) {
        LOG.error("Unexpected exception caught while shutdown", e);
    }
}

 

 

当有客户端接入时selector.select()方法返回大于0的整数，并进入accept()方法进行处理，具体以下：

 

 private void accept() {
    Iterator<SelectionKey> it = selector.selectedKeys().iterator();
    while (it.hasNext()) {
        SelectionKey key = it.next();
        it.remove();
        AbstractSession session = (AbstractSession) acceptByProtocol(key);
        Processor processor = pool.get(session);
        session.setProcessor(processor);
        processor.add(session);
    }
}

 

 

 

 protected Session acceptByProtocol(SelectionKey key) {
    if (key == null || !key.isValid() || !key.isAcceptable()) {
           <span style="white-space:pre">       </span>return null;
       <span style="white-space:pre">   </span>}

    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
    SocketChannel sc = null;
    try {
        sc = ssc.accept();
        if(sc == null) {
            return null;
        }
        sc.configureBlocking(false);
    } catch (IOException e) {
        LOG.warn(e.getMessage(), e);
        if(sc != null) {
            try {
                sc.close();
            } catch (IOException ex) {
                LOG.warn(ex.getMessage(), ex);
            }
        }
    }

    Session session = new TcpSession(sc, config);

    return session;
}

 

为每个接入的客户端经过调用NIO原生accept()方法返回一个SocketChannel的抽象，并封装成一个session对象（session的概念来自mina框架）

注意：此时与客户链接的通道还没有注册对读/写事件感兴趣，由于它的注册与前文绑定过程同样须要异步进行。

所以将封装通道的session转交给一个processor对象（io读写处理器，该概念也是来自mina），processor内部维持了一个新建session的队列，在其内部reactor线程循环中进行注册处理。

有关processor处理读写事件的细节设计见下文。

 

 

NIO系列4：TCP服务数据读写

分类： 踏莎行·术2012-04-01 08:10 1289人阅读 评论(0) 收藏 举报

nio NIO tcp TCP

注：本文适合对象需对java NIO API的使用及异步事件模型（Reactor模式）有必定程度的了解，主要讲述使用java原生NIO实现一个TCP服务的过程及细节设计。

 

上文讲到当客户端完成与服务端的链接创建后，为其SocketChannel封装了一个session对象表明这个链接，并交给processor处理。

processor的内部有3个重要的队列，分别存放新建立的session、须要写数据的session和准备关闭的session，以下：

 

/** A Session queue containing the newly created sessions */
private final Queue<AbstractSession> newSessions = new ConcurrentLinkedQueue<AbstractSession>();

/** A queue used to store the sessions to be flushed */
private final Queue<AbstractSession> flushingSessions = new ConcurrentLinkedQueue<AbstractSession>();

/** A queue used to store the sessions to be closed */
private final Queue<AbstractSession> closingSessions = new ConcurrentLinkedQueue<AbstractSession>();

在processor的reactor循环处理线程中，每轮循环的处理包括以下步骤：

 

1. selector.select()，其中为了处理链接超时的状况，select方法中传递了超时参数以避免其永久阻塞，一般是1秒。该方法即时在没有事件发生时每秒返回一次，进入循环检测超时

 

int selected = selector.select(SELECT_TIMEOUT);

2. 从select返回后，首先检查newSessions队列是否有新的session加入，并为其注册监听事件（读事件监听）。session只有在注册完事件后，咱们才认为其状态为open并派发打开事件。（关于session状态，有建立、打开、关闭中、已关闭几种）

 

 

for (AbstractSession session = newSessions.poll(); session != null; session = newSessions.poll()) {
    SelectableChannel sc = session.getChannel();
    SelectionKey key = sc.register(selector, SelectionKey.OP_READ, session);
    session.setSelectionKey(key);

    // set session state open, so we can read / write
    session.setOpened();

    // fire session opened event
    eventDispatcher.dispatch(new Event(EventType.SESSION_OPENED, session, null, handler));

    n++;
}

 

 

3. 有读/写事件时，进行相关处理，每次读写事件发生时更新一次最后的IO时间。

 

// set last IO time
session.setLastIoTime(System.currentTimeMillis());

// Process reads
if (session.isOpened() && isReadable(session)) {
    read(session);
}

// Process writes
if (session.isOpened() && isWritable(session)) {
    asyWrite(session);
}

读取数据时有一个小技巧在于灵活自适应buffer分配（来自mina的一个实现策略），每次判断读取到的字节数若乘以2依然小于buffer大小，则收缩buffer为原来一半，若读取的字节数已装满buffer则扩大一倍。

 

int readBytes = 0;
int ret;
while ((ret = ((SocketChannel) session.getChannel()).read(buf)) > 0) {
    readBytes += ret;
    if (!buf.hasRemaining()) {
        break;
    }
}

if (readBytes > 0) {
    if ((readBytes << 1) < session.getReadBufferSize()) {
        shrinkReadBufferSize(session);
    } else if (readBytes == session.getReadBufferSize()) {
        extendReadBufferSize(session);
    }

    fireMessageReceived(session, buf, readBytes);
}

// read end-of-stream, remote peer may close channel so close session.
if (ret < 0) {
    asyClose(session);
}

 

处理写操做实际上是异步的，老是放入flushSessions中等待写出。

 

 

private void asyWrite(AbstractSession session) {
// Add session to flushing queue, soon after it will be flushed in the same select loop.
flushingSessions.add(session);

4. 如有须要写数据的session，则进行flush操做。

写事件通常默认都是不去关注的，由于在TCP缓冲区可写或远端断开或IO错误发生时都会触发该事件，容易诱发服务端忙循环从而CPU100%问题。为了保证读写公平，写buffer的大小设置为读buffer的1.5倍（来自mina的实现策略），每次写数据前设置为对写事件再也不感兴趣。限制每次写出数据大小的缘由除了避免读写不公平，也避免某些链接有大量数据须要写出时一次占用了过多的网络带宽而其余链接的数据写出被延迟从而影响了公平性。

 

// First set not be interested to write event
etInterestedInWrite(session, false);

首先向TCP缓冲区写出数据（NIO的原生API操做都是不阻塞的）

 

 

int qota = maxWrittenBytes - writtenBytes;
int localWrittenBytes = write(session, buf, qota);

写完后根据返回的写出数据字节数，可能存在如下多种状况：

 

- - buffer一次写完，则派发消息已经发送事件

 

// The buffer is all flushed, remove it from write queue
    if (!buf.hasRemaining()) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("The buffer is all flushed, remove it from write queue");
        }

        writeQueue.remove();

        // fire message sent event
        eventDispatcher.dispatch(new Event(EventType.MESSAGE_SENT, session, buf.array(), handler));
    }

- - 若返回的写入字节数为0，多是TCP缓存buffer已满，则注册对写事件感兴趣，稍待下次再写。

 

 

// 0 byte be written, maybe kernel buffer is full so we re-interest in writing and later flush it.
if (localWrittenBytes == 0) {
    if (LOG.isDebugEnabled()) {
        LOG.debug("0 byte be written, maybe kernel buffer is full so we re-interest in writing and later flush it");
    }

    setInterestedInWrite(session, true);
    flushingSessions.add(session);
    return;
}

- - 若一次写入没有写完buffer中的数据，依然注册对写事件感兴趣，稍待下次再写。

 

 

// The buffer isn't empty(bytes to flush more than max bytes), we re-interest in writing and later flush it.
    if (localWrittenBytes > 0 && buf.hasRemaining()) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("The buffer isn't empty(bytes to flush more than max bytes), we re-interest in writing and later flush it");
        }

        setInterestedInWrite(session, true);
        flushingSessions.add(session);
        return;
    }

- - 一次写入数据太多时，为了保证公平性，依然下次再写入

 

 

// Wrote too much, so we re-interest in writing and later flush other bytes.
if (writtenBytes >= maxWrittenBytes && buf.hasRemaining()) {
    if (LOG.isDebugEnabled()) {
        LOG.debug("Wrote too much, so we re-interest in writing and later flush other bytes");
    }

    setInterestedInWrite(session, true);
    flushingSessions.add(session);
    return;
}

5. 有须要关闭的session，则进行关闭操做。引起关闭session的操做可能来自应用方主动关闭，也多是因为IO异常后自动关闭。因为关闭session可能存在多线程调用，为了不锁同步，咱们经过状态检测来规避用锁机制提升效率。

 

关闭session的操做具体来讲就是对channel.close()和key.cancel()，这2个操做后其实尚未彻底释放socket占用的文件描述符，需等到下次select()操做后，一些NIO框架会主动调用，因为咱们这里select(TIMEOUT)带有超时参数会自动唤醒，所以不存在这个问题。

 

private int close() throws IOException {
    int n = 0;
    for (AbstractSession session = closingSessions.poll(); session != null; session = closingSessions.poll()) {
        if (LOG.isDebugEnabled()) { LOG.debug("Closing session: " + session); }

        if (session.isClosed()) {
            if (LOG.isDebugEnabled()) { LOG.debug("Escape close session, it has been closed: " + session); }
            continue;
        }

        session.setClosing();

        close(session);
        n++;

        session.setClosed();

        // fire session closed event
        eventDispatcher.dispatch(new Event(EventType.SESSION_CLOSED, session, null, handler));

        if (LOG.isDebugEnabled()) { LOG.debug("Closed session: " + session); }
    }
    return n;

 

NIO系列5：事件模型

分类： 踏莎行·术2012-04-01 08:10 1263人阅读 评论(0) 收藏 举报

event Event nio NIO

前文讲述了NIO数据读写处理，那么这些数据最终如何被递交给上层业务程序进行处理的呢？

NIO框架通常都采用了事件派发模型来与业务处理器交互，它与原生NIO的事件机制是模型匹配的，缺点是带来了业务处理的碎片化。须要业务程序开发者对事件的生命周期有一个清晰的了解，不像传统方式那么直观。

事件派发器（EventDispatcher）就成为了NIO框架中IO处理线程和业务处理回调接口（Handler）之间的桥梁。

因为业务处理的时间长短是难以肯定的，因此通常事件处理器都会分离IO处理线程，使用新的业务处理线程池来进行事件派发，回调业务接口实现。

 

下面经过一段示例代码来讲明事件的派发过程：

这是processor从网络中读取到一段字节后发起的MESSAGE_RECEIVED事件，调用了eventDispatcher.dispatch(Event e)方法。

 

private void fireMessageReceived(AbstractSession session, ByteBuffer buf, int length) {
        // fire message received event, here we copy buffer bytes to a new byte array to avoid handler expose <code>ByteBuffer</code> to end user.
        byte[] barr = new byte[length];
        System.arraycopy(buf.array(), 0, barr, 0, length);
        eventDispatcher.dispatch(new Event(EventType.MESSAGE_RECEIVED, session, barr, handler));
    }

dispatch的方法实现有如下关键点须要考虑：

1. 事件派发是多线程的，派发线程最终会调用业务回调接口来进行事件处理，回调接口由业务方实现自身去保证线程并发性和安全性。

2. 对于TCP应用来讲，由同一session（这里可表明同一个链接）收到的数据必须保证有序派发，不一样的session可无序。

3. 不一样session的事件派发要尽量保证公平性，例如：session1有大量事件产生致使派发线程繁忙时，session2产生一个事件不会由于派发线程都在忙于处理session1的事件而被积压，session2的事件也能尽快获得及时派发。

下面是一个实现思路的代码示例：

 

public void dispatch(Event event) {
        AbstractSession s = (AbstractSession) event.getSession();
        s.add(event);
        if (!s.isEventProcessing()) {
            squeue.offer(s);
        }
    }

为了保证每一个session的事件有序，咱们将事件存放在每一个session自身包含的队列中，而后再将session放入一个公共的阻塞队列中。

 

有一组worker线程在监听阻塞队列，一旦有session进入队列，它们被激活对session进行事件派发，以下：

 

public void run() {
            try {
                for (AbstractSession s = squeue.take(); s != null; s = squeue.take()) {
                    // first check any worker is processing this session? if any other worker thread is processing this event with same session, just ignore it.
                    synchronized (s) {
                        if (!s.isEventProcessing()) {
                            s.setEventProcessing(true);
                        } else {
                            continue;
                        }
                    }

                    // fire events with same session
                    fire(s);

                    // last reset processing flag and quit current thread processing
                    s.setEventProcessing(false);

                    // if remaining events, so re-insert to session queue
                    if (s.getEventQueue().size() > 0 && !s.isEventProcessing()) {
                        squeue.offer(s);
                    }
                }
            } catch (InterruptedException e) {
                LOG.warn(e.getMessage(), e);
            }
        }

这里的要点在于，worker不止一个，但为了保证每一个session的事件有序咱们只能让惟一一个线程对session进行处理，所以能够看到上面的代码中一开始对session进行了加锁，并改变了session的状态（置为事件处理中）。

 

退出临界区后，进入事件派发处理方法fire()，在fire()方法退出前其余线程都没有机会对该session进行处理，保证了同一时刻只有一个线程进行处理的约束。

若是某个session一直不断有数据进入，则派发线程可能在fire()方法中停留很长时间，具体看fire()的实现以下：

 

private void fire(Session s) {
            int count = 0;
            Queue<Event> q = s.getEventQueue();
            for (Event event = q.poll(); event != null; event = q.poll()) {
                event.fire();
                count++;
                if (count > SPIN_COUNT) {
                    // quit loop to avoid stick same worker thread by same session
                    break;
                }
            }
        }

从上面代码能够看出，每次fire()的循环数被设置了一个上限，若事件太多时每次达到上限会退出循环释放线程，等下一次再处理。

 

当前线程释放对session的控制权只需简单置事件处理状态为false，其余线程就有机会从新获取该session的控制权。

在最后退出前为了不事件遗漏，由于可能当前线程由于处理事件达到上限数被退出循环而又没有新的事件进入阻塞队列触发新的线程激活，则由当前线程主动去从新将该session放入阻塞队列中激活新线程。