一、基礎簡介
二、同步阻塞
- 1、模型圖解
- 2、參考案例
三、同步非阻塞
- 1、模型圖解
- 2、參考案例
四、異步非阻塞
- 1、模型圖解
- 2、參考案例
五、Reactor模型
- 1、模型圖解
  - 1.1 Reactor設計原理
  - 1.2 單Reactor單執行緒
  - 1.3 單Reactor多執行緒
  - 1.4 主從Reactor多執行緒
- 2、參考案例
六、參考原始碼

IO流模塊：經常看、經常用、經常忘；

一、基礎簡介

在IO流的網路模型中，以常見的「客戶端-服務端」互動場景為例；

客戶端與服務端進行通信「互動」，可能是同步或者異步，服務端進行「流」處理時，可能是阻塞或者非阻塞模式，當然也有自定義的業務流程需要執行，從處理邏輯看就是「讀取資料-業務執行-應答寫資料」的形式；

Java提供「三種」IO網路編程模型，即：「BIO同步阻塞」、「NIO同步非阻塞」、「AIO異步非阻塞」；

二、同步阻塞

1、模型圖解

BIO即同步阻塞，服務端收到客戶端的請求時，會啟動一個執行緒處理，「互動」會阻塞直到整個流程結束；

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

2、參考案例

【服務端】啟動ServerSocket接收客戶端的請求，經過一系列邏輯之后，向客戶端發送訊息，注意這里執行緒的10秒休眠；

public class SocketServer01 {
    public static void main(String[] args) throws Exception {
        // 1、創建Socket服務端
        ServerSocket serverSocket = new ServerSocket(8080);
        // 2、方法阻塞等待，直到有客戶端連接
        Socket socket = serverSocket.accept();
        // 3、輸入流，輸出流
        InputStream inStream = socket.getInputStream();
        OutputStream outStream = socket.getOutputStream();
        // 4、資料接收和回應
        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);
        }
        // 回應資料
        Thread.sleep(10000);
        outStream.write("sever-8080-write;".getBytes());
        // 5、資源關閉
        IoClose.ioClose(outStream,inStream,socket,serverSocket);
    }
}

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

public class SocketClient01 {
    public static void main(String[] args) throws Exception {
        // 1、創建Socket客戶端
        Socket socket = new Socket(InetAddress.getLocalHost(), 8080);
        // 2、輸入流，輸出流
        OutputStream outStream = socket.getOutputStream();
        InputStream inStream = socket.getInputStream();
        // 3、資料發送和回應接收
        // 發送資料
        outStream.write("client-hello".getBytes());
        // 接收資料
        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即同步非阻塞，服務端可以實作一個執行緒，處理多個客戶端請求連接，服務端的并發能力得到極大的提升；

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

2、參考案例

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

public class SocketServer01 {
    public static void main(String[] args) throws Exception {
        try {
            //啟動服務開啟監聽
            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<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> 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() + "埠資料:" + new String(readArr));
                            buffer.clear();
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

【客戶端】每隔3秒持續的向通道內寫資料，服務端通過輪詢多路復用器，持續的讀取資料；

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發送一次資料
            while (true) {
                Thread.sleep(3000);
                writeBuffer.rewind();
                socketChannel.write(writeBuffer);
                writeBuffer.clear();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

四、異步非阻塞

1、模型圖解

AIO即異步非阻塞，對于通道內資料的「讀」和「寫」動作，都是采用異步的模式，對于性能的提升是巨大的；

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

2、參考案例

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

public class SocketServer01 {
    public static void main(String[] args) throws Exception {
        // 啟動服務開啟監聽
        AsynchronousServerSocketChannel socketChannel = AsynchronousServerSocketChannel.open() ;
        socketChannel.bind(new InetSocketAddress("127.0.0.1", 8989));
        // 指定30秒內獲取客戶端連接，否則超時
        Future<AsynchronousSocketChannel> acceptFuture = socketChannel.accept();
        AsynchronousSocketChannel asyChannel = acceptFuture.get(30, TimeUnit.SECONDS);

        if (asyChannel != null && asyChannel.isOpen()){
            // 讀資料
            ByteBuffer inBuffer = ByteBuffer.allocate(1024);
            Future<Integer> readResult = asyChannel.read(inBuffer);
            readResult.get();
            System.out.println("read："+new String(inBuffer.array()));

            // 寫資料
            inBuffer.flip();
            Future<Integer> writeResult = asyChannel.write(ByteBuffer.wrap("server-hello".getBytes()));
            writeResult.get();
        }

        // 關閉資源
        asyChannel.close();
    }
}

【客戶端】相關「connect」、「read」、「write」方法呼叫是異步的，通過Future來獲取計算的結果；

public class SocketClient01 {
    public static void main(String[] args) throws Exception {
        // 連接服務端
        AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
        Future<Void> result = socketChannel.connect(new InetSocketAddress("127.0.0.1", 8989));
        result.get();

        // 寫資料
        String conVar = "client-hello";
        ByteBuffer reqBuffer = ByteBuffer.wrap(conVar.getBytes());
        Future<Integer> writeFuture = socketChannel.write(reqBuffer);
        writeFuture.get();

        // 讀資料
        ByteBuffer inBuffer = ByteBuffer.allocate(1024);
        Future<Integer> readFuture = socketChannel.read(inBuffer);
        readFuture.get();
        System.out.println("read："+new String(inBuffer.array()));

        // 關閉資源
        socketChannel.close();
    }
}

五、Reactor模型

1、模型圖解

這部分內容，可以參考「Doug Lea的《IO》」檔案，查看更多細節；

1.1 Reactor設計原理

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

Reactor：負責事件的監聽和分發；Handler：負責處理事件，核心邏輯「read讀」、「decode解碼」、「compute業務計算」、「encode編碼」、「send應答資料」；

1.2 單Reactor單執行緒

【1】Reactor執行緒通過select監聽客戶端的請求事件，收到事件后通過Dispatch進行分發；

【2】如果是建立連接請求事件，Acceptor通過「accept」方法獲取連接，并創建一個Handler物件來處理后續業務；

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

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

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

1.3 單Reactor多執行緒

【1】Reactor執行緒通過select監聽客戶端的請求事件，收到事件后通過Dispatch進行分發；

【2】如果是建立連接請求事件，Acceptor通過「accept」方法獲取連接，并創建一個Handler物件來處理后續業務；

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

【4】在Handler中，只負責事件回應不處理具體業務，將資料發送給Worker執行緒池來處理；

【5】Worker執行緒池會分配具體的執行緒來處理業務，最后把結果回傳給Handler做回應；

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

1.4 主從Reactor多執行緒

【1】 MainReactor主執行緒通過select監聽客戶端的請求事件，收到事件后通過Dispatch進行分發；

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

【3】如果不是連接請求事件，則MainReactor將連接分配給SubReactor，SubReactor呼叫當前連接的Handler來處理；

【4】在Handler中，只負責事件回應不處理具體業務，將資料發送給Worker執行緒池來處理；

【5】Worker執行緒池會分配具體的執行緒來處理業務，最后把結果回傳給Handler做回應；

這種模式Reactor執行緒分工明確，MainReactor負責接收新的請求連接，SubReactor負責后續的互動業務，適應于高并發的處理場景，是Netty組件通信框架的所采用的模式；

2、參考案例

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

@Slf4j
public 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的結果
            ChannelFuture channelFuture = serverBootstrap.bind(8989).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }
}

class ServerChannelInit extends ChannelInitializer<SocketChannel> {
    @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的結果
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8989).sync();
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

class ClientChannelInit extends ChannelInitializer<SocketChannel> {
    @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

Gitee主頁： https://gitee.com/cicadasmile/butte-java-note

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/549382.html

標籤：Java

上一篇：生產事故-記一次特殊的OOM排查

下一篇：Java GenericObjectPool 物件池化技術--SpringBoot sftp 連接池工具類

IO流中「執行緒」模型總結

一、基礎簡介

二、同步阻塞

1、模型圖解

2、參考案例

三、同步非阻塞

1、模型圖解

2、參考案例

四、異步非阻塞

1、模型圖解

2、參考案例

五、Reactor模型

1、模型圖解

1.1 Reactor設計原理

1.2 單Reactor單執行緒

1.3 單Reactor多執行緒

1.4 主從Reactor多執行緒

2、參考案例

六、參考原始碼