Java NIO principle analysis and basic use

Java NIO principle analysis

This article mainly focuses on Java NIO, from the basic use of Java NIO, to the introduction of NIO API under Linux, to the underlying implementation principle of Java Selector .

• Basic use of Java NIO

• Introduction to NIO system calls under Linux

• Selector principle

• Off-heap memory between Channel and Buffer

Basic use of Java NIO

It can be found from the JDK NIO document, Java divides it into three major blocks: Channel, Buffer and multiplexing Selector. The existence of Channel encapsulates the connection channel to any entity (such as network/file); Buffer encapsulates the buffer storage of data. Finally, Selector provides a single-threaded non-blocking way to process multiple connect.

Basic application example

The basic steps of NIO are to create Selector and ServerSocketChannel, then register the ACCEPT event of the channel, call the select method, wait for the arrival of the connection, and register it after receiving the connection. Selector. The following is an example of Echo Server:

public class SelectorDemo {

    public static void main(String[] args) throws IOException {


        Selector selector = Selector.open();
        ServerSocketChannel socketChannel = ServerSocketChannel.open();
        socketChannel.bind(new InetSocketAddress(8080));
        socketChannel.configureBlocking(false);
        socketChannel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            int ready = selector.select();
            if (ready == 0) {
                continue;
            } else if (ready < 0) {
                break;
            }

            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = keys.iterator();
            while (iterator.hasNext()) {

                SelectionKey key = iterator.next();
                if (key.isAcceptable()) {

                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    SocketChannel accept = channel.accept();
                    if (accept == null) {
                        continue;
                    }
                    accept.configureBlocking(false);
                    accept.register(selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    // 读事件
                    deal((SocketChannel) key.channel(), key);
                } else if (key.isWritable()) {
                    // 写事件
                    resp((SocketChannel) key.channel(), key);
                }
                // 注：处理完成后要从中移除掉
                iterator.remove();
            }
        }
        selector.close();
        socketChannel.close();
    }

    private static void deal(SocketChannel channel, SelectionKey key) throws IOException {

        ByteBuffer buffer = ByteBuffer.allocate(1024);
        ByteBuffer responseBuffer = ByteBuffer.allocate(1024);

        int read = channel.read(buffer);

        if (read > 0) {
            buffer.flip();
            responseBuffer.put(buffer);
        } else if (read == -1) {
            System.out.println("socket close");
            channel.close();
            return;
        }

        key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        key.attach(responseBuffer);
    }

    private static void resp(SocketChannel channel, SelectionKey key) throws IOException {

        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.flip();

        channel.write(buffer);
        if (!buffer.hasRemaining()) {
            key.attach(null);
            key.interestOps(SelectionKey.OP_READ);
        }
    }
}

Introduction to NIO system calls under Linux

In the Linux environment, several ways are provided to implement NIO, such as epoll, poll, select, etc. For select/poll, each call requires FD and monitoring events to be passed in from the outside. This results in the need to copy these data from the user state to the kernel state every time the call is made, resulting in a cost comparison for each call. It is large, and every time it is returned from select/poll, it is the full amount of data. You need to traverse it yourself to check which ones are READY. For epoll, it is incremental. The system maintains the required FD and monitoring events internally. When you want to register, just call epoll_ctl. Every time you call, you no longer need to pass it in. When you return, you only return READY listening events and FD. Here is a simple pseudocode:
For details, you can read the previous article:

// 1. 创建server socket
// 2. 绑定地址
// 3. 监听端口
// 4. 创建epoll
int epollFd = epoll_create(1024);
// 5. 注册监听事件
struct epoll_event event;
event.events = EPOLLIN | EPOLLRDHUP | EPOLLET;
event.data.fd = serverFd;
epoll_ctl(epollFd, EPOLL_CTL_ADD, serverFd, &event);

while(true) {
    readyNums = epoll_wait( epollFd, events, 1024, -1 );
    
    if ( readyNums < 0 )
     {
         printf("epoll_wait error\n");
         exit(-1);
     }

     for ( i = 0; i <  readyNums; ++i)
     {
         if ( events[i].data.fd == serverFd )
         {
             clientFd = accept( serverFd, NULL, NULL );
             // 注册监听事件
             ...
         }else if ( events[i].events & EPOLLIN )
         {
            // 处理读事件
         }else if ( events[i].events & EPOLLRDHUP )
         {
            // 关闭连接事件
            close( events[i].data.fd );
         }
}

Selector principle

SelectionKey

From the perspective of Java top-level users, The channel returns SelectionKey through registration, and the Selector.select method is also used by returning SelectionKey. So why is this class needed here? What does this class do? Regardless of the language, it is inseparable from the support of the underlying system. Through the above-mentioned basic applications under Linux, we can know that through system calls, parameters such as FD and events are passed and returned to it. From a design perspective, , there needs to be a mapping relationship so that it can be associated. The Channel encapsulation here is passed. If you put the parameters of the READY event inside, it is not appropriate. At this time, the SelectionKey appears. Inside the SelectionKey, save the reference to the Channel. And some event information, and then the Selector finds the SelectionKey through FD for association. In the underlying EP, there is an attribute: Map<Integer,SelectionKeyImpl> fdToKey. <h3>EPollSelectorImpl</h3><p>In Linux 2.6+ version, Java NIO uses epoll (i.e. <code>EPollSelectorImpl class). For 2.4.x, use poll (i.e. PollSelectorImpl class), here we take epoll as an example.

select method

The top-level Selector, by calling the select method, will eventually call the EPollSelectorImpl.doSelect method. Through this method, you can see that it will first process some events that are no longer registered. Call pollWrapper.poll(timeout);, and then clean it up again. Finally, you can see that the mapping relationship needs to be processed

protected int doSelect(long timeout)
    throws IOException
{
    if (closed)
        throw new ClosedSelectorException();
    // 处理一些不再注册的事件
    processDeregisterQueue();
    try {
        begin();
        pollWrapper.poll(timeout);
    } finally {
        end();
    }
    // 再进行一次清理
    processDeregisterQueue();
    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++) {
        // 获取FD
        int nextFD = pollWrapper.getDescriptor(i);
        // 根据FD找到对应的SelectionKey
        SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
        // ski is null in the case of an interrupt
        if (ski != null) {
            // 找到该FD的READY事件
            int rOps = pollWrapper.getEventOps(i);
            if (selectedKeys.contains(ski)) {
                // 将底层的事件转换为Java封装的事件,SelectionKey.OP_READ等
                if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
                    numKeysUpdated++;
                }
            } else {
                // 没有在原有的SelectedKey里面，说明是在等待过程中加入的
                ski.channel.translateAndSetReadyOps(rOps, ski);
                if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
                    // 需要更新selectedKeys集合
                    selectedKeys.add(ski);
                    numKeysUpdated++;
                }
            }
        }
    }
    // 返回Ready的Channel个数
    return numKeysUpdated;
}

EPollArrayWrapper

EpollArrayWrapper encapsulates the underlying calls , which contains several native methods, such as:

private native int epollCreate();
private native void epollCtl(int epfd, int opcode, int fd, int events);
private native int epollWait(long pollAddress, int numfds, long timeout,
                             int epfd) throws IOException;

The corresponding implementation EPollArrayWrapper.c can be found in the native directory of openjdk (native/sun/nio/ch).
(By the way, to implement the native method, you can add the native keyword to the method in the class, then compile it into a class file, and then convert and output .h, the bottom layer of c/c++ to implement the method of the header file, compile into the so library, just put it in the corresponding directory)
In the initialization file method, you can see that it is loaded in through dynamic parsing, and the epoll_create and other methods are finally called.

JNIEXPORT void JNICALL
Java_sun_nio_ch_EPollArrayWrapper_init(JNIEnv *env, jclass this)
{
    epoll_create_func = (epoll_create_t) dlsym(RTLD_DEFAULT, "epoll_create");
    epoll_ctl_func    = (epoll_ctl_t)    dlsym(RTLD_DEFAULT, "epoll_ctl");
    epoll_wait_func   = (epoll_wait_t)   dlsym(RTLD_DEFAULT, "epoll_wait");

    if ((epoll_create_func == NULL) || (epoll_ctl_func == NULL) ||
        (epoll_wait_func == NULL)) {
        JNU_ThrowInternalError(env, "unable to get address of epoll functions, pre-2.6 kernel?");
    }
}

Off-heap memory between Channel and Buffer

I often hear people say that off-heap memory is easy to leak, and the Netty framework uses off-heap memory to reduce copies and improve performance. So what does the off-heap memory here refer to? With a curiosity, through the read method, I finally traced it to the read method in SocketChannelImpl, which called the read method of IOUtil. It will first determine whether the incoming Buffer is a DirectBuffer. If not (it is a HeapByteBuffer), a temporary DirectBuffer will be created and then copied to the heap. IOUtil.read method:

static int read(FileDescriptor var0, ByteBuffer var1, long var2, NativeDispatcher var4, Object var5) throws IOException {
    if(var1.isReadOnly()) {
        throw new IllegalArgumentException("Read-only buffer");
    } else if(var1 instanceof DirectBuffer) {
        // 为堆外内存，则直接读取
        return readIntoNativeBuffer(var0, var1, var2, var4, var5);
    } else {
        // 为堆内内存，先获取临时堆外内存
        ByteBuffer var6 = Util.getTemporaryDirectBuffer(var1.remaining());

        int var8;
        try {
            // 读取到堆外内存
            int var7 = readIntoNativeBuffer(var0, var6, var2, var4, var5);
            var6.flip();
            if(var7 > 0) {
                // 复制到堆内
                var1.put(var6);
            }

            var8 = var7;
        } finally {
            // 释放临时堆外内存
            Util.offerFirstTemporaryDirectBuffer(var6);
        }

        return var8;
    }
}

这里有一个问题就是，为什么会需要DirectBuffer以及堆外内存？通过对DirectByteBuffer的创建来分析，可以知道，通过unsafe.allocateMemory(size);来分配内存的，而对于该方法来说，可以说是直接调用malloc返回，这一块内存是不受GC管理的，也就是所说的：堆外内存容易泄漏。但是对于使用DirectByteBuffer来说，会创建一个Deallocator，注册到Cleaner里面，当对象被回收的时候，则会被直接，从而释放掉内存，减少内存泄漏。要用堆外内存，从上面的创建来看，堆外内存创建后，以long型地址保存的，而堆内内存会受到GC影响，对象会被移动，如果采用堆内内存，进行系统调用的时候，那么GC就需要停止，否则就会有问题，基于这一点，采用了堆外内存（这一块参考了R大的理解：）。

注：堆外内存的创建（unsafe.cpp）：

// 仅仅作了对齐以及将长度放在数组前方就返回了
UNSAFE_ENTRY(jlong, Unsafe_AllocateMemory(JNIEnv *env, jobject unsafe, jlong size))
  UnsafeWrapper("Unsafe_AllocateMemory");
  size_t sz = (size_t)size;
  if (sz != (julong)size || size < 0) {
    THROW_0(vmSymbols::java_lang_IllegalArgumentException());
  }
  if (sz == 0) {
    return 0;
  }
  sz = round_to(sz, HeapWordSize);
  void* x = os::malloc(sz);
  if (x == NULL) {
    THROW_0(vmSymbols::java_lang_OutOfMemoryError());
  }
  //Copy::fill_to_words((HeapWord*)x, sz / HeapWordSize);
  return addr_to_java(x);
UNSAFE_END

The above is the detailed content of Java NIO principle analysis and basic use. For more information, please follow other related articles on the PHP Chinese website!