IO通訊模型(二)同步非阻塞模式NIO(NonBlocking IO)
阿新 • • 發佈:2018-12-17
同步非阻塞模式(NonBlocking IO)
在非阻塞模式中,發出Socket的accept()
和read()
操作時,如果核心中的資料還沒有準備好,那麼它並不會阻塞使用者程序,而是立刻返回一個資訊。也就是說程序發起一個read操作後,並不需要一直阻塞等待,而是馬上就得到了一個結果。
如果結果發現數據準備完畢就可以讀取資料,然後拷貝到使用者記憶體。如果結果發現數據沒有就緒也會返回,程序繼續不斷的主動詢問
資料的準備情況是非阻塞模式的一個特點。
偽程式碼表示如下:
{ while(read(socket, buffer) != SUCCESS){ } process(buffer); }
Java同步非阻塞模式
如上所述,Java的Socket是阻塞模式的典型應用。在發起accpet()
和read()
請求之後會持續阻塞,但是Java中提供了setSoTimeout()
方法設定超時時間,在固定時間內沒有得到結果,就會結束本次阻塞,等待進行下一次的阻塞輪訓。這是,也就實現了應用層面的非阻塞。
Java中Socket中的setSoTimeout()
方法:
public synchronized void setSoTimeout(int timeout) throws SocketException { if (isClosed()) throw new SocketException("Socket is closed"); if (timeout < 0) throw new IllegalArgumentException("timeout can't be negative"); getImpl().setOption(SocketOptions.SO_TIMEOUT, new Integer(timeout)); }
Java同步非阻塞模式編碼
通過設定setSoTimeout()
使阻塞模式的服務端accpet()
和read()
優化為非阻塞模式。
SocketServerNioListenAndRead.java
import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; import java.net.SocketTimeoutException; /** * <p> * 非阻塞IO - 監聽非阻塞 - 讀取非阻塞 * * @Author niujinpeng * @Date 2018/10/15 14:53 */ public class SocketServerNioListenAndRead { /** * 日誌 */ private static final Logger logger = LoggerFactory.getLogger(SocketServerNioListenAndRead.class); private static Object xWait = new Object(); public static void main(String[] args) throws IOException { ServerSocket serverSocket = null; try { serverSocket = new ServerSocket(83); serverSocket.setSoTimeout(100); while (true) { Socket socket = null; try { socket = serverSocket.accept(); } catch (SocketTimeoutException e) { synchronized (SocketServerNioListenAndRead.xWait) { logger.info("沒有從底層接收到任務資料報文,等待10ms,,模擬事件X的處理時間"); SocketServerNioListenAndRead.xWait.wait(10); } continue; } InputStream input = socket.getInputStream(); OutputStream output = socket.getOutputStream(); Integer sourcePort = socket.getPort(); int maxLen = 2048; byte[] contentBytes = new byte[maxLen]; int realLen; StringBuffer message = new StringBuffer(); // 接收訊息非阻塞實現 socket.setSoTimeout(10); BIORead: while (true) { try { // 讀取的時候,程式會阻塞,知道系統把網路傳過來的資料準備完畢 while ((realLen = input.read(contentBytes, 0, maxLen)) != -1) { message.append(new String(contentBytes, 0, realLen)); /** * 如果收到over,表示傳送完畢 */ if (message.toString().endsWith("over")) { break BIORead; } } } catch (SocketTimeoutException e) { //=========================================================== // 執行到這裡,說明本次read沒有接收到任何資料流 // 主執行緒在這裡又可以做一些事情,記為Y //=========================================================== logger.info("這次沒有從底層接收到任務資料報文,等待10毫秒,模擬事件Y的處理時間"); continue; } } // 輸出資訊 logger.info("伺服器收到來自埠" + sourcePort + "的訊息:" + message.toString()); // 響應 output.write("Done!".getBytes()); output.close(); input.close(); socket.close(); } } catch (SocketException | InterruptedException e) { logger.error(e.getMessage(), e); } finally { if (serverSocket != null) { serverSocket.close(); } } } }
上面的程式碼可以實現監聽和讀取資料的非阻塞,但是還是隻能一個一個的處理,可以使用多執行緒稍微改進
。
SocketServerNioListenThread.java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
/**
* <p>
* 非阻塞IO - 監聽非阻塞 - 讀取非阻塞
* 通過加入執行緒的概念,讓socket server能夠在應用層面
* 通過非阻塞的方式同時處理多個socket套接字
* <p>
* 此時可以實現非阻塞的IO,但是因為呼叫了系統底層的阻塞同步IO,
* 因此仍然沒有從根本上解決問題
*
* @Author niujinpeng
* @Date 2018/10/15 15:23
*/
public class SocketServerNioListenThread {
private static Object xWait = new Object();
private static final Logger LOGGER = LoggerFactory.getLogger(SocketServerNioListenThread.class);
public static void main(String[] args) throws Exception {
ServerSocket serverSocket = new ServerSocket(83);
serverSocket.setSoTimeout(100);
try {
while (true) {
Socket socket = null;
try {
socket = serverSocket.accept();
} catch (SocketTimeoutException e1) {
//===========================================================
// 執行到這裡,說明本次accept沒有接收到任何TCP連線
// 主執行緒在這裡就可以做一些事情,記為X
//===========================================================
synchronized (SocketServerNioListenThread.xWait) {
LOGGER.info("這次沒有從底層接收到任何TCP連線,等待10毫秒,模擬事件X的處理時間");
SocketServerNioListenThread.xWait.wait(10);
}
continue;
}
//當然業務處理過程可以交給一個執行緒(這裡可以使用執行緒池),並且執行緒的建立是很耗資源的。
//最終改變不了.accept()只能一個一個接受socket連線的情況
SocketServerThread socketServerThread = new SocketServerThread(socket);
new Thread(socketServerThread).start();
}
} catch (Exception e) {
LOGGER.error(e.getMessage(), e);
} finally {
if (serverSocket != null) {
serverSocket.close();
}
}
}
}
/**
* 當然,接收到客戶端的socket後,業務的處理過程可以交給一個執行緒來做。
* 但還是改變不了socket被一個一個的做accept()的情況。
*
* @author niujinpeng
*/
class SocketServerThread implements Runnable {
/**
* 日誌
*/
private static final Logger LOGGER = LoggerFactory.getLogger(SocketServerThread.class);
private Socket socket;
public SocketServerThread(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
InputStream in = null;
OutputStream out = null;
try {
in = socket.getInputStream();
out = socket.getOutputStream();
Integer sourcePort = socket.getPort();
int maxLen = 2048;
byte[] contextBytes = new byte[maxLen];
int realLen;
StringBuffer message = new StringBuffer();
//下面我們收取資訊(設定成非阻塞方式,這樣read資訊的時候,又可以做一些其他事情)
this.socket.setSoTimeout(10);
BIORead:
while (true) {
try {
while ((realLen = in.read(contextBytes, 0, maxLen)) != -1) {
message.append(new String(contextBytes, 0, realLen));
/*
* 我們假設讀取到“over”關鍵字,
* 表示客戶端的所有資訊在經過若干次傳送後,完成
* */
if (message.indexOf("over") != -1) {
break BIORead;
}
}
} catch (SocketTimeoutException e2) {
//===========================================================
// 執行到這裡,說明本次read沒有接收到任何資料流
// 主執行緒在這裡又可以做一些事情,記為Y
//===========================================================
LOGGER.info("這次沒有從底層接收到任務資料報文,等待10毫秒,模擬事件Y的處理時間");
continue;
}
}
//下面列印資訊
Long threadId = Thread.currentThread().getId();
LOGGER.info("伺服器(執行緒:" + threadId + ")收到來自於埠:" + sourcePort + "的資訊:" + message);
//下面開始傳送資訊
out.write("回發響應資訊!".getBytes());
//關閉
out.close();
in.close();
this.socket.close();
} catch (Exception e) {
LOGGER.error(e.getMessage(), e);
}
}
}
同步非阻塞模式總結
使用者需要不斷地呼叫,嘗試讀取資料,直到讀取成功後,才繼續處理接收的資料。整個IO請求的過程中,雖然使用者執行緒每次發起IO請求後可以立即返回,但是為了等到資料,仍需要不斷地輪詢、重複請求,消耗了大量的CPU的資源。一般很少直接使用這種模型,而是在其他IO模型中使用非阻塞IO這一特性。
開發難度相對於阻塞IO模式較難,適合併發小且不需要及時響應的網路應用開發。