Netty中怎么使用wakeup實現線程喚醒

本篇文章為大家展示了Netty中怎么使用wakeup實現線程喚醒，內容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

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

1.輪詢IO事件
2.處理IO事件
3.執行任務

在輪詢IO事件的過程中,在Linux系統下, 使用epoll實現.
涉及的Netty代碼如下

private void select() {

    // ...
    int selectedKeys = selector.select(timeoutMillis);
    // ...

}

具體源碼位置:
io.netty.channel.nio.NioEventLoop#select

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

當任務提交到任務隊列后, 那么就會面臨一個問題.此時的IO線程處于阻塞狀態, 是否需要喚醒它呢?
答案是需要喚醒, 之所以要把它喚醒, 是需要讓IO線程可以及時的處理剛剛非IO線程提交的任務.

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

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

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

接下來通過查看它的系統調用, 弄清楚它到底是如何實現的.

代碼如下

// WakeUp.java
import 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編譯以上代碼, 然后使用一個查看系統調用的命令strace.
strace -ff -o strace java WakeUp

具體如何使用strace請童鞋自行Google

執行以后, 通過以下步驟進行分析

獲得PID=1141

0,1,2這三個文件描述符是標準輸入,標準輸出和錯誤輸出.
4號文件描述符是在使用epoll實現的多路復用IO創建的一個文件描述符.
5,6這兩個文件描述符是一對管道.
7,8這兩個文件描述符是一對套接字.

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

使用grep命令搜索關鍵字pipe

程序調用socketpair這個系統調用創建套接字.
其中的8和9是兩個文件描述符,也就是在/proc/1141/fd目錄下的那兩個8和9文件描述符. 8這個描述符用來讀取數據, 9這個描述符用來寫入數據, 這樣就實現了兩個進程之間的通信.

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

在上面的Java代碼中, 當調用int readyChannels = selector.select()方法的時候, 底層就會調用epoll_wait方法, 那么線程就會阻塞在此.
當另一個線程調用selector.wakeup()的時候, 它就會向6號文件描述符寫入數據, 通過pipe通信的方式, 喚醒另一個阻塞的線程.
可以通過grep搜索關鍵字write驗證結論.

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

上述內容就是Netty中怎么使用wakeup實現線程喚醒，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。