一文詳解Reactor模型與實(shí)現(xiàn)示例

更新時(shí)間：2023年03月16日 14:00:01 作者：半夏之沫

這篇文章主要為大家介紹了Reactor模型與實(shí)現(xiàn)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

周六在公司寫Reactor模型，

一女同事問我為啥都2023年了還在學(xué)習(xí)Reactor模型呀，

我問她為啥快30的年紀(jì)了，周六還在公司看我寫Reactor呀，

一時(shí)間辦公室里，男的，女的，都沉默了。

在網(wǎng)絡(luò)IO設(shè)計(jì)中，有兩種高性能模型：Reactor模型和Proactor模型。Reactor基于同步IO模式，Proactor基于異步IO模式。

Netty網(wǎng)絡(luò)框架，Redis等中間件中都有使用到Reactor模型。本文將對(duì)Reactor模型的如下三種分類進(jìn)行學(xué)習(xí)和實(shí)現(xiàn)。

單Reactor單線程模型；
單Reactor多線程模型；
主從Reactor多線程模型。

如果不具備網(wǎng)絡(luò)IO的相關(guān)知識(shí)，建議先閱讀Java網(wǎng)絡(luò)IO模型分析與實(shí)現(xiàn)。

正文

一. Reactor設(shè)計(jì)模式

Reactor翻譯過來的意思是：反應(yīng)堆，所以Reactor設(shè)計(jì)模式本質(zhì)是基于事件驅(qū)動(dòng)的。在Reactor設(shè)計(jì)模式中，存在如下幾個(gè)角色。

Handle（事件）。Reactor整體是基于Handle進(jìn)行驅(qū)動(dòng)，這里的Handle叫做事件，可以類比為BIO中的Socket，NIO中的Socket管道。比如當(dāng)Socket管道有連接建立，或者有數(shù)據(jù)可讀，那么此時(shí)就稱作事件發(fā)生；
EventHandler(事件處理器)。有事件發(fā)生，就需要有相應(yīng)的組件來處理事件，那么這里的組件就叫做事件處理器。EventHandler是一個(gè)抽象概念，其會(huì)有不同的具體實(shí)現(xiàn)，因?yàn)槭录?huì)有不同的類型，那么不同類型的事件，肯定都需要有相應(yīng)的具體處理邏輯，這里的具體處理邏輯，就由EventHandler的具體實(shí)現(xiàn)來承載；
Concrete Event Handler（具體事件處理器）。是EventHandler的具體實(shí)現(xiàn)，用于處理不同類型的事件；
Synchronous Event Demultiplexer（事件多路分解器）。（這里將Synchronous Event Demultiplexer簡(jiǎn)稱為Demultiplexer）Demultiplexer用于監(jiān)聽事件并得到所有發(fā)生事件的集合，在監(jiān)聽的狀態(tài)下是阻塞的，直到有事件發(fā)生為止。Demultiplexer有一個(gè)很好的類比，就是NIO中的多路復(fù)用器Selector，當(dāng)調(diào)用Selector的select() 方法后，會(huì)進(jìn)入監(jiān)聽狀態(tài)，當(dāng)從select() 方法返回時(shí)，會(huì)得到SelectionKey的一個(gè)集合，而每一個(gè)SelectionKey中就保存著有事件發(fā)生的Socket管道；
Initiation Dispatcher（事件分發(fā)器）?，F(xiàn)在已經(jīng)有Concrete Event Handler（具體事件處理器）來處理不同的事件，也能通過Synchronous Event Demultiplexer（事件多路分解器）拿到發(fā)生的事件，那么最后需要做的事情，肯定就是將事件分發(fā)到正確的事件處理器上進(jìn)行處理，而Initiation Dispatcher就是完成這個(gè)分發(fā)的事情。

Reactor設(shè)計(jì)模式的一個(gè)簡(jiǎn)單類圖，如下所示。

通常，Reactor設(shè)計(jì)模式中的Reactor，可以理解為上述圖中的Synchronous Event Demultiplexer + Initiation Dispatcher。

二. 單Reactor單線程模型

單Reactor單線程模型中，只有一個(gè)Reactor在監(jiān)聽事件和分發(fā)事件，并且監(jiān)聽事件，分發(fā)事件和處理事件都在一個(gè)線程中完成。示意圖如下所示。

上述示意圖中，一次完整的處理流程可以概括如下。

Reactor監(jiān)聽到ACCEPT事件發(fā)生，表示此時(shí)有客戶端建立連接；
Reactor將ACCEPT事件分發(fā)給Acceptor處理；
Acceptor會(huì)在服務(wù)端創(chuàng)建與客戶端通信的client-socket管道，然后注冊(cè)到IO多路復(fù)用器selector上，并監(jiān)聽READ事件；
Reactor監(jiān)聽到READ事件發(fā)生，表示此時(shí)客戶端數(shù)據(jù)可讀；
Reactor將ACCEPT事件分發(fā)給Handler處理，Handler處理READ事件就會(huì)基于client-socket管道完成客戶端數(shù)據(jù)的讀取。

下面將基于Java語言，實(shí)現(xiàn)一個(gè)簡(jiǎn)單的單Reactor單線程模型的服務(wù)端，整體代碼實(shí)現(xiàn)完全符合上述示意圖，大家可以進(jìn)行參照閱讀。

首先實(shí)現(xiàn)Reactor，如下所示。

public class Reactor implements Runnable {
    private final Selector selector;
    public Reactor(int port) throws IOException {
        // 開啟多路復(fù)用
        selector = Selector.open();
        // 服務(wù)端創(chuàng)建listen-socket管道
        ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
        // 綁定端口
        listenSocketChannel.socket().bind(new InetSocketAddress(port));
        // 設(shè)置為非阻塞模式
        listenSocketChannel.configureBlocking(false);
        // ACCEPT事件的附加器是Acceptor
        listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT,
                new Acceptor(selector, listenSocketChannel));
    }
    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                // 獲取發(fā)生的事件
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterable = selectionKeys.iterator();
                while (iterable.hasNext()) {
                    // 對(duì)事件進(jìn)行分發(fā)
                    dispatch(iterable.next());
                    iterable.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            LockSupport.parkNanos(1000 * 1000 * 1000);
        }
    }
    private void dispatch(SelectionKey selectionKey) {
        // 獲取事件的附加器
        // ACCEPT事件的附加器是Acceptor，故由Acceptor來處理ACCEPT事件
        // READ事件的附加器是Handler，故由Handler來處理READ事件
        Runnable attachment = (Runnable) selectionKey.attachment();
        if (attachment != null) {
            attachment.run();
        }
    }
}

已知Reactor會(huì)監(jiān)聽客戶端連接的ACCEPT事件，還已知ACCEPT事件由Acceptor處理，所以在向多路復(fù)用器注冊(cè)服務(wù)端用于監(jiān)聽客戶端連接的listen-socket管道時(shí)，添加了一個(gè)Acceptor作為附加器，那么當(dāng)發(fā)生ACCEPT事件時(shí)，就能夠獲取到作為ACCEPT事件附加器的Acceptor來處理ACCEPT事件。

下面看一下Acceptor的實(shí)現(xiàn)，如下所示。

public class Acceptor implements Runnable {
    private final Selector selector;
    private final ServerSocketChannel listenSocketChannel;
    public Acceptor(Selector selector, ServerSocketChannel listenSocketChannel) {
        this.selector = selector;
        this.listenSocketChannel = listenSocketChannel;
    }
    @Override
    public void run() {
        try {
            // 為連接的客戶端創(chuàng)建client-socket管道
            SocketChannel clientSocketChannel = listenSocketChannel.accept();
            // 設(shè)置為非阻塞
            clientSocketChannel.configureBlocking(false);
            // READ事件的附加器是Handler
            clientSocketChannel.register(selector, SelectionKey.OP_READ,
                    new Handler(clientSocketChannel));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

在Acceptor中就是在服務(wù)端創(chuàng)建與客戶端通信的client-socket管道，然后注冊(cè)到多路復(fù)用器上并指定監(jiān)聽READ事件，同時(shí)又因?yàn)镽EAD事件由Handler處理，所以還添加了一個(gè)Handler作為附加器，當(dāng)READ事件發(fā)生時(shí)可以獲取到作為READ事件附加器的Handler來處理READ事件。

下面看一下Handler的實(shí)現(xiàn)，如下所示。

public class Handler implements Runnable {
    private final SocketChannel clientSocketChannel;
    public Handler(SocketChannel clientSocketChannel) {
        this.clientSocketChannel = clientSocketChannel;
    }
    @Override
    public void run() {
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        try {
            // 讀取數(shù)據(jù)
            int read = clientSocketChannel.read(byteBuffer);
            if (read <= 0) {
                clientSocketChannel.close();
            } else {
                System.out.println(new String(byteBuffer.array()));
            }
        } catch (IOException e1) {
            try {
                clientSocketChannel.close();
            } catch (IOException e2) {
                e2.printStackTrace();
            }
            e1.printStackTrace();
        }
    }
}

在Handler中就是簡(jiǎn)單的讀取數(shù)據(jù)并打印，當(dāng)讀取數(shù)據(jù)為空或者發(fā)生異常時(shí)，需要及時(shí)將管道關(guān)閉。

最后編寫一個(gè)主程序?qū)eactor運(yùn)行起來，如下所示。

public class MainServer {
    public static void main(String[] args) throws IOException {
        Thread reactorThread = new Thread(new Reactor(8080));
        reactorThread.start();
    }
}

現(xiàn)在來思考一下，單Reactor單線程模型有什么優(yōu)點(diǎn)和缺點(diǎn)。優(yōu)點(diǎn)其實(shí)就是模型簡(jiǎn)單，實(shí)現(xiàn)方便。缺點(diǎn)有兩點(diǎn)，如下所示。

一個(gè)Reactor同時(shí)負(fù)責(zé)監(jiān)聽ACCEPT事件和READ事件；
只有一個(gè)線程在工作，處理效率低，無法利用多核CPU的優(yōu)勢(shì)。

但是盡管單Reactor單線程模型有上述的缺點(diǎn)，但是著名的緩存中間件Redis的服務(wù)端，就是使用的單Reactor單線程模型，示意圖如下。

那為什么以性能著稱的Redis會(huì)采取單Reactor單線程模型呢，其實(shí)就是因?yàn)镽edis的操作都在內(nèi)存中，讀寫都非常快速，所以單Reactor單線程模型也能運(yùn)行得很流暢，同時(shí)還避免了多線程下的各種并發(fā)問題。

三. 單Reactor多線程模型

在理解了單Reactor單線程模型后，那么肯定就能想到，假如在Handler中處理READ事件的這個(gè)事情能夠使用一個(gè)線程池來完成，從而就可以實(shí)現(xiàn)READ事件的處理不會(huì)阻塞主線程。而這樣的一個(gè)模型，其實(shí)就是單Reactor多線程模型，示意圖如下所示。

和單Reactor單線程模型唯一的不同，就是在Handler中多了一個(gè)線程池。

單Reactor多線程模型的代碼實(shí)現(xiàn)，除了Handler以外，其余和單Reactor單線程模型一摸一樣，所以下面就看一下單Reactor多線程模型中的Handler實(shí)現(xiàn)，如下所示。

public class Handler implements Runnable {
    private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(16, 32,
            60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
    private final SocketChannel clientSocketChannel;
    public Handler(SocketChannel clientSocketChannel) {
        this.clientSocketChannel = clientSocketChannel;
    }
    @Override
    public void run() {
        threadPool.execute(() -> {
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            try {
                // 讀取數(shù)據(jù)
                int read = clientSocketChannel.read(byteBuffer);
                if (read <= 0) {
                    clientSocketChannel.close();
                } else {
                    System.out.println(new String(byteBuffer.array()));
                }
                // 睡眠10S，演示任務(wù)執(zhí)行耗時(shí)長(zhǎng)也不會(huì)阻塞處理其它客戶端請(qǐng)求
                LockSupport.parkNanos(1000 * 1000 * 1000 * 10L);
            } catch (IOException e1) {
                try {
                    clientSocketChannel.close();
                } catch (IOException e2) {
                    e2.printStackTrace();
                }
                e1.printStackTrace();
            }
        });
    }
}

其實(shí)就是每一個(gè)READ事件的處理會(huì)作為一個(gè)任務(wù)被扔到線程池中去處理。

單Reactor多線程模型雖然解決了只有一個(gè)線程的問題，但是可以發(fā)現(xiàn)，仍舊是只有一個(gè)Reactor在同時(shí)監(jiān)聽ACCEPT事件和READ事件。

那么現(xiàn)在思考一下，為什么一個(gè)Reactor同時(shí)監(jiān)聽ACCEPT事件和READ事件是不好的。其實(shí)就是因?yàn)橥ǔ？蛻舳诉B接的建立是不頻繁的，但是連接建立后數(shù)據(jù)的收發(fā)是頻繁的，所以如果能夠?qū)⒈O(jiān)聽READ事件這個(gè)動(dòng)作拆分出來，讓多個(gè)子Reactor來監(jiān)聽READ事件，而原來的主Reactor只監(jiān)聽ACCEPT事件，那么整體的效率，會(huì)進(jìn)一步提升，而這，就是主從Reactor多線程模型。

四. 主從Reactor多線程模型

主從Reactor模型中，有一個(gè)主Reactor，專門監(jiān)聽ACCEPT事件，然后有多個(gè)從Reactor，專門監(jiān)聽READ事件，示意圖如下所示。

上述示意圖中，一次完整的處理流程可以概括如下。

主Reactor監(jiān)聽到ACCEPT事件發(fā)生，表示此時(shí)有客戶端建立連接；
主Reactor將ACCEPT事件分發(fā)給Acceptor處理；
Acceptor會(huì)在服務(wù)端創(chuàng)建與客戶端通信的client-socket管道，然后注冊(cè)到從Reactor的IO多路復(fù)用器selector上，并監(jiān)聽READ事件；
從Reactor監(jiān)聽到READ事件發(fā)生，表示此時(shí)客戶端數(shù)據(jù)可讀；
從Reactor將ACCEPT事件分發(fā)給Handler處理，Handler處理READ事件就會(huì)基于client-socket管道完成客戶端數(shù)據(jù)的讀取。

下面將基于Java語言，實(shí)現(xiàn)一個(gè)簡(jiǎn)單的主從Reactor多線程模型的服務(wù)端，整體代碼實(shí)現(xiàn)完全符合上述示意圖，大家可以進(jìn)行參照閱讀。

首先是主Reactor的實(shí)現(xiàn)，如下所示。

public class MainReactor implements Runnable {
    private final Selector selector;
    public MainReactor(int port) throws IOException {
        // 開多路復(fù)用器
        selector = Selector.open();
        // 服務(wù)端創(chuàng)建listen-socket管道
        ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
        // 設(shè)置為非阻塞
        listenSocketChannel.configureBlocking(false);
        // 綁定監(jiān)聽端口
        listenSocketChannel.socket().bind(new InetSocketAddress(port));
        // 將listen-socket管道綁定到主Reactor的多路復(fù)用器上
        // 并且主Reactor上只會(huì)注冊(cè)listen-socket管道，用于監(jiān)聽ACCEPT事件
        listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT,
                new Acceptor(listenSocketChannel));
    }
    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterable = selectionKeys.iterator();
                while (iterable.hasNext()) {
                    // 對(duì)事件進(jìn)行分發(fā)
                    dispatch(iterable.next());
                    iterable.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            LockSupport.parkNanos(1000 * 1000 * 1000);
        }
    }
    private void dispatch(SelectionKey selectionKey) {
        // 獲取事件附加器，只會(huì)是Acceptor
        Runnable attachment = (Runnable) selectionKey.attachment();
        if (attachment != null) {
            attachment.run();
        }
    }
}

主Reactor的實(shí)現(xiàn)中，還是先創(chuàng)建服務(wù)端監(jiān)聽客戶端連接的listen-socket管道，然后注冊(cè)到主Reactor的IO多路復(fù)用器上，并監(jiān)聽ACCEPT事件，同時(shí)我們現(xiàn)在知道，主Reactor的IO多路復(fù)用器上只會(huì)注冊(cè)listen-socket管道且只會(huì)監(jiān)聽ACCEPT事件。同樣，也添加了一個(gè)Acceptor作為附加器，那么當(dāng)發(fā)生ACCEPT事件時(shí)，就能夠獲取到作為ACCEPT事件附加器的Acceptor來處理ACCEPT事件。

下面是Acceptor的實(shí)現(xiàn)，如下所示。

public class Acceptor implements Runnable {
    // 指定從Reactor一共有16個(gè)
    private static final int TOTAL_SUBREACTOR_NUM = 16;
    // 服務(wù)端的listen-socket管道
    private final ServerSocketChannel listenSocketChannel;
    // 用于運(yùn)行從Reactor
    private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            TOTAL_SUBREACTOR_NUM, TOTAL_SUBREACTOR_NUM * 2,
            60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
    // 從Reactor集合
    private final List<SubReactor> subReactors = new ArrayList<>(TOTAL_SUBREACTOR_NUM);
    public Acceptor(ServerSocketChannel listenSocketChannel) throws IOException {
        this.listenSocketChannel = listenSocketChannel;
        // 將從Reactor初始化出來并運(yùn)行
        for (int i = 0; i < TOTAL_SUBREACTOR_NUM; i++) {
            SubReactor subReactor = new SubReactor(Selector.open());
            subReactors.add(subReactor);
            threadPool.execute(subReactor);
        }
    }
    @Override
    public void run() {
        try {
            // 為連接的客戶端創(chuàng)建client-socket管道
            SocketChannel clientSocketChannel = listenSocketChannel.accept();
            // 設(shè)置為非阻塞
            clientSocketChannel.configureBlocking(false);
            // 任意選擇一個(gè)從Reactor，讓其監(jiān)聽連接的客戶端的READ事件
            Optional<SubReactor> anySubReactor = subReactors.stream().findAny();
            if (anySubReactor.isPresent()) {
                SubReactor subReactor = anySubReactor.get();
                // 從Reactor的多路復(fù)用器會(huì)阻塞在select()方法上
                // 這里需要先喚醒多路復(fù)用器，立即從select()方法返回
                subReactor.getSelector().wakeup();
                // 讓從Reactor負(fù)責(zé)處理客戶端的READ事件
                clientSocketChannel.register(subReactor.getSelector(), SelectionKey.OP_READ,
                        new Handler(clientSocketChannel));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

首先在Acceptor的構(gòu)造函數(shù)中，會(huì)將所有從Reactor初始化出來，并且每一個(gè)從Reactor都會(huì)持有一個(gè)IO多路復(fù)用器。當(dāng)一個(gè)從Reactor創(chuàng)建出來后就會(huì)立即運(yùn)行，此時(shí)從Reactor的IO多路復(fù)用器就會(huì)開始監(jiān)聽，即阻塞在select() 方法上。

然后在Acceptor的主體邏輯中，會(huì)為連接的客戶端創(chuàng)建client-socket管道，然后從所有從Reactor中基于某種策略（隨機(jī)）選擇一個(gè)從Reactor，并將client-socket管道注冊(cè)在選擇的從Reactor的IO多路復(fù)用器上，有一點(diǎn)需要注意，此時(shí)從Reactor的IO多路復(fù)用器可能會(huì)阻塞在select() 方法上，所以注冊(cè)前需要先通過wakeup() 方法進(jìn)行喚醒。

接下來繼續(xù)看從Reactor的實(shí)現(xiàn)，如下所示。

public class SubReactor implements Runnable {
    private final Selector selector;
    public SubReactor(Selector selector) {
        this.selector = selector;
    }
    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    // 對(duì)事件進(jìn)行分發(fā)
                    dispatch(iterator.next());
                    iterator.remove();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
            LockSupport.parkNanos(1000 * 1000 * 1000);
        }
    }
    private void dispatch(SelectionKey selectionKey) {
        // 獲取事件附加器，只會(huì)是Handler
        Runnable runnable = (Runnable) selectionKey.attachment();
        if (runnable != null) {
            runnable.run();
        }
    }
    public Selector getSelector() {
        return selector;
    }
}

從Reactor的實(shí)現(xiàn)中，會(huì)監(jiān)聽服務(wù)端為連接的客戶端創(chuàng)建的client-socket管道上的READ事件，一旦有READ事件發(fā)生，就會(huì)使用作為附加器的Handler來處理READ事件。同樣，從Reactor的IO多路復(fù)用器上只會(huì)注冊(cè)client-socket管道且只會(huì)監(jiān)聽READ事件。

然后是Handler，因?yàn)槭嵌嗑€程模型，所以其實(shí)現(xiàn)和第三節(jié)中的Handler完全一樣，下面再貼一下代碼。

public class Handler implements Runnable {
    private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(16, 32,
            60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
    private final SocketChannel clientSocketChannel;
    public Handler(SocketChannel clientSocketChannel) {
        this.clientSocketChannel = clientSocketChannel;
    }
    @Override
    public void run() {
        threadPool.execute(() -> {
            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
            try {
                // 讀取數(shù)據(jù)
                int read = clientSocketChannel.read(byteBuffer);
                if (read <= 0) {
                    clientSocketChannel.close();
                } else {
                    System.out.println(new String(byteBuffer.array()));
                }
                // 睡眠10S，演示任務(wù)執(zhí)行耗時(shí)長(zhǎng)也不會(huì)阻塞處理其它客戶端請(qǐng)求
                LockSupport.parkNanos(1000 * 1000 * 1000 * 10L);
            } catch (IOException e1) {
                try {
                    clientSocketChannel.close();
                } catch (IOException e2) {
                    e2.printStackTrace();
                }
                e1.printStackTrace();
            }
        });
    }
}

最后編寫一個(gè)主程序?qū)⒅鱎eactor運(yùn)行起來，如下所示。

public class MainServer {
    public static void main(String[] args) throws IOException {
        Thread mainReactorThread = new Thread(new MainReactor(8080));
        mainReactorThread.start();
    }
}