Java NIO原理分析

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 keys = selector.selectedKeys;
Iterator 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;

while(true) {
readyNums = epoll_wait( epollFd, events, 1024, -1 );

private int updateSelectedKeys {
int entries = pollWrapper.updated;
int numKeysUpdated = 0;
for (int i=0; i

EPollArrayWrapper

EpollArrayWrapper封裝了底層的調用，裡面包含幾個native方法，如：

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;

在openjdk的native目錄（native/sun/nio/ch）裡面可以找到對應的實現EPollArrayWrapper.c。

（這裡順帶提一下，要實現native方法，可以在類里的方法加上native關鍵字，然後編譯成class文件，再轉換輸出.h，c/c++底層實現該頭文件的方法，編譯成so庫，放到對應目錄即可）

在初始化文件方法裡面，可以看到，是通過動態解析載入進來的，最終調用的epoll_create等方法。

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

Channel和Buffer之間的堆外內存

經常會聽見別人說，堆外內存容易泄漏，以及Netty框架裡面採用了堆外內存，減少拷貝提高性能。那麼這裡面的堆外內存指的是什麼?之前懷著一個好奇心，通過read方法，最後追蹤到SocketChannelImpl裡面read方法，裡面調用了IOUtil的read方法。裡面會首先判斷傳入的Buffer是不是DirectBuffer，如果不是（則是HeapByteBuffer），則會創建一個臨時的DirectBuffer，然後再將其複製到堆內。IOUtil.read方法：

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

return var8;
}
}

這裡有一個問題就是，為什麼會需要DirectBuffer以及堆外內存？通過對DirectByteBuffer的創建來分析，可以知道，通過unsafe.allocateMemory(size);來分配內存的，而對於該方法來說，可以說是直接調用malloc返回，這一塊內存是不受GC管理的，也就是所說的：堆外內存容易泄漏。但是對於使用DirectByteBuffer來說，會創建一個Deallocator，註冊到Cleaner裡面，當對象被回收的時候，則會被直接，從而釋放掉內存，減少內存泄漏。要用堆外內存，從上面的創建來看，堆外內存創建後，以long型地址保存的，而堆內內存會受到GC影響，對象會被移動，如果採用堆內內存，進行系統調用的時候，那麼GC就需要停止，否則就會有問題，基於這一點，採用了堆外內存（這一塊參考了R大的理解：https://www.zhihu.com/question/57374068）。

註：堆外內存的創建（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

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！