摘要:前言本篇主要講解中的機制和網絡通訊中處理高并發的分為兩塊第一塊講解多線程下的機制第二塊講解如何在機制下優化資源的浪費服務器單線程下的機制就不用我介紹了,不懂得可以去查閱下資料那么多線程下,如果進行套接字的使用呢我們使用最簡單的服務器來幫助大
前言
本篇主要講解Java中的IO機制和網絡通訊中處理高并發的NIO
分為兩塊:
第一塊講解多線程下的IO機制
第二塊講解如何在IO機制下優化CPU資源的浪費(New IO)
單線程下的socket機制就不用我介紹了,不懂得可以去查閱下資料
那么多線程下,如果進行套接字的使用呢?
我們使用最簡單的echo服務器來幫助大家理解
首先,來看下多線程下服務端和客戶端的工作流程圖:
可以看到,多個客戶端同時向服務端發送請求
服務端做出的措施是開啟多個線程來匹配相對應的客戶端
并且每個線程去獨自完成他們的客戶端請求
原理講完了我們來看下是如何實現的
在這里我寫了一個簡單的服務器
用到了線程池的技術來創建線程(具體代碼作用我已經加了注釋):
public class MyServer { private static ExecutorService executorService = Executors.newCachedThreadPool(); //創建一個線程池 private static class HandleMsg implements Runnable{ //一旦有新的客戶端請求,創建這個線程進行處理 Socket client; //創建一個客戶端 public HandleMsg(Socket client){ //構造傳參綁定 this.client = client; } @Override public void run() { BufferedReader bufferedReader = null; //創建字符緩存輸入流 PrintWriter printWriter = null; //創建字符寫入流 try { bufferedReader = new BufferedReader(new InputStreamReader(client.getInputStream())); //獲取客戶端的輸入流 printWriter = new PrintWriter(client.getOutputStream(),true); //獲取客戶端的輸出流,true是隨時刷新 String inputLine = null; long a = System.currentTimeMillis(); while ((inputLine = bufferedReader.readLine())!=null){ printWriter.println(inputLine); } long b = System.currentTimeMillis(); System.out.println("此線程花費了:"+(b-a)+"秒!"); } catch (IOException e) { e.printStackTrace(); }finally { try { bufferedReader.close(); printWriter.close(); client.close(); } catch (IOException e) { e.printStackTrace(); } } } } public static void main(String[] args) throws IOException { //服務端的主線程是用來循環監聽客戶端請求 ServerSocket server = new ServerSocket(8686); //創建一個服務端且端口為8686 Socket client = null; while (true){ //循環監聽 client = server.accept(); //服務端監聽到一個客戶端請求 System.out.println(client.getRemoteSocketAddress()+"地址的客戶端連接成功!"); executorService.submit(new HandleMsg(client)); //將該客戶端請求通過線程池放入HandlMsg線程中進行處理 } } }
上述代碼中我們使用一個類編寫了一個簡單的echo服務器
在主線程中用死循環來開啟端口監聽
有了服務器,我們就可以對其進行訪問,并且發送一些字符串數據
服務器的功能是返回這些字符串,并且打印出線程占用時間
下面來寫個簡單的客戶端來響應服務端:
public class MyClient { public static void main(String[] args) throws IOException { Socket client = null; PrintWriter printWriter = null; BufferedReader bufferedReader = null; try { client = new Socket(); client.connect(new InetSocketAddress("localhost",8686)); printWriter = new PrintWriter(client.getOutputStream(),true); printWriter.println("hello"); printWriter.flush(); bufferedReader = new BufferedReader(new InputStreamReader(client.getInputStream())); //讀取服務器返回的信息并進行輸出 System.out.println("來自服務器的信息是:"+bufferedReader.readLine()); } catch (IOException e) { e.printStackTrace(); }finally { printWriter.close(); bufferedReader.close(); client.close(); } } }
代碼中,我們用字符流發送了一個hello字符串過去,如果代碼沒問題
服務器會返回一個hello數據,并且打印出我們設置的日志信息
我們來運行:
1.打開server,開啟循環監聽:
2.打開一個客戶端:
可以看到客戶端打印出了返回結果
3.查看服務端日志:
很好,一個簡單的多線程套接字編程就實現了
但是試想一下:
如果一個客戶端請求中,在IO寫入到服務端過程中加入Sleep,
使每個請求占用服務端線程10秒
然后有大量的客戶端請求,每個請求都占用那么長時間
那么服務端的并能能力就會大幅度下降
這并不是因為服務端有多少繁重的任務,而僅僅是因為服務線程在等待IO(因為accept,read,write都是阻塞式的)
讓高速運行的CPU去等待及其低效的網絡IO是非常不合算的行為
這時候該怎么辦?
NIONew IO成功的解決了上述問題,它是怎樣解決的呢?
IO處理客戶端請求的最小單位是線程
而NIO使用了比線程還小一級的單位:通道(Channel)
可以說,NIO中只需要一個線程就能完成所有接收,讀,寫等操作
要學習NIO,首先要理解它的三大核心
Selector,選擇器
Buffer,緩沖區
Channel,通道
博主不才,畫了張丑圖給大家加深下印象 ^ . ^
再給一張TCP下的NIO工作流程圖(好難畫的線條...)
大家大致看懂就行,我們一步步來
Buffer首先要知道什么是Buffer
在NIO中數據交互不再像IO機制那樣使用流
而是使用Buffer(緩沖區)
博主覺得圖才是最容易理解的
所以...
可以看出Buffer在整個工作流程中的位置
buffer實際上是一個容器,一個連續數組,它通過幾個變量來保存這個數據的當前位置狀態:
1.capacity:容量,緩沖區能容納元素的數量
2.position:當前位置,是緩沖區中下一次發生讀取和寫入操作的索引,當前位置通過大多數讀寫操作向前推進
3.limit:界限,是緩沖區中最后一個有效位置之后下一個位置的索引
如圖:
幾個常用方法:
.flip() //將limit設置為position,然后position重置為0,返回對緩沖區的引用 .clear() //清空調用緩沖區并返回對緩沖區的引用
來點實際點的,上面圖中的具體代碼如下:
1.首先給Buffer分配空間,以字節為單位
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
創建一個ByteBuffer對象并且指定內存大小
2.向Buffer中寫入數據:
1).數據從Channel到Buffer:channel.read(byteBuffer); 2).數據從Client到Buffer:byteBuffer.put(...);
3.從Buffer中讀取數據:
1).數據從Buffer到Channel:channel.write(byteBuffer); 2).數據從Buffer到Server:byteBuffer.get(...);Selector
選擇器是NIO的核心,它是channel的管理者
通過執行select()阻塞方法,監聽是否有channel準備好
一旦有數據可讀,此方法的返回值是SelectionKey的數量
所以服務端通常會死循環執行select()方法,直到有channl準備就緒,然后開始工作
每個channel都會和Selector綁定一個事件,然后生成一個SelectionKey的對象
需要注意的是:
channel和Selector綁定時,channel必須是非阻塞模式
而FileChannel不能切換到非阻塞模式,因為它不是套接字通道,所以FileChannel不能和Selector綁定事件
在NIO中一共有四種事件:
1.SelectionKey.OP_CONNECT:連接事件
2.SelectionKey.OP_ACCEPT:接收事件
3.SelectionKey.OP_READ:讀事件
4.SelectionKey.OP_WRITE:寫事件
共有四種通道:
FileChannel:作用于IO文件流
DatagramChannel:作用于UDP協議
SocketChannel:作用于TCP協議
ServerSocketChannel:作用于TCP協議
本篇文章通過常用的TCP協議來講解NIO
我們以ServerSocketChannel為例:
打開一個ServerSocketChannel通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
關閉ServerSocketChannel通道:
serverSocketChannel.close();
循環監聽SocketChannel:
while(true){ SocketChannel socketChannel = serverSocketChannel.accept(); clientChannel.configureBlocking(false); }
clientChannel.configureBlocking(false);語句是將此通道設置為非阻塞,也就是異步
自由控制阻塞或非阻塞便是NIO的特性之一
SelectionKey是通道和選擇器交互的核心組件
比如在SocketChannel上綁定一個Selector,并注冊為連接事件:
SocketChannel clientChannel = SocketChannel.open(); clientChannel.configureBlocking(false); clientChannel.connect(new InetSocketAddress(port)); clientChannel.register(selector, SelectionKey.OP_CONNECT);
核心在register()方法,它返回一個SelectionKey對象
來檢測channel事件是那種事件可以使用以下方法:
selectionKey.isAcceptable(); selectionKey.isConnectable(); selectionKey.isReadable(); selectionKey.isWritable();
服務端便是通過這些方法 在輪詢中執行相對應操作
當然通過Channel與Selector綁定的key也可以反過來拿到他們
Channel channel = selectionKey.channel(); Selector selector = selectionKey.selector();
在Channel上注冊事件時,我們也可以順帶綁定一個Buffer:
clientChannel.register(key.selector(), SelectionKey.OP_READ,ByteBuffer.allocateDirect(1024));
或者綁定一個Object:
selectionKey.attach(Object); Object anthorObj = selectionKey.attachment();NIO的TCP服務端
講了這么多,都是理論
我們來看下最簡單也是最核心的代碼(加那么多注釋很不優雅,但方便大家看懂):
package cn.blog.test.NioTest; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.*; import java.nio.charset.Charset; import java.util.Iterator; import java.util.Set; public class MyNioServer { private Selector selector; //創建一個選擇器 private final static int port = 8686; private final static int BUF_SIZE = 10240; private void initServer() throws IOException { //創建通道管理器對象selector this.selector=Selector.open(); //創建一個通道對象channel ServerSocketChannel channel = ServerSocketChannel.open(); channel.configureBlocking(false); //將通道設置為非阻塞 channel.socket().bind(new InetSocketAddress(port)); //將通道綁定在8686端口 //將上述的通道管理器和通道綁定,并為該通道注冊OP_ACCEPT事件 //注冊事件后,當該事件到達時,selector.select()會返回(一個key),如果該事件沒到達selector.select()會一直阻塞 SelectionKey selectionKey = channel.register(selector,SelectionKey.OP_ACCEPT); while (true){ //輪詢 selector.select(); //這是一個阻塞方法,一直等待直到有數據可讀,返回值是key的數量(可以有多個) Set keys = selector.selectedKeys(); //如果channel有數據了,將生成的key訪入keys集合中 Iterator iterator = keys.iterator(); //得到這個keys集合的迭代器 while (iterator.hasNext()){ //使用迭代器遍歷集合 SelectionKey key = (SelectionKey) iterator.next(); //得到集合中的一個key實例 iterator.remove(); //拿到當前key實例之后記得在迭代器中將這個元素刪除,非常重要,否則會出錯 if (key.isAcceptable()){ //判斷當前key所代表的channel是否在Acceptable狀態,如果是就進行接收 doAccept(key); }else if (key.isReadable()){ doRead(key); }else if (key.isWritable() && key.isValid()){ doWrite(key); }else if (key.isConnectable()){ System.out.println("連接成功!"); } } } } public void doAccept(SelectionKey key) throws IOException { ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); System.out.println("ServerSocketChannel正在循環監聽"); SocketChannel clientChannel = serverChannel.accept(); clientChannel.configureBlocking(false); clientChannel.register(key.selector(),SelectionKey.OP_READ); } public void doRead(SelectionKey key) throws IOException { SocketChannel clientChannel = (SocketChannel) key.channel(); ByteBuffer byteBuffer = ByteBuffer.allocate(BUF_SIZE); long bytesRead = clientChannel.read(byteBuffer); while (bytesRead>0){ byteBuffer.flip(); byte[] data = byteBuffer.array(); String info = new String(data).trim(); System.out.println("從客戶端發送過來的消息是:"+info); byteBuffer.clear(); bytesRead = clientChannel.read(byteBuffer); } if (bytesRead==-1){ clientChannel.close(); } } public void doWrite(SelectionKey key) throws IOException { ByteBuffer byteBuffer = ByteBuffer.allocate(BUF_SIZE); byteBuffer.flip(); SocketChannel clientChannel = (SocketChannel) key.channel(); while (byteBuffer.hasRemaining()){ clientChannel.write(byteBuffer); } byteBuffer.compact(); } public static void main(String[] args) throws IOException { MyNioServer myNioServer = new MyNioServer(); myNioServer.initServer(); } }
我打印了監聽channel,告訴大家ServerSocketChannel是在什么時候開始運行的
如果配合NIO客戶端的debug,就能很清楚的發現,進入select()輪詢前
雖然已經有了ACCEPT事件的KEY,但select()默認并不會去調用
而是要等待有其它感興趣事件被select()捕獲之后,才會去調用ACCEPT的SelectionKey
這時候ServerSocketChannel才開始進行循環監聽
也就是說一個Selector中,始終保持著ServerSocketChannel的運行
而serverChannel.accept();真正做到了異步(在initServer方法中的channel.configureBlocking(false);)
如果沒有接受到connect,會返回一個null
如果成功連接了一個SocketChannel,則此SocketChannel會注冊寫入(READ)事件
并且設置為異步
有服務端必定有客戶端
其實如果能完全理解了服務端
客戶端的代碼大同小異
package cn.blog.test.NioTest; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; import java.util.Iterator; public class MyNioClient { private Selector selector; //創建一個選擇器 private final static int port = 8686; private final static int BUF_SIZE = 10240; private static ByteBuffer byteBuffer = ByteBuffer.allocate(BUF_SIZE); private void initClient() throws IOException { this.selector = Selector.open(); SocketChannel clientChannel = SocketChannel.open(); clientChannel.configureBlocking(false); clientChannel.connect(new InetSocketAddress(port)); clientChannel.register(selector, SelectionKey.OP_CONNECT); while (true){ selector.select(); Iterator輸出結果iterator = selector.selectedKeys().iterator(); while (iterator.hasNext()){ SelectionKey key = iterator.next(); iterator.remove(); if (key.isConnectable()){ doConnect(key); }else if (key.isReadable()){ doRead(key); } } } } public void doConnect(SelectionKey key) throws IOException { SocketChannel clientChannel = (SocketChannel) key.channel(); if (clientChannel.isConnectionPending()){ clientChannel.finishConnect(); } clientChannel.configureBlocking(false); String info = "服務端你好!!"; byteBuffer.clear(); byteBuffer.put(info.getBytes("UTF-8")); byteBuffer.flip(); clientChannel.write(byteBuffer); //clientChannel.register(key.selector(),SelectionKey.OP_READ); clientChannel.close(); } public void doRead(SelectionKey key) throws IOException { SocketChannel clientChannel = (SocketChannel) key.channel(); clientChannel.read(byteBuffer); byte[] data = byteBuffer.array(); String msg = new String(data).trim(); System.out.println("服務端發送消息:"+msg); clientChannel.close(); key.selector().close(); } public static void main(String[] args) throws IOException { MyNioClient myNioClient = new MyNioClient(); myNioClient.initClient(); } }
這里我打開一個服務端,兩個客戶端:
接下來,你可以試下同時打開一千個客戶端,只要你的CPU夠給力,服務端就不可能因為阻塞而降低性能
以上便是Java NIO的基礎詳解
謝謝閱讀和關注~
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/70727.html
摘要:的選擇器允許單個線程監視多個輸入通道。一旦執行的線程已經超過讀取代碼中的某個數據片段,該線程就不會在數據中向后移動通常不會。 1、引言 很多初涉網絡編程的程序員,在研究Java NIO(即異步IO)和經典IO(也就是常說的阻塞式IO)的API時,很快就會發現一個問題:我什么時候應該使用經典IO,什么時候應該使用NIO? 在本文中,將嘗試用簡明扼要的文字,闡明Java NIO和經典IO之...
摘要:好了,目前還不難,我們起碼知道這個抽象類上面的部分關系,因此當然也有自己的方法,如下。又來一個的供應商好吧,大佬們總是喜歡用一些設計模式的東西,沒錯,也是一個抽象類,這個現在不用太在意了。 前言 java nio,一個入門netty之前需要了解下的非阻塞I/O實現,傳統的Socket通信,啟動監聽后accept會一直處于阻塞狀態,那么如果你想要多個(并發)通信時,那么我們就需要多個線性...
摘要:是一個用來替代標準的新型數據傳遞方式,像現在分布式架構中會經常存在他的身影。這個方法會一直阻塞到某個注冊的通道有事件就緒。保持不變,仍然表示能從中讀取多少個元素等與通過調用方法,可以標記中的一個特定。 Java NIO是一個用來替代標準Java IO API的新型數據傳遞方式,像現在分布式架構中會經常存在他的身影。其比傳統的IO更加高效,非阻塞,異步,雙向 NIO主體結構 showIm...
摘要:當我們需要與進行交互時我們就需要使用到即數據從讀取到中并且從中寫入到中實際上一個其實就是一塊內存區域我們可以在這個內存區域中進行數據的讀寫其實是這樣的內存塊的一個封裝并提供了一些操作方法讓我們能夠方便地進行數據的讀寫類型有這些覆蓋了能從中傳 Java NIO Buffer 當我們需要與 NIO Channel 進行交互時, 我們就需要使用到 NIO Buffer, 即數據從 Buffe...
摘要:允許一個單一的線程來操作多個如果我們的應用程序中使用了多個那么使用很方便的實現這樣的目的但是因為在一個線程中使用了多個因此也會造成了每個傳輸效率的降低使用的圖解如下為了使用我們首先需要將注冊到中隨后調用的方法這個方法會阻塞直到注冊在中的發送 Selector Selector 允許一個單一的線程來操作多個 Channel. 如果我們的應用程序中使用了多個 Channel, 那么使用 S...
閱讀 3655·2021-09-22 15:15
閱讀 3559·2021-08-12 13:24
閱讀 1312·2019-08-30 15:53
閱讀 1822·2019-08-30 15:43
閱讀 1184·2019-08-29 17:04
閱讀 2793·2019-08-29 15:08
閱讀 1581·2019-08-29 13:13
閱讀 3088·2019-08-29 11:06