目錄

---

BIO

server端：

?client端：

NIO

單線程模型

Reactor模型

代碼實(shí)現(xiàn)

AIO

單線程AIO代碼實(shí)現(xiàn)

線程池AIO代碼實(shí)現(xiàn)

---

BIO

當(dāng)確定客戶端連接數(shù)很少時(shí)，BIO也可以使用，簡單不易出錯(cuò)，效率低下不代表沒有用武之地。

BIO的accept,read,write都是阻塞的，一個(gè)線程老在那阻塞著，其實(shí)它沒干事，但是它腦門上貼著一個(gè)“I`m busying”我很忙，效率極其低下，即使用線程池還不是一堆線程在那"Busy"

server端：

import java.io.IOException; import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket;public class Server {public static void main(String[] args) throws IOException {ServerSocket ss = new ServerSocket();ss.bind(new InetSocketAddress("127.0.0.1", 8888));while(true) {Socket s = ss.accept(); //阻塞方法new Thread(() -> {handle(s);}).start();}}static void handle(Socket s) {try {byte[] bytes = new byte[1024];int len = s.getInputStream().read(bytes);System.out.println(new String(bytes, 0, len));s.getOutputStream().write(bytes, 0, len);s.getOutputStream().flush();} catch (IOException e) {e.printStackTrace();}} }

?client端：

import java.io.IOException; import java.io.OutputStream; import java.net.Socket;public class Client {public static void main(String[] args) throws IOException {Socket s = new Socket("127.0.0.1", 8888);s.getOutputStream().write("HelloServer".getBytes());s.getOutputStream().flush();//s.getOutputStream().close();System.out.println("write over, waiting for msg back...");byte[] bytes = new byte[1024];int len = s.getInputStream().read(bytes);System.out.println(new String(bytes, 0, len));s.close();} }

NIO

單線程模型

selector選擇器大管家，專門做監(jiān)聽這件事，監(jiān)聽連接、讀、寫事件。

只需要一個(gè)連接就可以處理客戶端的連接，客戶端的讀，客戶端的寫，selector全部都能處理。

使用到的設(shè)計(jì)模式就是observer觀察者模式。

selector說：我對某些事感興趣，哪些事呢？關(guān)于客戶端的連接、讀、寫這些事我統(tǒng)統(tǒng)感興趣！我會每隔一段時(shí)間就去檢查server這些事件有沒有發(fā)生，如果有一個(gè)客戶端要連接了，selector就幫助客戶端和server建立連接，來一個(gè)就幫忙建立一個(gè)連接，來一個(gè)建一個(gè)。selector除了負(fù)責(zé)客戶端的連接之外，還會盯著已經(jīng)連接好的客戶端通道有沒有需要讀和寫的數(shù)據(jù)，需要讀就讀過來，需要寫就寫出去。

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.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set;public class Server {public static void main(String[] args) throws IOException {ServerSocketChannel ssc = ServerSocketChannel.open();ssc.socket().bind(new InetSocketAddress("127.0.0.1", 8888));ssc.configureBlocking(false);System.out.println("server started, listening on :" + ssc.getLocalAddress());Selector selector = Selector.open();ssc.register(selector, SelectionKey.OP_ACCEPT);while(true) {selector.select();Set<SelectionKey> keys = selector.selectedKeys();Iterator<SelectionKey> it = keys.iterator();while(it.hasNext()) {SelectionKey key = it.next();it.remove();handle(key);}}}private static void handle(SelectionKey key) {if(key.isAcceptable()) {try {ServerSocketChannel ssc = (ServerSocketChannel) key.channel();SocketChannel sc = ssc.accept();sc.configureBlocking(false);//new Client////String hostIP = ((InetSocketAddress)sc.getRemoteAddress()).getHostString();/*log.info("client " + hostIP + " trying to connect");for(int i=0; i<clients.size(); i++) {String clientHostIP = clients.get(i).clientAddress.getHostString();if(hostIP.equals(clientHostIP)) {log.info("this client has already connected! is he alvie " + clients.get(i).live);sc.close();return;}}*/sc.register(key.selector(), SelectionKey.OP_READ );} catch (IOException e) {e.printStackTrace();} finally {}} else if (key.isReadable()) { //flipSocketChannel sc = null;try {sc = (SocketChannel)key.channel();ByteBuffer buffer = ByteBuffer.allocate(512);buffer.clear();int len = sc.read(buffer);if(len != -1) {System.out.println(new String(buffer.array(), 0, len));}ByteBuffer bufferToWrite = ByteBuffer.wrap("HelloClient".getBytes());sc.write(bufferToWrite);} catch (IOException e) {e.printStackTrace();} finally {if(sc != null) {try {sc.close();} catch (IOException e) {e.printStackTrace();}}}}} }

?

Reactor模型

引入了線程池，selector領(lǐng)著一幫工人在干活，成了一個(gè)包工頭，不再是一個(gè)人單打獨(dú)斗了。

selector就是一個(gè)boss，只負(fù)責(zé)客戶端的連接，線程池就是worker，需要讀寫了就交給worker來處理，worker中誰閑著就交給誰來處理，而不是像NIO單線程模型一樣有讀有寫了就new一個(gè)工人出來。

代碼實(shí)現(xiàn)

import java.io.ByteArrayOutputStream; 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.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;public class PoolServer {ExecutorService pool = Executors.newFixedThreadPool(50);private Selector selector;//中文測試/**** @throws IOException*/public static void main(String[] args) throws IOException {PoolServer server = new PoolServer();server.initServer(8000);server.listen();}/**** @param port* @throws IOException*/public void initServer(int port) throws IOException {//ServerSocketChannel serverChannel = ServerSocketChannel.open();//serverChannel.configureBlocking(false);//serverChannel.socket().bind(new InetSocketAddress(port));//this.selector = Selector.open();serverChannel.register(selector, SelectionKey.OP_ACCEPT);System.out.println("服務(wù)端啟動成功！");}/**** @throws IOException*/@SuppressWarnings("unchecked")public void listen() throws IOException {// 輪詢訪問selector while (true) {//selector.select();//Iterator ite = this.selector.selectedKeys().iterator();while (ite.hasNext()) {SelectionKey key = (SelectionKey) ite.next();//ite.remove();//if (key.isAcceptable()) {ServerSocketChannel server = (ServerSocketChannel) key.channel();//SocketChannel channel = server.accept();//channel.configureBlocking(false);//channel.register(this.selector, SelectionKey.OP_READ);//} else if (key.isReadable()) {//key.interestOps(key.interestOps()&(~SelectionKey.OP_READ));//pool.execute(new ThreadHandlerChannel(key));}}}} }/**** @param* @throws IOException*/ class ThreadHandlerChannel extends Thread{private SelectionKey key;ThreadHandlerChannel(SelectionKey key){this.key=key;}@Overridepublic void run() {//SocketChannel channel = (SocketChannel) key.channel();//ByteBuffer buffer = ByteBuffer.allocate(1024);//ByteArrayOutputStream baos = new ByteArrayOutputStream();try {int size = 0;while ((size = channel.read(buffer)) > 0) {buffer.flip();baos.write(buffer.array(),0,size);buffer.clear();}baos.close();//byte[] content=baos.toByteArray();ByteBuffer writeBuf = ByteBuffer.allocate(content.length);writeBuf.put(content);writeBuf.flip();channel.write(writeBuf);//if(size==-1){channel.close();}else{//key.interestOps(key.interestOps()|SelectionKey.OP_READ);key.selector().wakeup();}}catch (Exception e) {System.out.println(e.getMessage());}} }

AIO

NIO需要不停地輪詢server有沒有事件發(fā)生。

AIO大概是這樣的：有客戶端要連接的時(shí)候，交給操作系統(tǒng)OS去連接，OS一旦連接上客戶端，就會給大管家selector發(fā)一消息說有人要連上來，要不要給它連，大管家可以有多個(gè)，但一般是一個(gè)，它只負(fù)責(zé)連接客戶端，跟NIO不同的是，selector只需要在那里坐著等就行，不用轉(zhuǎn)圈去輪詢server。

AIO與NIO底層都是使用OS的epoll系統(tǒng)調(diào)用函數(shù)實(shí)現(xiàn)的，epoll就是輪詢，而NIO也是輪詢，所以netty就直接封裝了NIO，其API封裝的更像是AIO。

Windows實(shí)現(xiàn)的AIO是真正的AIO異步IO，效率要比Linux要高。但是服務(wù)器大多數(shù)選擇的是Linux而非Windows。

AIO用到的設(shè)計(jì)模式是鉤子函數(shù)（模板方法）。

并不是線程數(shù)越多越好，線程數(shù)太多會增加線程切換的開銷。

IO都是OS操作系統(tǒng)來實(shí)現(xiàn)的，要不然你程序可以隨便往硬盤里寫點(diǎn)東西，不經(jīng)過OS老大也太放肆了吧？

單線程AIO代碼實(shí)現(xiàn)

import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousServerSocketChannel; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler;public class Server {public static void main(String[] args) throws Exception {final AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(8888));serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {@Overridepublic void completed(AsynchronousSocketChannel client, Object attachment) {serverChannel.accept(null, this);try {System.out.println(client.getRemoteAddress());ByteBuffer buffer = ByteBuffer.allocate(1024);client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {attachment.flip();System.out.println(new String(attachment.array(), 0, result));client.write(ByteBuffer.wrap("HelloClient".getBytes()));}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {exc.printStackTrace();}});} catch (IOException e) {e.printStackTrace();}}@Overridepublic void failed(Throwable exc, Object attachment) {exc.printStackTrace();}});while (true) {Thread.sleep(1000);}} }

線程池AIO代碼實(shí)現(xiàn)

import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousChannelGroup; import java.nio.channels.AsynchronousServerSocketChannel; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors;public class ServerWithThreadGroup {public static void main(String[] args) throws Exception {ExecutorService executorService = Executors.newCachedThreadPool();AsynchronousChannelGroup threadGroup = AsynchronousChannelGroup.withCachedThreadPool(executorService, 1);//中文測試final AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open(threadGroup).bind(new InetSocketAddress(8888));serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {@Overridepublic void completed(AsynchronousSocketChannel client, Object attachment) {serverChannel.accept(null, this);try {System.out.println(client.getRemoteAddress());ByteBuffer buffer = ByteBuffer.allocate(1024);client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {@Overridepublic void completed(Integer result, ByteBuffer attachment) {attachment.flip();System.out.println(new String(attachment.array(), 0, result));client.write(ByteBuffer.wrap("HelloClient".getBytes()));}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {exc.printStackTrace();}});} catch (IOException e) {e.printStackTrace();}}@Overridepublic void failed(Throwable exc, Object attachment) {exc.printStackTrace();}});while (true) {Thread.sleep(1000);}} }

?

總結(jié)

以上是生活随笔為你收集整理的Java IO模型--BIO、NIO Single Thread、NIO Reactor、AIO单线程及多线程AIO的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。