目錄

• # 摘要
• 高性能服務器
• Reactor模式
  • Reactor單線程模型設計
  • Reactor多線程模型設計
  • 主從Reactor多線程模型設計
  • Netty Reactor模型設計
• 參考
• 你的鼓勵也是我創作的動力

• Posted by 微博@Yangsc_o
• 原創文章，版權聲明：自由轉載-非商用-非衍生-保持署名 | Creative Commons BY-NC-ND 3.0

# 摘要

在前兩篇《快速理解Linux網絡I_O》、《java的I_O模型-BIO&NIO&AIO》兩邊中介紹了Linux下的I/O模型和java中的I/O模型，今天我們介紹Reactor模型，並探究Netty的實現

高性能服務器

在互聯網時代，我們使用的軟件基本上全是C/S架構，C/S架構的軟件一個明顯的好處就是：只要有網絡，你可以在任何地方干同一件事。C/S架構可以抽象為如下模型：

• C就是Client(客戶端),上面的B是Browser(瀏覽器)
• S就是Server(服務器)：服務器管理某種資源，並且通過操作這種資源來為它的客戶端提供某種服務

那服務器如何能快速的處理用戶的請求呢？在我看來高性能服務器至少要滿足如下幾個需求：

• 效率高：既然是高性能，那處理客戶端請求的效率當然要很高了
• 高可用：不能隨便就掛掉了
• 編程簡單：基於此服務器進行業務開發需要足夠簡單
• 可擴展：可方便的擴展功能
• 可伸縮：可簡單的通過部署的方式進行容量的伸縮，也就是服務需要無狀態

而滿足如上需求的一個基礎就是高性能的IO!

Reactor模式

什麼是Reactor模式？

兩種I/O多路復用模式：Reactor和Proactor，兩個與事件分離器有關的模式是Reactor和Proactor。Reactor模式採用同步IO，而Proactor採用異步IO。

在Reactor中，事件分離器負責等待文件描述符或socket為讀寫操作準備就緒，然後將就緒事件傳遞給對應的處理器，最後由處理器負責完成實際的讀寫工作。

在Proactor模式中，處理器–或者兼任處理器的事件分離器，只負責發起異步讀寫操作。IO操作本身由操作系統來完成。傳遞給操作系統的參數需要包括用戶定義的數據緩衝區地址和數據大小，操作系統才能從中得到寫出操作所需數據，或寫入從socket讀到的數據。事件分離器捕獲IO操作完成事件，然後將事件傳遞給對應處理器。

說人話的方式理解：

• reactor：能收了你跟俺說一聲。
• proactor: 你給我收十個字節，收好了跟俺說一聲。

Doug Lea是這樣類比的

• Reactor通過調度適當的處理程序來響應IO事件；
• 處理程序執行非阻塞操作
• 通過將處理程序綁定到事件來管理；

Reactor單線程模型設計

單線程版本Java NIO的支持：

• Channels：與支持非阻塞讀取的文件，套接字等的連接

• Buffers：類似於數組的對象，可由Channels直接讀取或寫入

• Selectors：通知一組通道中哪一個有IO事件

• SelectionKeys：維護IO事件狀態和綁定

• Reactor 代碼如下

public class Reactor implements Runnable {
    final Selector selector;
    final ServerSocketChannel serverSocketChannel;

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

    @Override
    public void run() {
        while (!Thread.interrupted()) {
            try {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                for (SelectionKey selectionKey : selectionKeys) {
                    dispatch(selectionKey);
                }
                selectionKeys.clear();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

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

    class Acceptor implements Runnable {
        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocketChannel.accept();
                if (channel != null) {
                    new Handler(selector, channel);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        new Thread(
                new Reactor(1234)
        ).start();
    }

}

• Handler代碼如下：

public class Handler implements Runnable{
    private final static int DEFAULT_SIZE = 1024;
    private final SocketChannel socketChannel;
    private final SelectionKey seletionKey;
    private static final int READING = 0;
    private static final int SENDING = 1;
    private int state = READING;

    ByteBuffer inputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);
    ByteBuffer outputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);

    public Handler(Selector selector, SocketChannel channel) throws IOException {
        this.socketChannel = channel;
        socketChannel.configureBlocking(false);
        this.seletionKey = socketChannel.register(selector, 0);
        seletionKey.attach(this);
        seletionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        if (state == READING) {
            read();
        } else if (state == SENDING) {
            write();
        }
    }


    private void write() {
        try {
            socketChannel.write(outputBuffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
        while (outIsComplete()) {
            seletionKey.cancel();
        }
    }

    private void read() {
        try {
            socketChannel.read(inputBuffer);
            if (inputIsComplete()) {
                process();
                System.out.println("接收到來自客戶端（" + socketChannel.socket().getInetAddress().getHostAddress()
                        + "）的消息：" + new String(inputBuffer.array()));
                seletionKey.attach(new Sender());
                seletionKey.interestOps(SelectionKey.OP_WRITE);
                seletionKey.selector().wakeup();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean inputIsComplete() {
        return true;
    }
    public boolean outIsComplete() {
        return true;
    }


    public void process() {
        // do something...
    }

    class Sender implements Runnable {
        @Override
        public void run() {
            try {
                socketChannel.write(outputBuffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (outIsComplete()) {
                seletionKey.cancel();
            }
        }
    }

}

這個模型和上面的NIO流程很類似，只是將消息相關處理獨立到了Handler中去了！雖然說到NIO一個線程就可以支持所有的IO處理。但是瓶頸也是顯而易見的！如果這個客戶端多次進行請求，如果在Handler中的處理速度較慢，那麼後續的客戶端請求都會被積壓，導致響應變慢！所以引入了Reactor多線程模型!

Reactor多線程模型設計

Reactor多線程模型就是將Handler中的IO操作和非IO操作分開，操作IO的線程稱為IO線程，非IO操作的線程稱為工作線程!這樣的話，客戶端的請求會直接被丟到線程池中，客戶端發送請求就不會堵塞！

Reactor保持不變，僅需要改動Handler代碼：

public class Handler implements Runnable{
    private final static int DEFAULT_SIZE = 1024;
    private final SocketChannel socketChannel;
    private final SelectionKey seletionKey;
    private static final int READING = 0;
    private static final int SENDING = 1;
    private int state = READING;

    ByteBuffer inputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);
    ByteBuffer outputBuffer = ByteBuffer.allocate(DEFAULT_SIZE);

    private static ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime()
            .availableProcessors());
    private static final int PROCESSING = 3;
    private Selector selector;


    public Handler(Selector selector, SocketChannel channel) throws IOException {
        this.selector = selector;
        this.socketChannel = channel;
        socketChannel.configureBlocking(false);
        this.seletionKey = socketChannel.register(selector, 0);
        seletionKey.attach(this);
        seletionKey.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        if (state == READING) {
            read();
        } else if (state == SENDING) {
            write();
        }
    }

    private void write() {
        try {
            socketChannel.write(outputBuffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
        while (outIsComplete()) {
            seletionKey.cancel();
        }
    }

    private void read() {
        try {
            socketChannel.read(inputBuffer);
            if (inputIsComplete()) {
                process();
                executorService.execute(new Processer());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean inputIsComplete() {
        return true;
    }
    public boolean outIsComplete() {
        return true;
    }


    public void process() {
        // do something...
    }

    class Sender implements Runnable {
        @Override
        public void run() {
            try {
                socketChannel.write(outputBuffer);
            } catch (IOException e) {
                e.printStackTrace();
            }
            if (outIsComplete()) {
                seletionKey.cancel();
            }
        }
    }

    synchronized void processAndHandOff() {
        process();
        // or rebind attachment
        state = SENDING;
        seletionKey.interestOps(SelectionKey.OP_WRITE);
        selector.wakeup();
    }

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

}

主從Reactor多線程模型設計

主從Reactor多線程模型是將Reactor分成兩部分，mainReactor負責監聽server socket，accept新連接，並將建立的socket分派給subReactor。subReactor負責多路分離已連接的socket，讀寫網絡數據，對業務處理功能，其扔給worker線程池完成。通常，subReactor個數上可與CPU個數等同：

Handler保持不變，僅需要改動Reactor代碼：

public class Reactor {
    // also create threads
    Selector[] selectors;
    AtomicInteger next = new AtomicInteger(0);
    final ServerSocketChannel serverSocketChannel;

    private static ExecutorService sunReactors = Executors.newFixedThreadPool(Runtime.getRuntime()
            .availableProcessors());
    private static final int PROCESSING = 3;

    public Reactor(int port) throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        serverSocketChannel.configureBlocking(false);
        selectors = new Selector[4];
        for (int i = 0; i < selectors.length; i++) {
            Selector selector = selectors[i];
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            serverSocketChannel.configureBlocking(false);
            SelectionKey key = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            key.attach(new Acceptor());
            new Thread(()->{
                while (!Thread.interrupted()) {
                    try {
                        selector.select();
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        for (SelectionKey selectionKey : selectionKeys) {
                            dispatch(selectionKey);
                        }
                        selectionKeys.clear();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }



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

    class Acceptor implements Runnable {
        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocketChannel.accept();
                if (channel != null) {
                    sunReactors.execute(new Handler(selectors[next.getAndIncrement() % selectors.length], channel));
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {
        new Reactor(1234);
    }

}

以上是三種不同的設計思路，接下來看一下Netty這個一個高性能NIO框架，其是如何實現Reactor模型的！

Netty Reactor模型設計

• 看一個最簡單的Netty服務端代碼

public final class EchoServer {
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    public static void main(String[] args) throws Exception {
        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 100)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(serverHandler);
                 }
             });
            ChannelFuture f = b.bind(PORT).sync();
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

• Netty Server Handler

public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.write(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

我們從Netty服務器代碼來看，與Reactor模型進行對應！

• EventLoopGroup就相當於是Reactor，bossGroup對應主Reactor,workerGroup對應從Reactor
• TimeServerHandler就是Handler
• child開頭的方法配置的是客戶端channel，非child開頭的方法配置的是服務端channel

當然Netty的線程模型並不是固定的，它支持Reactor單線程模型、Reactor多線程模型、主從模型，上面的例子是一個主從模型的，下面進行詳細的分析，如圖所示：

服務啟動時，創建了兩個EventLoopGroup，它們實際上是兩個Reactor線程池，一個用於接收TCP連接、一個用於處理I/O相關的讀寫操作、或者執行系統task、定時task等；

• Netty用於接收客戶端請求連接池職責如下：
  • 接收客戶端請求並初始化channel參數；
  • 講鏈路變更事件通知給ChannelPipiline；
• Netty用於處理I/O連接池職責如下：
  • 異步讀取通信對端的數據報，發送讀事件到ChannelPipiline；
  • 異步發送消息對端的數據報，調用ChannelPipiline的消息發送接口；
  • 執行系統調用task；
  • 執行系統定時任務task，例如鏈路空閑狀態檢測定時任務；

參考

Scalable IO in Java

高性能Server—Reactor模型

NIO技術概覽

《Netty 權威指南》第二版 — 李林峰

你的鼓勵也是我創作的動力

打賞地址

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※教你寫出一流的銷售文案?

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※回頭車貨運收費標準

※別再煩惱如何寫文案,掌握八大原則!

※超省錢租車方案

※產品缺大量曝光嗎?你需要的是一流包裝設計!

※推薦台中搬家公司優質服務,可到府估價

能合法上牌嗎。全國牌可以上嗎。答：該款車所以手續都是合法齊全的，絕對可以合法上牌，並且全國各地都可以上牌，而廣州之外的地區經銷商在全國各地都有合作的夥伴會幫助上牌。問：該款車上牌的費用是多少錢。答：全國上牌都是6000元。

說到SUV公路之王，相信很多車迷朋友第一個想到的就是寶馬X5了。那為什麼是寶馬X5呢？同級別的SUV還不少車型，例如保時捷-卡宴、奧迪Q7、奔馳GLE等，而寶馬X5能從中脫穎而出，被譽為“SUV公路之王”，首先寶馬品牌本身流淌着出色的操控性血統就已經起到決定性的作用，再加上動力十足的3.0T直列六缸發動機和xDrive全時四驅系統的搭配，您說寶馬X5的公路表現能不“稱王”嗎!

然而作為SUV公路之王，其售價自然就不低了，中規進口寶馬X5 xDrive 35i指導價是83.80萬元起售，按落地價算起來已接近百萬元的售價了，高昂的價格令不少預算不足的客戶只能退求其次選擇75.80萬元起售的xDrive 28i，但是xDrive 28i搭載的只是245匹馬力的四缸發動機，動力的不足和直列六缸的缺失實在是令人糾結不已。

為了滿足車友們的需求，令更多的車友擁有這款SUV公路之王，再次出馬，聯合廣州當地知名豪車經銷商推出平行進口中東版寶馬X5 35i，特惠售價為74.80萬元。該款中東版寶馬X5 35i配置對應的是售價為95.80萬元的中規版35i豪華型，按目前廣州當地中規版寶馬X5優惠8個點來計算，中規版寶馬X5 35i豪華型優惠後車價是88.13萬元，就算是中規版35i最低配典雅型83.80萬元優惠8個點后也要77.09萬元，對比之下這次推出的中東版寶馬X5 35i是十分優惠的。

另外很重要的一點是該款中東版寶馬X5是原廠帶有中文菜單和導航的，而更重要的是該款中東版寶馬X5是現車銷售的，在現在年底購車高峰期的節點上，有現車可能比起優惠來說更為實在，因為據了解到的情況中規版寶馬X5現車是很少了，訂車也趕不上明年春節前提車了，所以這次推出的中東版寶馬X5在明年春節前完成提車手續絕對沒有問題，可以說是現車與優惠並存，有興趣的朋友可以致電購車熱線諮詢：400-855-5813

下面該款中東版寶馬X5 35i的具體情況以問答的方式來進行說明

問：這款平行進口2016款中東版寶馬X5 35i有什麼顏色？

答：該款車有黑色、棕色、白色三種外觀，內飾均為棕色，其中黑色和棕色外觀現車較多，白色現車較少。

問：這款平行進口2016款中東版寶馬X5 35i只能在廣州提車嗎？

答：該款車不需要一定在廣州提車，因為是平行進口車沒有區域限制，是可以售全國的，如果跟經銷商協商預先付了全款，長江以南地區是可以免費發車的，其他地區按運輸強度收取運費發車。

問：這款平行進口2016款中東版寶馬X5 35i手續齊全嗎？能合法上牌嗎？全國牌可以上嗎？

答：該款車所以手續都是合法齊全的，絕對可以合法上牌，並且全國各地都可以上牌，而廣州之外的地區經銷商在全國各地都有合作的夥伴會幫助上牌。

問：該款車上牌的費用是多少錢？

答：全國上牌都是6000元。

問：該款車的保險費用是多少錢？

答：保險費用是27000元左右，需要在經銷商處購買，根據選擇險種的不同，保險費用會相應變動。

問：該款車的購置稅是多少錢？算上所以費用落地價是多少錢？

答：按74.80萬元車價算，購置稅約是63932元左右，算上所有費用落地價約是84.49萬元左右。

問：該款車是否享有與中規進口寶馬X5一樣的保修？能在4S店保養嗎？

答：因為是平行進口車的關係，該款車是不能享有中規進口車在4S店的保修，不過該車可以由售車經銷商提供保修，是店保兩年不限公里數，並且售車經銷商在廣州已有經營德國名車維修保養十多年的實力，不論是保修還是保養都無需擔心。另外車主如果不想在售車經銷商處保修保養，是可以付費到全國各地的寶馬4S店進行維修和保養。

問：如果在購車平台上下單？

答：點擊“閱讀原文”直接進入下單界面，訂金為1000元，可支付寶或微信支付，訂金直接抵扣車款，則實際銷售合同是74.70萬元；下單后不購車訂金隨時可退，成功下單後會有工作人員與您聯繫安排購車事宜，請保持電話暢通。

注：有任何疑問可致電購車熱線諮詢：400-855-5813本站聲明:網站內容來源於http://www.auto6s.com/,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※教你寫出一流的銷售文案?

※廣告預算用在刀口上，台北網頁設計公司幫您達到更多曝光效益

※回頭車貨運收費標準

※別再煩惱如何寫文案,掌握八大原則!

※超省錢租車方案

※產品缺大量曝光嗎?你需要的是一流包裝設計!

※推薦台中搬家公司優質服務,可到府估價