Java三種IO模型,一次搞懂!
大家好,我是老三,上一節(jié)我們討論了Linux的五種IO模型,接下來,我們從Java語言層面,來看看對(duì)IO的實(shí)現(xiàn)。
在Java中,一共有三種IO模型,分別是阻塞IO(BIO)、非阻塞IO(NIO)和異步IO(AIO)。
Linux五種IO模型和Java三種IO模型
Java BIO
Java BIO就是Java的傳統(tǒng)IO模型,對(duì)應(yīng)了操作系統(tǒng)IO模型里的阻塞IO。
Java BIO相關(guān)的實(shí)現(xiàn)都位于java.io包下,其通信原理是客戶端、服務(wù)端之間通過Socket套接字建立管道連接,然后從管道中獲取對(duì)應(yīng)的輸入/輸出流,最后利用輸入/輸出流對(duì)象實(shí)現(xiàn)發(fā)送/接收信息。
我們來看個(gè)Demo:
- BioServer:
/**
* @Author 三分惡
* @Date 2023/4/30
* @Description BIO服務(wù)端
*/
public class BioServer {
public static void main(String[] args) throws IOException {
//定義一個(gè)ServerSocket服務(wù)端對(duì)象,并為其綁定端口號(hào)
ServerSocket server = new ServerSocket(8888);
System.out.println("===========BIO服務(wù)端啟動(dòng)================");
//對(duì)BIO來講,每個(gè)Socket都需要一個(gè)Thread
while (true) {
//監(jiān)聽客戶端Socket連接
Socket socket = server.accept();
new BioServerThread(socket).start();
}
}
/**
* BIO Server線程
*/
static class BioServerThread extends Thread{
//socket連接
private Socket socket;
public BioServerThread(Socket socket){
this.socket=socket;
}
@Override
public void run() {
try {
//從socket中獲取輸入流
InputStream inputStream=socket.getInputStream();
//轉(zhuǎn)換為
BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(inputStream));
String msg;
//從Buffer中讀取信息,如果讀取到信息則輸出
while((msg=bufferedReader.readLine())!=null){
System.out.println("收到客戶端消息:"+msg);
}
//從socket中獲取輸出流
OutputStream outputStream=socket.getOutputStream();
PrintStream printStream=new PrintStream(outputStream);
//通過輸出流對(duì)象向客戶端傳遞信息
printStream.println("你好,吊毛!");
//清空輸出流
printStream.flush();
//關(guān)閉socket
socket.shutdownOutput();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
- BioClient
/**
* @Author 三分惡
* @Date 2023/4/30
* @Description BIO客戶端
*/
public class BioClient {
public static void main(String[] args) throws IOException {
List<String> names= Arrays.asList("帥哥","靚仔","坤坤");
//通過循環(huán)創(chuàng)建多個(gè)多個(gè)client
for (String name:names){
//創(chuàng)建socket并根據(jù)IP地址與端口連接服務(wù)端
Socket socket=new Socket("127.0.0.1",8888);
System.out.println("===========BIO客戶端啟動(dòng)================");
//從socket中獲取字節(jié)輸出流
OutputStream outputStream=socket.getOutputStream();
//通過輸出流向服務(wù)端傳遞信息
String hello="你好,"+name+"!";
outputStream.write(hello.getBytes());
//清空流,關(guān)閉socket輸出
outputStream.flush();
socket.shutdownOutput();
//從socket中獲取字節(jié)輸入流
InputStream inputStream=socket.getInputStream();
BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(inputStream));
//讀取服務(wù)端消息
String msg;
while((msg=bufferedReader.readLine())!=null){
System.out.println("收到服務(wù)端消息:"+msg);
}
inputStream.close();
outputStream.close();
socket.close();
}
}
}
- 先啟動(dòng)BioServer,再啟動(dòng)BioClient,運(yùn)行結(jié)果
===========BIO服務(wù)端啟動(dòng)================
收到客戶端消息:你好,帥哥!
收到客戶端消息:你好,靚仔!
收到客戶端消息:你好,坤坤!
===========BIO客戶端啟動(dòng)================
收到服務(wù)端消息:你好,吊毛!
===========BIO客戶端啟動(dòng)================
收到服務(wù)端消息:你好,吊毛!
===========BIO客戶端啟動(dòng)================
收到服務(wù)端消息:你好,吊毛!
在上述Java-BIO的通信過程中,如果客戶端一直沒有發(fā)送消息過來,服務(wù)端則會(huì)一直等待下去,從而服務(wù)端陷入阻塞狀態(tài)。同理,由于客戶端也一直在等待服務(wù)端的消息,如果服務(wù)端一直未響應(yīng)消息回來,客戶端也會(huì)陷入阻塞狀態(tài)。
在BioServer定義了一個(gè)類BioServerThread,繼承了Thread類,run方法里主要是通過socket和流來讀取客戶端的消息,以及發(fā)送消息給客戶端,每處理一個(gè)客戶端的Socket連接,就得新建一個(gè)線程。
同時(shí),IO讀寫操作也是阻塞的,如果客戶端一直沒有發(fā)送消息過來,線程就會(huì)進(jìn)入阻塞狀態(tài),一直等待下去。
在BioClient里,循環(huán)創(chuàng)建Socket,向服務(wù)端收發(fā)消息,客戶端的讀寫也是阻塞的。
在這個(gè)Demo里就體現(xiàn)了BIO的兩個(gè)特點(diǎn):
- 一個(gè)客戶端連接對(duì)應(yīng)一個(gè)處理線程
- 讀寫操作都是阻塞的
Java BIO
毫無疑問,不管是創(chuàng)建太多線程,還是阻塞讀寫,都會(huì)浪費(fèi)服務(wù)器的資源。
Java NIO
那么我們就進(jìn)入Java的下一種IO模型——Java NIO,它對(duì)應(yīng)操作系統(tǒng)IO模型中的多路復(fù)用IO,底層采用了epoll實(shí)現(xiàn)。
Java-NIO則是JDK1.4中新引入的API,它在BIO功能的基礎(chǔ)上實(shí)現(xiàn)了非阻塞式的特性,其所有實(shí)現(xiàn)都位于java.nio包下。NIO是一種基于通道、面向緩沖區(qū)的IO操作,相較BIO而言,它能夠更為高效的對(duì)數(shù)據(jù)進(jìn)行讀寫操作,同時(shí)與原先的BIO使用方式也大有不同。
我們還是先來看個(gè)Demo:
- NioServer
/**
* @Author 三分惡
* @Date 2023/4/30
* @Description NIO服務(wù)端
*/
public class NioServer {
public static void main(String[] args) throws IOException {
//創(chuàng)建一個(gè)選擇器selector
Selector selector= Selector.open();
//創(chuàng)建serverSocketChannel
ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
//綁定端口
serverSocketChannel.socket().bind(new InetSocketAddress(8888));
//必須得設(shè)置成非阻塞模式
serverSocketChannel.configureBlocking(false);
//將channel注冊(cè)到selector并設(shè)置監(jiān)聽事件為ACCEPT
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("===========NIO服務(wù)端啟動(dòng)============");
while(true){
//超時(shí)等待
if(selector.select(1000)==0){
System.out.println("===========NIO服務(wù)端超時(shí)等待============");
continue;
}
// 有客戶端請(qǐng)求被輪詢監(jiān)聽到,獲取返回的SelectionKey集合
Iterator<SelectionKey> iterator=selector.selectedKeys().iterator();
//迭代器遍歷SelectionKey集合
while (iterator.hasNext()){
SelectionKey key=iterator.next();
// 判斷是否為ACCEPT事件
if (key.isAcceptable()){
// 處理接收請(qǐng)求事件
SocketChannel socketChannel=((ServerSocketChannel) key.channel()).accept();
//非阻塞模式
socketChannel.configureBlocking(false);
// 注冊(cè)到Selector并設(shè)置監(jiān)聽事件為READ
socketChannel.register(selector,SelectionKey.OP_READ, ByteBuffer.allocate(1024));
System.out.println("成功連接客戶端");
}
//判斷是否為READ事件
if (key.isReadable()){
SocketChannel socketChannel = (SocketChannel) key.channel();
try {
// 獲取以前設(shè)置的附件對(duì)象,如果沒有則新建一個(gè)
ByteBuffer buffer = (ByteBuffer) key.attachment();
if (buffer == null) {
buffer = ByteBuffer.allocate(1024);
key.attach(buffer);
}
// 清空緩沖區(qū)
buffer.clear();
// 將通道中的數(shù)據(jù)讀到緩沖區(qū)
int len = socketChannel.read(buffer);
if (len > 0) {
buffer.flip();
String message = new String(buffer.array(), 0, len);
System.out.println("收到客戶端消息:" + message);
} else if (len < 0) {
// 接收到-1,表示連接已關(guān)閉
key.cancel();
socketChannel.close();
continue;
}
// 注冊(cè)寫事件,下次向客戶端發(fā)送消息
socketChannel.register(selector, SelectionKey.OP_WRITE, buffer);
} catch (IOException e) {
// 取消SelectionKey并關(guān)閉對(duì)應(yīng)的SocketChannel
key.cancel();
socketChannel.close();
}
}
//判斷是否為WRITE事件
if (key.isWritable()){
SocketChannel socketChannel = (SocketChannel) key.channel();
//獲取buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
String hello = "你好,坤坤!";
//清空buffer
buffer.clear();
//buffer中寫入消息
buffer.put(hello.getBytes());
buffer.flip();
//向channel中寫入消息
socketChannel.write(buffer);
buffer.clear();
System.out.println("向客戶端發(fā)送消息:" + hello);
// 設(shè)置下次讀寫操作,向 Selector 進(jìn)行注冊(cè)
socketChannel.register(selector, SelectionKey.OP_READ, buffer);
}
// 移除本次處理的SelectionKey,防止重復(fù)處理
iterator.remove();
}
}
}
}
- NioClient
public class NioClient {
public static void main(String[] args) throws IOException {
// 創(chuàng)建SocketChannel并指定ip地址和端口號(hào)
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8888));
System.out.println("==============NIO客戶端啟動(dòng)================");
// 非阻塞模式
socketChannel.configureBlocking(false);
String hello="你好,靚仔!";
ByteBuffer buffer = ByteBuffer.wrap(hello.getBytes());
// 向通道中寫入數(shù)據(jù)
socketChannel.write(buffer);
System.out.println("發(fā)送消息:" + hello);
buffer.clear();
// 將channel注冊(cè)到Selector并監(jiān)聽READ事件
socketChannel.register(Selector.open(), SelectionKey.OP_READ, buffer);
while (true) {
// 讀取服務(wù)端數(shù)據(jù)
if (socketChannel.read(buffer) > 0) {
buffer.flip();
String msg = new String(buffer.array(), 0, buffer.limit());
System.out.println("收到服務(wù)端消息:" + msg);
break;
}
}
// 關(guān)閉輸入流
socketChannel.shutdownInput();
// 關(guān)閉SocketChannel連接
socketChannel.close();
}
}
- 先運(yùn)行NioServer,再運(yùn)行NioClient,運(yùn)行結(jié)果:
===========NIO服務(wù)端啟動(dòng)============
===========NIO服務(wù)端超時(shí)等待============
===========NIO服務(wù)端超時(shí)等待============
成功連接客戶端
收到客戶端消息:你好,靚仔!
向客戶端發(fā)送消息:你好,坤坤!
==============NIO客戶端啟動(dòng)================
發(fā)送消息:你好,靚仔!
收到服務(wù)端消息:你好,坤坤!
我們?cè)谶@個(gè)案例里實(shí)現(xiàn)了一個(gè)比較簡(jiǎn)單的Java NIO 客戶端服務(wù)端通信,里面有兩個(gè)小的點(diǎn)需要注意,注冊(cè)到選擇器上的通道都必須要為非阻塞模型,同時(shí)通過緩沖區(qū)傳輸數(shù)據(jù)時(shí),必須要調(diào)用flip()方法切換為讀取模式。
代碼流程示意圖
Java-NIO中有三個(gè)核心概念:**Buffer(緩沖區(qū))、Channel(通道)、Selector(選擇器)**。
Java NIO
- 每個(gè)客戶端連連接本質(zhì)上對(duì)應(yīng)著一個(gè)Channel通道,每個(gè)通道都有自己的Buffer緩沖區(qū)來進(jìn)行讀寫,這些Channel被Selector選擇器管理調(diào)度
- Selector負(fù)責(zé)輪詢所有已注冊(cè)的Channel,監(jiān)聽到有事件發(fā)生,才提交給服務(wù)端線程處理,服務(wù)端線程不需要做任何阻塞等待,直接在Buffer里處理Channel事件的數(shù)據(jù)即可,處理完馬上結(jié)束,或返回線程池供其他客戶端事件繼續(xù)使用。
- 通過Selector,服務(wù)端的一個(gè)Thread就可以處理多個(gè)客戶端的請(qǐng)求
- Buffer(緩沖區(qū))就是飯店用來存放食材的儲(chǔ)藏室,當(dāng)服務(wù)員點(diǎn)餐時(shí),需要從儲(chǔ)藏室中取出食材進(jìn)行制作。
- Channel(通道)是用于傳輸數(shù)據(jù)的車道,就像飯店里的上菜窗口,可以快速把點(diǎn)好的菜品送到客人的桌上。
- Selector(選擇器)就是大堂經(jīng)理,負(fù)責(zé)協(xié)調(diào)服務(wù)員、廚師和客人的配合和溝通,以保證整個(gè)就餐過程的效率和順暢。
Java AIO
Java-AIO也被成為NIO2,它是在NIO的基礎(chǔ)上,引入了新的異步通道的概念,并提供了異步文件通道和異步套接字的實(shí)現(xiàn)。
異步通道的實(shí)現(xiàn)體系
它們的主要區(qū)別就在于這個(gè)異步通道,見名知意:使用異步通道去進(jìn)行IO操作時(shí),所有操作都為異步非阻塞的,當(dāng)調(diào)用read()/write()/accept()/connect()方法時(shí),本質(zhì)上都會(huì)交由操作系統(tǒng)去完成,比如要接收一個(gè)客戶端的數(shù)據(jù)時(shí),操作系統(tǒng)會(huì)先將通道中可讀的數(shù)據(jù)先傳入read()回調(diào)方法指定的緩沖區(qū)中,然后再主動(dòng)通知Java程序去處理。
我們還是先來看個(gè)Demo:
- AioServer
/**
* @Author 三分惡
* @Date 2023/5/1
* @Description AIO服務(wù)端
*/
public class AioServer {
public static void main(String[] args) throws Exception {
// 創(chuàng)建異步通道組,處理IO事件
AsynchronousChannelGroup group = AsynchronousChannelGroup.withFixedThreadPool(10, Executors.defaultThreadFactory());
//創(chuàng)建異步服務(wù)器Socket通道,并綁定端口
AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(group).bind(new InetSocketAddress(8888));
System.out.println("=============AIO服務(wù)端啟動(dòng)=========");
// 異步等待接收客戶端連接
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
// 創(chuàng)建ByteBuffer
final ByteBuffer buffer = ByteBuffer.allocate(1024);
@Override
public void completed(AsynchronousSocketChannel channel, Object attachment) {
System.out.println("客戶端連接成功");
try {
buffer.clear();
// 異步讀取客戶端發(fā)送的消息
channel.read(buffer, null, new CompletionHandler<Integer, Object>() {
@Override
public void completed(Integer len, Object attachment) {
buffer.flip();
String message = new String(buffer.array(), 0, len);
System.out.println("收到客戶端消息:" + message);
// 異步發(fā)送消息給客戶端
channel.write(ByteBuffer.wrap(("你好,阿坤!").getBytes()), null, new CompletionHandler<Integer, Object>() {
@Override
public void completed(Integer result, Object attachment) {
// 關(guān)閉輸出流
try {
channel.shutdownOutput();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
try {
channel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
} catch (Exception e) {
e.printStackTrace();
}
// 繼續(xù)異步等待接收客戶端連接
server.accept(null, this);
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
// 繼續(xù)異步等待接收客戶端連接
server.accept(null, this);
}
});
// 等待所有連接都處理完畢
group.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
}
}
- AioClient
/**
* @Author 三分惡
* @Date 2023/5/1
* @Description AIO客戶端
*/
public class AioClient {
public static void main(String[] args) throws Exception {
// 創(chuàng)建異步Socket通道
AsynchronousSocketChannel client = AsynchronousSocketChannel.open();
// 異步連接服務(wù)器
client.connect(new InetSocketAddress("127.0.0.1", 8888), null, new CompletionHandler<Void, Object>() {
// 創(chuàng)建ByteBuffer
final ByteBuffer buffer = ByteBuffer.wrap(("你好,靚仔!").getBytes());
@Override
public void completed(Void result, Object attachment) {
// 異步發(fā)送消息給服務(wù)器
client.write(buffer, null, new CompletionHandler<Integer, Object>() {
// 創(chuàng)建ByteBuffer
final ByteBuffer readBuffer = ByteBuffer.allocate(1024);
@Override
public void completed(Integer result, Object attachment) {
readBuffer.clear();
// 異步讀取服務(wù)器發(fā)送的消息
client.read(readBuffer, null, new CompletionHandler<Integer, Object>() {
@Override
public void completed(Integer result, Object attachment) {
readBuffer.flip();
String msg = new String(readBuffer.array(), 0, result);
System.out.println("收到服務(wù)端消息:" + msg);
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
});
// 等待連接處理完畢
Thread.sleep(1000);
// 關(guān)閉輸入流和Socket通道
client.shutdownInput();
client.close();
}
}
- 看下運(yùn)行結(jié)果
=============AIO服務(wù)端啟動(dòng)=========
客戶端連接成功
收到客戶端消息:你好,靚仔!
收到服務(wù)端消息:你好,阿坤!
可以看到,所有的操作都是異步進(jìn)行,通過completed接收異步回調(diào),通過failed接收錯(cuò)誤回調(diào)。
而且我們發(fā)現(xiàn),相較于之前的NIO而言,AIO其中少了Selector選擇器這個(gè)核心組件,選擇器在NIO中充當(dāng)了協(xié)調(diào)者的角色。
但在Java-AIO中,類似的角色直接由操作系統(tǒng)擔(dān)當(dāng),而且不是采用輪詢的方式監(jiān)聽I(yíng)O事件,而是采用一種類似于“訂閱-通知”的模式。
Java AIO簡(jiǎn)圖
在AIO中,所有創(chuàng)建的通道都會(huì)直接在OS上注冊(cè)監(jiān)聽,當(dāng)出現(xiàn)IO請(qǐng)求時(shí),會(huì)先由操作系統(tǒng)接收、準(zhǔn)備、拷貝好數(shù)據(jù),然后再通知監(jiān)聽對(duì)應(yīng)通道的程序處理數(shù)據(jù)。
Java-AIO這種異步非阻塞式IO也是由操作系統(tǒng)進(jìn)行支持的,在Windows系統(tǒng)中提供了一種異步IO技術(shù):IOCP(I/O Completion Port,所以Windows下的Java-AIO則是依賴于這種機(jī)制實(shí)現(xiàn)。不過在Linux系統(tǒng)中由于沒有這種異步IO技術(shù),所以Java-AIO在Linux環(huán)境中使用的還是epoll這種多路復(fù)用技術(shù)進(jìn)行模擬實(shí)現(xiàn)的。
因?yàn)長(zhǎng)inux的異步IO技術(shù)實(shí)際上不太成熟,所以Java-AIO的實(shí)際應(yīng)用并不是太多,比如大名鼎鼎的網(wǎng)絡(luò)通信框架Netty就沒有采用Java-AIO,而是使用Java-NIO,在代碼層面,自行實(shí)現(xiàn)異步。
小結(jié)
那么這期我們就快速過了一下Java的三種IO機(jī)制,它們的特點(diǎn),我們直接看下圖:
Java三種IO模型
我們也發(fā)現(xiàn),雖然Java-NIO、Java-AIO,在性能上比Java-BIO要強(qiáng)很多,但是可以看到,寫法上一個(gè)比一個(gè)難搞,不過好在基本也沒人直接用Java-NIO、Java-AIO,如果要進(jìn)行網(wǎng)絡(luò)通信,一般都會(huì)采用Netty,它對(duì)原生的Java-NIO進(jìn)行了封裝優(yōu)化,接下來,我們會(huì)繼續(xù)走近Netty,敬請(qǐng)期待。
參考:
[1].《Netty權(quán)威指南》
[2].https://juejin.cn/post/7130952602350534693#heading-14
[3].https://www.jianshu.com/p/670033e5b916