Netty线程模型

作者: 隔壁王哥 | 来源:发表于2022-04-30 22:28 被阅读0次

概述

Netty是一个高性能网络通信框架，其高性能和采用的线程模型密切相关。Netty采用Reactor模式，Reactor 线程模型有三种：Reactor单线程模型、Reactor多线程模型、主从Reactor多线程模型。Netty可以在启动时通过配置灵活的支持这三种线程模型。

传统服务设计

accept方法会阻塞请求，需要为每个请求分配线程进行处理，如果并发量大会耗尽系统资源，导致系统崩溃。

class Server implements Runnable {
    public void run() {
        try {
            ServerSocket ss = new ServerSocket(PORT);
            while (!Thread.interrupted())
                new Thread(new Handler(ss.accept())).start();
        // or, single-threaded, or a thread pool
        } catch (IOException ex) { /* ... */ }
    }

    static class Handler implements Runnable {
        final Socket socket;

        Handler(Socket s) {
            socket = s;
        }

        public void run() {
            try {
                byte[] input = new byte[MAX_INPUT];
                socket.getInputStream().read(input);
                byte[] output = process(input);
                socket.getOutputStream().write(output);
            } catch (IOException ex) { /* ... */ }
        }

        private byte[] process(byte[] cmd) { /* ... */ }
    }
}

Reactor模式

Reactor模式又叫反应堆模式，是基于事件驱动的线程模型。使用IO多路复用模型，Reactor线程通过select监听事件，监听到事件后通过dispatch进行分发，如果是连接事件则通过acceptor处理，如果是读写事件则通过对应的handler进行处理。

Reactor单线程模型

所有的连接与读写事件都由Reactor线程进行分发处理，如果某个处理阻塞会导致其他连接请求无法处理。

class Reactor implements Runnable {
    final Selector selector;
    final ServerSocketChannel serverSocket;

    Reactor(int port) throws IOException {
        selector = Selector.open();
        serverSocket = ServerSocketChannel.open();
        serverSocket.socket().bind(new InetSocketAddress(port));
        serverSocket.configureBlocking(false);
        SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
        sk.attach(new Acceptor());
    }

    public void run() {
        {
            while (!Thread.interrupted()) {
                selector.select();
                Set selected = selector.selectedKeys();
                Iterator it = selected.iterator();
                while (it.hasNext()) dispatch((SelectionKey) (it.next()));
                selected.clear();
            }
        } catch(IOException ex){ /* ... */ }
    }

    void dispatch(SelectionKey k) {
        Runnable r = (Runnable) (k.attachment());
        if (r != null) r.run();
    }

    class Acceptor implements Runnable {
        public void run() {
            try {
                SocketChannel c = serverSocket.accept();
                if (c != null) new Handler(selector, c);
            } catch (IOException ex) { /* ... */ }
        }
    }
}
final class Handler implements Runnable {
    final SocketChannel socket;
    final SelectionKey sk;
    ByteBuffer input = ByteBuffer.allocate(MAXIN);
    ByteBuffer output = ByteBuffer.allocate(MAXOUT);
    static final int READING = 0, SENDING = 1;
    int state = READING;

    Handler(Selector sel, SocketChannel c) throws IOException {
        socket = c;
        c.configureBlocking(false);
        sk = socket.register(sel, 0);
        sk.attach(this);
        sk.interestOps(SelectionKey.OP_READ);
        sel.wakeup();
    }

    boolean inputIsComplete() { /* ... */ }

    boolean outputIsComplete() { /* ... */ }

    void process() { /* ... */ }

    public void run() {
        try {
            if (state == READING) read();
            else if (state == SENDING) send();
        } catch (IOException ex) { /* ... */ }
    }

    void read() throws IOException {
        socket.read(input);
        if (inputIsComplete()) {
            process();
            state = SENDING;
            sk.interestOps(SelectionKey.OP_WRITE);
        }
    }

    void send() throws IOException {
        socket.write(output);
        if (outputIsComplete()) sk.cancel();
    }
}

Reactor多线程模型

Reactor线程监听到连接事件则分发给acceptor进行处理，如果不是连接事件则分发给handler进行处理，handler读取数据后交给线程池处理具体的业务逻辑，等处理完成后再由handler将结果发送给客户端。

class Handler implements Runnable {
    // uses util.concurrent thread pool
    static PooledExecutor pool = new PooledExecutor(...);
    static final int PROCESSING = 3;

    // ...
    synchronized void read() { // ...
        socket.read(input);
        if (inputIsComplete()) {
            state = PROCESSING;
            pool.execute(new Processer());
        }
    }

    synchronized void processAndHandOff() {
        process();
        state = SENDING; // or rebind attachment
        sk.interest(SelectionKey.OP_WRITE);
    }

    class Processer implements Runnable {
        public void run() {
            processAndHandOff();
        }
    }
}

主从Reactor多线程模型

连接与响应事件分离，mainReactor只负责处理连接事件，处理完成后将连接绑定到subReactor中的select上，由subReactor负责监听后续事件进行处理。

Selector[] selectors;// also create threads
int next = 0;
class Acceptor { // ...
    public synchronized void run() { ...
        Socket connection = serverSocket.accept();
        if (connection != null)
            new Handler(selectors[next], connection);
        if (++next == selectors.length)
            next = 0;
    }
}