Netty之线程唤醒wakeup-技术圈

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

IO线程三件事

轮询IO事件

处理IO事件

执行任务

在轮询IO事件的过程中,在Linux系统下, 使用epoll实现.

涉及的Netty代码如下

private void select() {

// ...

int selectedKeys =

selector.select(timeoutMillis);

// ...

}

具体源码位置:

io.netty.channel.nio.NioEventLoop#select

多路复用

常见的多路复用实现所依赖的系统调用方法包括select(),poll(),epoll(). 不同的平台, 具体的系统调用方法也不同.

当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.

IO线程调用select方法被阻塞, 非IO线程通过调用wakeup方法将IO线程唤醒.

接下来通过查看它的系统调用, 弄清楚它到底是如何实现的.

环境

1. Oracle VM VirtualBox

2. 5.10.0-kali3-amd64

代码如下

// 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();
    }}