在线乱码卡一卡二卡新HD,最近韩国免费观看视频,国产色无码精品视频国产,亚洲男人的天堂久久香蕉

視訊!IO流中「線程」模型總結
來源:知了一笑    時間:2023-04-07 08:57:06
一、基礎簡介

在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 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() + "端口數據:" + 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 acceptFuture = socketChannel.accept(); AsynchronousSocketChannel asyChannel = acceptFuture.get(30, TimeUnit.SECONDS); if (asyChannel != null && asyChannel.isOpen()){ // 讀數據 ByteBuffer inBuffer = ByteBuffer.allocate(1024); Future readResult = asyChannel.read(inBuffer); readResult.get(); System.out.println("read:"+new String(inBuffer.array())); // 寫數據 inBuffer.flip(); Future 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 result = socketChannel.connect(new InetSocketAddress("127.0.0.1", 8989)); result.get(); // 寫數據 String conVar = "client-hello"; ByteBuffer reqBuffer = ByteBuffer.wrap(conVar.getBytes()); Future writeFuture = socketChannel.write(reqBuffer); writeFuture.get(); // 讀數據 ByteBuffer inBuffer = ByteBuffer.allocate(1024); Future 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多線程模型;

@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的結果 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的結果 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

關鍵詞:

上一篇:

下一篇:

X 關閉

X 關閉

<蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <蜘蛛词>| <文本链> <文本链> <文本链> <文本链> <文本链> <文本链>