Netty之線程喚醒wakeup

首先回顧下, Netty中的IO線程主要完成三件事

在輪詢IO事件的過程中,在Linux系統(tǒng)下, 使用epoll實現(xiàn).

涉及的Netty代碼如下

當IO線程執(zhí)行以上代碼的時候, 如果超時時間timeoutMillis還沒有到達的情況下, IO線程就會處于阻塞狀態(tài). 這個時候如果非IO線程需要向對端寫數(shù)據(jù), 由于Netty是異步的框架, 它的實現(xiàn)是非IO線程將寫數(shù)據(jù)封裝成一個任務提交到IO線程的任務隊列里.

@Override
protected void wakeup(boolean inEventLoop) {
    if (!inEventLoop && wakenUp.compareAndSet(false, true)) {
        // 喚醒IO線程
        selector.wakeup();
    }
}

源碼位置: io.netty.channel.nio.NioEventLoop#wakeup

以上代碼, 就是喚醒的代碼, 主要調用的方法就是wakeup.

接下來通過查看它的系統(tǒng)調用, 弄清楚它到底是如何實現(xiàn)的.

代碼如下

// WakeUp.javaimport java.net.InetSocketAddress;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;
public class WakeUp {
    public static void main(String[] args) throws Exception {
        ServerSocketChannel serverSocketChannel;
        Selector selector = Selector.open();        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress("127.0.0.1", 8080), 64);        serverSocketChannel.configureBlocking(false);        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        new Thread() {            @Override            public void run() {                try {                    System.out.print("Thread[" + Thread.currentThread().getName() + "]invoke select\r\n");                    // 底層調用epoll_wait而阻塞                    int readyChannels = selector.select();                } catch (Exception x) {                    x.printStackTrace();                }                System.out.print("Success...\r\n");            }        }.start();
        // 之所以設置的時間比較久, 是為了讓程序暫時不結束        Thread.sleep(5_60_000);        System.out.print("Thread[" + Thread.currentThread().getName() + "]invoke wakeup\r\n");        // 喚醒阻塞線程        selector.wakeup();
    }}

以上代碼的邏輯比較簡單, 一個線程調用select()方法阻塞, 另一個線程喚醒它.

首先javac編譯以上代碼, 然后使用一個查看系統(tǒng)調用的命令strace.

strace -ff -o strace java WakeUp

執(zhí)行以后, 通過以下步驟進行分析

使用jps查看進程ID號

獲得PID=2114

進入 /proc/2114/fd目錄下, 就可以查看到當前進程(PID=2114)打開的文件描述符

0,1,2這三個文件描述符是標準輸入,標準輸出和錯誤輸出.

4號文件描述符是在使用epoll實現(xiàn)的多路復用IO創(chuàng)建的一個文件描述符.

5,6這兩個文件描述符是一對管道.

7,8這兩個文件描述符是一對套接字.

在上面執(zhí)行strace命令的時候, 在它的同目錄下會生成如下文件

通過搜索strace命令打印的文件內容, 查看具體的系統(tǒng)調用方法.

使用grep命令搜索關鍵字pipe

程序調用pipe這個系統(tǒng)調用創(chuàng)建管道. 

其中的5和6是兩個文件描述符,也就是在/proc/2114/fd目錄下的那兩個5和6文件描述符.

5這個描述符用來讀取數(shù)據(jù), 6這個描述符用來寫入數(shù)據(jù), 這樣就實現(xiàn)了兩個進程之間的通信.

使用grep命令搜索關鍵字socket

程序調用socketpair這個系統(tǒng)調用創(chuàng)建套接字.

其中的8和9是兩個文件描述符,也就是在/proc/2114/fd目錄下的那兩個8和9文件描述符. 8這個描述符用來讀取數(shù)據(jù), 9這個描述符用來寫入數(shù)據(jù), 這樣就實現(xiàn)了兩個進程之間的通信.

使用grep命令搜索關鍵字epoll

通過epoll_create創(chuàng)建4號文件描述符.

5和7這兩個文件描述符添加到epoll上(底層是添加到內核的紅黑樹).

在上面的Java代碼中, 當調用int readyChannels = selector.select()方法的時候, 底層就會調用epoll_wait方法, 那么線程就會阻塞在此.

當另一個線程調用selector.wakeup()的時候, 它就會向6號文件描述符寫入數(shù)據(jù), 通過pipe通信的方式, 喚醒另一個阻塞的線程.

可以通過grep搜索關鍵字write驗證結論.

通過write系統(tǒng)調用向6號文件描述符寫入數(shù)據(jù), 具體數(shù)據(jù)沒有任何含義, 它就是想喚醒阻塞的線程.  與6號文件描述符對應的是5號文件描述符. 由于epoll管理著5號文件描述符, 這樣epoll發(fā)現(xiàn)有文件描述符就緒(5號文件描述符就緒), 被阻塞的線程也就會被操作系統(tǒng)重新調度.

簡單介紹了Netty中IO線程如何阻塞和被喚醒的底層系統(tǒng)調用.