一、基礎簡介

在IO流的網(wǎng)絡模型中，以常見的「客戶端-服務端」交互場景為例；

客戶端與服務端進行通信「交互」，可能是同步或者異步，服務端進行「流」處理時，可能是阻塞或者非阻塞模式，當然也有自定義的業(yè)務流程需要執(zhí)行，從處理邏輯看就是「讀取數(shù)據(jù)-業(yè)務執(zhí)行-應答寫數(shù)據(jù)」的形式；

(資料圖)

Java提供「三種」IO網(wǎng)絡編程模型，即：「BIO同步阻塞」、「NIO同步非阻塞」、「AIO異步非阻塞」；

二、同步阻塞1、模型圖解

BIO即同步阻塞，服務端收到客戶端的請求時，會啟動一個線程處理，「交互」會阻塞直到整個流程結(jié)束；

這種模式如果在高并發(fā)且流程復雜耗時的場景下，客戶端的請求響應會存在嚴重的性能問題，并且占用過多資源；

2、參考案例

【服務端】啟動ServerSocket接收客戶端的請求，經(jīng)過一系列邏輯之后，向客戶端發(fā)送消息，注意這里線程的10秒休眠；

public class SocketServer01 {    public static void main(String[] args) throws Exception {        // 1、創(chuàng)建Socket服務端        ServerSocket serverSocket = new ServerSocket(8080);        // 2、方法阻塞等待，直到有客戶端連接        Socket socket = serverSocket.accept();        // 3、輸入流，輸出流        InputStream inStream = socket.getInputStream();        OutputStream outStream = socket.getOutputStream();        // 4、數(shù)據(jù)接收和響應        int readLen = 0;        byte[] buf = new byte[1024];        if ((readLen=inStream.read(buf)) != -1){            // 接收數(shù)據(jù)            String readVar = new String(buf, 0, readLen) ;            System.out.println("readVar======="+readVar);        }        // 響應數(shù)據(jù)        Thread.sleep(10000);        outStream.write("sever-8080-write;".getBytes());        // 5、資源關閉        IoClose.ioClose(outStream,inStream,socket,serverSocket);    }}

【客戶端】Socket連接，先向ServerSocket發(fā)送請求，再接收其響應，由于Server端模擬耗時，Client處于長時間阻塞狀態(tài)；

public class SocketClient01 {    public static void main(String[] args) throws Exception {        // 1、創(chuàng)建Socket客戶端        Socket socket = new Socket(InetAddress.getLocalHost(), 8080);        // 2、輸入流，輸出流        OutputStream outStream = socket.getOutputStream();        InputStream inStream = socket.getInputStream();        // 3、數(shù)據(jù)發(fā)送和響應接收        // 發(fā)送數(shù)據(jù)        outStream.write("client-hello".getBytes());        // 接收數(shù)據(jù)        int readLen = 0;        byte[] buf = new byte[1024];        if ((readLen=inStream.read(buf)) != -1){            String readVar = new String(buf, 0, readLen) ;            System.out.println("readVar======="+readVar);        }        // 4、資源關閉        IoClose.ioClose(inStream,outStream,socket);    }}

三、同步非阻塞1、模型圖解

NIO即同步非阻塞，服務端可以實現(xiàn)一個線程，處理多個客戶端請求連接，服務端的并發(fā)能力得到極大的提升；

這種模式下客戶端的請求連接都會注冊到Selector多路復用器上，多路復用器會進行輪詢，對請求連接的IO流進行處理；

2、參考案例

【服務端】單線程可以處理多個客戶端請求，通過輪詢多路復用器查看是否有IO請求；

public class SocketServer01 {    public static void main(String[] args) throws Exception {        try {            //啟動服務開啟監(jiān)聽            ServerSocketChannel socketChannel = ServerSocketChannel.open();            socketChannel.socket().bind(new InetSocketAddress("127.0.0.1", 8989));            // 設置非阻塞，接受客戶端            socketChannel.configureBlocking(false);            // 打開多路復用器            Selector selector = Selector.open();            // 服務端Socket注冊到多路復用器，指定興趣事件            socketChannel.register(selector, SelectionKey.OP_ACCEPT);            // 多路復用器輪詢            ByteBuffer buffer = ByteBuffer.allocateDirect(1024);            while (selector.select() > 0){                Set selectionKeys = selector.selectedKeys();                Iterator selectionKeyIter = selectionKeys.iterator();                while (selectionKeyIter.hasNext()){                    SelectionKey selectionKey = selectionKeyIter.next() ;                    selectionKeyIter.remove();                    if(selectionKey.isAcceptable()) {                        // 接受新的連接                        SocketChannel client = socketChannel.accept();                        // 設置讀非阻塞                        client.configureBlocking(false);                        // 注冊到多路復用器                        client.register(selector, SelectionKey.OP_READ);                    } else if (selectionKey.isReadable()) {                        // 通道可讀                        SocketChannel client = (SocketChannel) selectionKey.channel();                        int len = client.read(buffer);                        if (len > 0){                            buffer.flip();                            byte[] readArr = new byte[buffer.limit()];                            buffer.get(readArr);                            System.out.println(client.socket().getPort() + "端口數(shù)據(jù):" + new String(readArr));                            buffer.clear();                        }                    }                }            }        } catch (Exception e) {            e.printStackTrace();        }    }}

【客戶端】每隔3秒持續(xù)的向通道內(nèi)寫數(shù)據(jù)，服務端通過輪詢多路復用器，持續(xù)的讀取數(shù)據(jù)；

public class SocketClient01 {    public static void main(String[] args) throws Exception {        try {            // 連接服務端            SocketChannel socketChannel = SocketChannel.open();            socketChannel.connect(new InetSocketAddress("127.0.0.1", 8989));            ByteBuffer writeBuffer = ByteBuffer.allocate(1024);            String conVar = "client-hello";            writeBuffer.put(conVar.getBytes());            writeBuffer.flip();            // 每隔3S發(fā)送一次數(shù)據(jù)            while (true) {                Thread.sleep(3000);                writeBuffer.rewind();                socketChannel.write(writeBuffer);                writeBuffer.clear();            }        } catch (Exception e) {            e.printStackTrace();        }    }}

四、異步非阻塞1、模型圖解

AIO即異步非阻塞，對于通道內(nèi)數(shù)據(jù)的「讀」和「寫」動作，都是采用異步的模式，對于性能的提升是巨大的；

這與常規(guī)的第三方對接模式很相似，本地服務在請求第三方服務時，請求過程耗時很大，會異步執(zhí)行，第三方第一次回調(diào)，確認請求可以被執(zhí)行；第二次回調(diào)則是推送處理結(jié)果，這種思想在處理復雜問題時，可以很大程度的提高性能，節(jié)省資源：

2、參考案例

【服務端】各種「accept」、「read」、「write」動作是異步，通過Future來獲取計算的結(jié)果；

public class SocketServer01 {    public static void main(String[] args) throws Exception {        // 啟動服務開啟監(jiān)聽        AsynchronousServerSocketChannel socketChannel = AsynchronousServerSocketChannel.open() ;        socketChannel.bind(new InetSocketAddress("127.0.0.1", 8989));        // 指定30秒內(nèi)獲取客戶端連接，否則超時        Future acceptFuture = socketChannel.accept();        AsynchronousSocketChannel asyChannel = acceptFuture.get(30, TimeUnit.SECONDS);        if (asyChannel != null && asyChannel.isOpen()){            // 讀數(shù)據(jù)            ByteBuffer inBuffer = ByteBuffer.allocate(1024);            Future readResult = asyChannel.read(inBuffer);            readResult.get();            System.out.println("read："+new String(inBuffer.array()));            // 寫數(shù)據(jù)            inBuffer.flip();            Future writeResult = asyChannel.write(ByteBuffer.wrap("server-hello".getBytes()));            writeResult.get();        }        // 關閉資源        asyChannel.close();    }}

【客戶端】相關「connect」、「read」、「write」方法調(diào)用是異步的，通過Future來獲取計算的結(jié)果；

public class SocketClient01 {    public static void main(String[] args) throws Exception {        // 連接服務端        AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();        Future result = socketChannel.connect(new InetSocketAddress("127.0.0.1", 8989));        result.get();        // 寫數(shù)據(jù)        String conVar = "client-hello";        ByteBuffer reqBuffer = ByteBuffer.wrap(conVar.getBytes());        Future writeFuture = socketChannel.write(reqBuffer);        writeFuture.get();        // 讀數(shù)據(jù)        ByteBuffer inBuffer = ByteBuffer.allocate(1024);        Future readFuture = socketChannel.read(inBuffer);        readFuture.get();        System.out.println("read："+new String(inBuffer.array()));        // 關閉資源        socketChannel.close();    }}

五、Reactor模型1、模型圖解

這部分內(nèi)容，可以參考「Doug Lea的《IO》」文檔，查看更多細節(jié)；

1.1 Reactor設計原理

Reactor模式基于事件驅(qū)動設計，也稱為「反應器」模式或者「分發(fā)者」模式；服務端收到多個客戶端請求后，會將請求分派給對應的線程處理；

Reactor：負責事件的監(jiān)聽和分發(fā)；Handler：負責處理事件，核心邏輯「read讀」、「decode解碼」、「compute業(yè)務計算」、「encode編碼」、「send應答數(shù)據(jù)」；

1.2 單Reactor單線程

【1】Reactor線程通過select監(jiān)聽客戶端的請求事件，收到事件后通過Dispatch進行分發(fā)；

【2】如果是建立連接請求事件，Acceptor通過「accept」方法獲取連接，并創(chuàng)建一個Handler對象來處理后續(xù)業(yè)務；

【3】如果不是連接請求事件，則Reactor會將該事件交由當前連接的Handler來處理；

【4】在Handler中，會完成相應的業(yè)務流程；

這種模式將所有邏輯「連接、讀寫、業(yè)務」放在一個線程中處理，避免多線程的通信，資源競爭等問題，但是存在明顯的并發(fā)和性能問題；

1.3 單Reactor多線程

【1】Reactor線程通過select監(jiān)聽客戶端的請求事件，收到事件后通過Dispatch進行分發(fā)；

【2】如果是建立連接請求事件，Acceptor通過「accept」方法獲取連接，并創(chuàng)建一個Handler對象來處理后續(xù)業(yè)務；

【3】如果不是連接請求事件，則Reactor會將該事件交由當前連接的Handler來處理；

【4】在Handler中，只負責事件響應不處理具體業(yè)務，將數(shù)據(jù)發(fā)送給Worker線程池來處理；

【5】Worker線程池會分配具體的線程來處理業(yè)務，最后把結(jié)果返回給Handler做響應；

這種模式將業(yè)務從Reactor單線程分離處理，可以讓其更專注于事件的分發(fā)和調(diào)度，Handler使用多線程也充分的利用cpu的處理能力，導致邏輯變的更加復雜，Reactor單線程依舊存在高并發(fā)的性能問題；

1.4 主從Reactor多線程

【1】 MainReactor主線程通過select監(jiān)聽客戶端的請求事件，收到事件后通過Dispatch進行分發(fā)；

【2】如果是建立連接請求事件，Acceptor通過「accept」方法獲取連接，之后MainReactor將連接分配給SubReactor；

【3】如果不是連接請求事件，則MainReactor將連接分配給SubReactor，SubReactor調(diào)用當前連接的Handler來處理；

【4】在Handler中，只負責事件響應不處理具體業(yè)務，將數(shù)據(jù)發(fā)送給Worker線程池來處理；

【5】Worker線程池會分配具體的線程來處理業(yè)務，最后把結(jié)果返回給Handler做響應；

這種模式Reactor線程分工明確，MainReactor負責接收新的請求連接，SubReactor負責后續(xù)的交互業(yè)務，適應于高并發(fā)的處理場景，是Netty組件通信框架的所采用的模式；

2、參考案例

【服務端】提供兩個EventLoopGroup，「ParentGroup」主要是用來接收客戶端的請求連接，真正的處理是轉(zhuǎn)交給「ChildGroup」執(zhí)行，即Reactor多線程模型；

@Slf4jpublic class NettyServer {    public static void main(String[] args) {        // EventLoop組，處理事件和IO        EventLoopGroup parentGroup = new NioEventLoopGroup();        EventLoopGroup childGroup = new NioEventLoopGroup();        try {            // 服務端啟動引導類            ServerBootstrap serverBootstrap = new ServerBootstrap();            serverBootstrap.group(parentGroup, childGroup)                    .channel(NioServerSocketChannel.class).childHandler(new ServerChannelInit());            // 異步IO的結(jié)果            ChannelFuture channelFuture = serverBootstrap.bind(8989).sync();            channelFuture.channel().closeFuture().sync();        } catch (Exception e){            e.printStackTrace();        } finally {            parentGroup.shutdownGracefully();            childGroup.shutdownGracefully();        }    }}class ServerChannelInit extends ChannelInitializer {    @Override    protected void initChannel(SocketChannel socketChannel) {        // 獲取管道        ChannelPipeline pipeline = socketChannel.pipeline();        // 編碼、解碼器        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));        // 添加自定義的handler        pipeline.addLast("serverHandler", new ServerHandler());    }}class ServerHandler extends ChannelInboundHandlerAdapter {    /**     * 通道讀和寫     */    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {        System.out.println("Server-Msg【"+msg+"】");        TimeUnit.MILLISECONDS.sleep(2000);        String nowTime = DateTime.now().toString(DatePattern.NORM_DATETIME_PATTERN) ;        ctx.channel().writeAndFlush("hello-client；time：" + nowTime);        ctx.fireChannelActive();    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {        cause.printStackTrace();        ctx.close();    }}

【客戶端】通過Bootstrap類，與服務器建立連接，服務端通過ServerBootstrap啟動服務，綁定在8989端口，然后服務端和客戶端進行通信；

public class NettyClient {    public static void main(String[] args) {        // EventLoop處理事件和IO        NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();        try {            // 客戶端通道引導            Bootstrap bootstrap = new Bootstrap();            bootstrap.group(eventLoopGroup)                    .channel(NioSocketChannel.class).handler(new ClientChannelInit());            // 異步IO的結(jié)果            ChannelFuture channelFuture = bootstrap.connect("localhost", 8989).sync();            channelFuture.channel().closeFuture().sync();        } catch (Exception e){            e.printStackTrace();        } finally {            eventLoopGroup.shutdownGracefully();        }    }}class ClientChannelInit extends ChannelInitializer {    @Override    protected void initChannel(SocketChannel socketChannel) {        // 獲取管道        ChannelPipeline pipeline = socketChannel.pipeline();        // 編碼、解碼器        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));        // 添加自定義的handler        pipeline.addLast("clientHandler", new ClientHandler());    }}class ClientHandler extends ChannelInboundHandlerAdapter {    /**     * 通道讀和寫     */    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {        System.out.println("Client-Msg【"+msg+"】");        TimeUnit.MILLISECONDS.sleep(2000);        String nowTime = DateTime.now().toString(DatePattern.NORM_DATETIME_PATTERN) ;        ctx.channel().writeAndFlush("hello-server；time：" + nowTime);    }    @Override    public void channelActive(ChannelHandlerContext ctx) throws Exception {        ctx.channel().writeAndFlush("channel...active");    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause) throws Exception {        cause.printStackTrace();        ctx.close();    }}

六、參考源碼

編程文檔：https://gitee.com/cicadasmile/butte-java-note應用倉庫：https://gitee.com/cicadasmile/butte-flyer-parent

關鍵詞：