Netty中怎么使用wakeup實(shí)現(xiàn)線程喚醒

本篇文章為大家展示了Netty中怎么使用wakeup實(shí)現(xiàn)線程喚醒，內(nèi)容簡(jiǎn)明扼要并且容易理解，絕對(duì)能使你眼前一亮，通過(guò)這篇文章的詳細(xì)介紹希望你能有所收獲。

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

1.輪詢IO事件
2.處理IO事件
3.執(zhí)行任務(wù)

在輪詢IO事件的過(guò)程中,在Linux系統(tǒng)下, 使用epoll實(shí)現(xiàn).
涉及的Netty代碼如下

private void select() {

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

}

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

當(dāng)IO線程執(zhí)行以上代碼的時(shí)候, 如果超時(shí)時(shí)間timeoutMillis還沒(méi)有到達(dá)的情況下, IO線程就會(huì)處于阻塞狀態(tài). 這個(gè)時(shí)候如果非IO線程需要向?qū)Χ藢?xiě)數(shù)據(jù), 由于Netty是異步的框架, 它的實(shí)現(xiàn)是非IO線程將寫(xiě)數(shù)據(jù)封裝成一個(gè)任務(wù)提交到IO線程的任務(wù)隊(duì)列里.

當(dāng)任務(wù)提交到任務(wù)隊(duì)列后, 那么就會(huì)面臨一個(gè)問(wèn)題.此時(shí)的IO線程處于阻塞狀態(tài), 是否需要喚醒它呢?
答案是需要喚醒, 之所以要把它喚醒, 是需要讓IO線程可以及時(shí)的處理剛剛非IO線程提交的任務(wù).

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

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

以上代碼, 就是喚醒的代碼, 主要調(diào)用的方法就是wakeup.

接下來(lái)通過(guò)查看它的系統(tǒng)調(diào)用, 弄清楚它到底是如何實(shí)現(xiàn)的.

代碼如下

// 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");
                    // 底層調(diào)用epoll_wait而阻塞
                    int readyChannels = selector.select();
                } catch (Exception x) {
                    x.printStackTrace();
                }
                System.out.print("Success...\r\n");
            }
        }.start();

        // 之所以設(shè)置的時(shí)間比較久, 是為了讓程序暫時(shí)不結(jié)束
        Thread.sleep(5_60_000);
        System.out.print("Thread[" + Thread.currentThread().getName() + "]invoke wakeup\r\n");
        // 喚醒阻塞線程
        selector.wakeup();

    }
}

以上代碼的邏輯比較簡(jiǎn)單, 一個(gè)線程調(diào)用select()方法阻塞, 另一個(gè)線程喚醒它.
首先javac編譯以上代碼, 然后使用一個(gè)查看系統(tǒng)調(diào)用的命令strace.
strace -ff -o strace java WakeUp

具體如何使用strace請(qǐng)童鞋自行Google

執(zhí)行以后, 通過(guò)以下步驟進(jìn)行分析

獲得PID=1141

0,1,2這三個(gè)文件描述符是標(biāo)準(zhǔn)輸入,標(biāo)準(zhǔn)輸出和錯(cuò)誤輸出.
4號(hào)文件描述符是在使用epoll實(shí)現(xiàn)的多路復(fù)用IO創(chuàng)建的一個(gè)文件描述符.
5,6這兩個(gè)文件描述符是一對(duì)管道.
7,8這兩個(gè)文件描述符是一對(duì)套接字.

通過(guò)搜索strace命令打印的文件內(nèi)容, 查看具體的系統(tǒng)調(diào)用方法.

使用grep命令搜索關(guān)鍵字pipe

程序調(diào)用socketpair這個(gè)系統(tǒng)調(diào)用創(chuàng)建套接字.
其中的8和9是兩個(gè)文件描述符,也就是在/proc/1141/fd目錄下的那兩個(gè)8和9文件描述符. 8這個(gè)描述符用來(lái)讀取數(shù)據(jù), 9這個(gè)描述符用來(lái)寫(xiě)入數(shù)據(jù), 這樣就實(shí)現(xiàn)了兩個(gè)進(jìn)程之間的通信.

通過(guò)epoll_create創(chuàng)建4號(hào)文件描述符.
5和7這兩個(gè)文件描述符添加到epoll上(底層是添加到內(nèi)核的紅黑樹(shù)).

在上面的Java代碼中, 當(dāng)調(diào)用int readyChannels = selector.select()方法的時(shí)候, 底層就會(huì)調(diào)用epoll_wait方法, 那么線程就會(huì)阻塞在此.
當(dāng)另一個(gè)線程調(diào)用selector.wakeup()的時(shí)候, 它就會(huì)向6號(hào)文件描述符寫(xiě)入數(shù)據(jù), 通過(guò)pipe通信的方式, 喚醒另一個(gè)阻塞的線程.
可以通過(guò)grep搜索關(guān)鍵字write驗(yàn)證結(jié)論.

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

上述內(nèi)容就是Netty中怎么使用wakeup實(shí)現(xiàn)線程喚醒，你們學(xué)到知識(shí)或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識(shí)儲(chǔ)備，歡迎關(guān)注億速云行業(yè)資訊頻道。