bio阻塞的缺点_java 中的 BIO/NIO/AIO 详解
java 的 IO 演进之路
我们在前面学习了 linux 的 5 种 I/O 模型详解
下面我们一起来学习下如何使用 java 实现 BIO/NIO/AIO 这 3 种不同的网络 IO 模型编程。
BIO 编程
BIO 作为最基础的 IO 版本,实现起来比较简单。
Server
import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;import java.io.PrintWriter;import java.net.ServerSocket;import java.net.Socket;
/** *
BIO 服务端
* @author 老马啸西风
*/
public class TimeServer {
public static void main(String[] args) throws IOException {
final int port = 8088;
ServerSocket serverSocket = new ServerSocket(port);
System.out.println("server started at port " + port);
// 循环监听
while (true) {
Socket socket = serverSocket.accept();
System.out.println("客户端连接成功");
// 读取客户端的信息
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
System.out.println("Server Recevie: " + bufferedReader.readLine());
// 读取客户端的信息
PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true);
String currentTime = System.currentTimeMillis()+"";
printWriter.println(currentTime);
}
}
}
client
import java.io.BufferedReader;import java.io.IOException;import java.io.InputStreamReader;import java.io.PrintWriter;import java.net.Socket;
/** *
BIO 客户端
*
* @author 老马啸西风
*/
public class TimeClient {
public static void main(String[] args) throws IOException {
final int port = 8088;
try(Socket clientSocket = new Socket("127.0.0.1", port)) {
System.out.println("Client started at port " + port);
// 写入信息
PrintWriter printWriter = new PrintWriter(clientSocket.getOutputStream(), true);
printWriter.println("hello bio");
// 读取反馈
BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
System.out.println("client recevie: " + bufferedReader.readLine());
}
}
}
启动测试
启动服务端
server started at port 8088
启动客户端
Client started at port 8088client recevie: 1568643464491
Process finished with exit code 0
再次查看服务端日志
server started at port 8088客户端连接成功Server Recevie: hello bio
线程池版本
BIO 的缺点
缺点其实非常明显,每次都要创建一个线程去处理。
比如我的实现是直接阻塞当前线程的,这当然非常的不友好。
可以使用线线程池的方式进行优化改进。
线程版本
public class TimeThreadServer {
public static void main(String[] args) throws IOException { final int port = 8088; ServerSocket serverSocket = new ServerSocket(port); System.out.println("server started at port " + port);
// 循环监听 while (true) { Socket socket = serverSocket.accept(); System.out.println("客户端连接成功");
new ServerHandler(socket).start(); } }
static class ServerHandler extends Thread {
private final Socket socket;
ServerHandler(Socket socket) { this.socket = socket; }
@Override public void run() { try { // 读取客户端的信息 BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream())); System.out.println("Server Recevie: " + bufferedReader.readLine());
// 读取客户端的信息 PrintWriter printWriter = new PrintWriter(socket.getOutputStream(), true); String currentTime = System.currentTimeMillis()+""; printWriter.println(currentTime); } catch (IOException e) { e.printStackTrace(); } } }}
线程池版本
public static void main(String[] args) throws IOException { final int port = 8088; ServerSocket serverSocket = new ServerSocket(port); System.out.println("server started at port " + port); ExecutorService executorService = Executors.newFixedThreadPool(2); // 循环监听 while (true) { Socket socket = serverSocket.accept(); System.out.println("客户端连接成功"); // 线程池处理 executorService.submit(new ServerHandler(socket)); }}
其他代码保持不变。
优缺点
线程池版本的 BIO 又被称作伪异步 IO。
属于在 NIO 还没有流行之前的一种实战解决方案。
这种方式的性能和 BIO 想比较提升了很多,实现起来也比较简单,但是可靠性相对较差。
NIO 基本概念
Buffer
Java NIO Buffers用于和NIO Channel交互。正如你已经知道的,我们从channel中读取数据到buffers里,从buffer把数据写入到channels.
buffer 本质上就是一块内存区,可以用来写入数据,并在稍后读取出来。
这块内存被NIO Buffer包裹起来,对外提供一系列的读写方便开发的接口。
Channel
Java NIO Channel通道和流非常相似,主要有以下几点区别:
通道可以读也可以写,流一般来说是单向的(只能读或者写)。
通道可以异步读写。
通道总是基于缓冲区Buffer来读写。
Selector
用单线程处理多个channels的好处是我需要更少的线程来处理channel。
实际上,你甚至可以用一个线程来处理所有的channels。
从操作系统的角度来看,切换线程开销是比较昂贵的,并且每个线程都需要占用系统资源,因此暂用线程越少越好。
需要留意的是,现代操作系统和CPU在多任务处理上已经变得越来越好,所以多线程带来的影响也越来越小。
如果一个CPU是多核的,如果不执行多任务反而是浪费了机器的性能。不过这些设计讨论是另外的话题了。
简而言之,通过Selector我们可以实现单线程操作多个channel。
NIO 实现方式
NIO 采取通道(Channel)和缓冲区(Buffer)来传输和保存数据,它是非阻塞式的 I/O,即在等待连接、读写数据(这些都是在一线程以客户端的程序中会阻塞线程的操作)的时候,程序也可以做其他事情,以实现线程的异步操作。
考虑一个即时消息服务器,可能有上千个客户端同时连接到服务器,但是在任何时刻只有非常少量的消息需要读取和分发(如果采用线程池或者一线程一客户端方式,则会非常浪费资源),这就需要一种方法能阻塞等待,直到有一个通道可以进行 I/O 操作。
NIO 的 Selector 选择器就实现了这样的功能,一个 Selector 实例可以同时检查一组信道的 I/O 状态,它就类似一个观察者,只要我们把需要探知的 SocketChannel 告诉 Selector,我们接着做别的事情,当有事件(比如,连接打开、数据到达等)发生时,它会通知我们,传回一组 SelectionKey,我们读取这些 Key,就会获得我们刚刚注册过的 SocketChannel,然后,我们从这个 Channel 中读取数据,接着我们可以处理这些数据。
Selector 内部原理实际是在做一个对所注册的 Channel 的轮询访问,不断的轮询(目前就这一个算法),一旦轮询到一个 Channel 有所注册的事情发生,比如数据来了,它就会读取 Channel 中的数据,并对其进行处理。
要使用选择器,需要创建一个 Selector 实例,并将其注册到想要监控的信道上(通过 Channel 的方法实现)。
最后调用选择器的 select()方法,该方法会阻塞等待,直到有一个或多个信道准备好了 I/O 操作或等待超时,或另一个线程调用了该选择器的 wakeup()方法。
现在,在一个单独的线程中,通过调用 select()方法,就能检查多个信道是否准备好进行 I/O 操作,由于非阻塞 I/O 的异步特性,在检查的同时,我们也可以执行其他任务。
服务端
步骤
(1)创建一个 Selector 实例;
(2)将其注册到各种信道,并指定每个信道上感兴趣的I/O操作;
(3)重复执行:
调用一种 select() 方法;
获取选取的键列表;
对于已选键集中的每个键:
获取信道,并从键中获取附件(如果为信道及其相关的 key 添加了附件的话);
确定准备就绪的操纵并执行,如果是 accept 操作,将接收的信道设置为非阻塞模式,并注册到选择器;
如果需要,修改键的兴趣操作集;
从已选键集中移除键。
代码实现
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.TimeUnit;
/** * @author binbin.hou * @since 1.0.0 */public class NioTcpServer {
/** * 缓冲区的长度 */ private static final int BUFSIZE = 256;
/** * select方法等待信道准备好的最长时间 */ private static final int TIMEOUT = 3000;
/** * 监听的端口号 */ private static final int PORT = 18888;
public static void main(String[] args) throws IOException, InterruptedException { // 1. 实例化一个通道 ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); // 设置为非阻塞模式 serverSocketChannel.configureBlocking(false); // 绑定监听的端口 serverSocketChannel.socket().bind(new InetSocketAddress(PORT)); System.out.println("Server started listen on: " + PORT);
// 2. 构建一个 Selector,用于监听 Channel 的状态 Selector selector = Selector.open(); serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
//3. 不断循环等待 while (true) { //3.1 循环等待直到有通道已经准备好 if(selector.select(TIMEOUT) == 0) { System.out.println("."); TimeUnit.SECONDS.sleep(1); continue; }
//3.2 遍历多有的 key Iterator selectionKeySetIter = selector.selectedKeys().iterator();while(selectionKeySetIter.hasNext()) { SelectionKey selectionKey = selectionKeySetIter.next();// accept I/O形式if(selectionKey.isAcceptable()) { ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) selectionKey.channel();// 获取客户端 channel SocketChannel socketChannel = serverSocketChannel1.accept(); socketChannel.configureBlocking(false);// 选择器注册监听的事件,同时制定关联的附件 socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ | SelectionKey.OP_WRITE, ByteBuffer.allocate(BUFSIZE)); }// 客户端信道已经准备好了读取数据到 bufferif(selectionKey.isReadable()) {// 读取代码 SocketChannel socketChannel = (SocketChannel) selectionKey.channel();// 获取对应的附件信息 ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();long bufferRead = socketChannel.read(byteBuffer);//客户端关闭的链接。可以安全关闭if(bufferRead == -1) { socketChannel.close(); } else {// 缓冲区读取到了数据,将其感兴趣的操作设置为可读可写。 selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);// 打印读取的内容 System.out.println("Server read: " + new String(byteBuffer.array())); } }// 写入处理if(selectionKey.isValid() && selectionKey.isWritable()) { SocketChannel socketChannel = (SocketChannel) selectionKey.channel();// 获取附件 ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();// 重置缓冲区,准备将数据写入到信道 byteBuffer.flip(); socketChannel.write(byteBuffer);//Tells whether there are any elements between the current position and the limit.// 如果已经全部写入到信道,则将该信道感兴趣的操作标识为读if(!byteBuffer.hasRemaining()) { selectionKey.interestOps(SelectionKey.OP_READ); }// 为读取更多的数据腾出空间 byteBuffer.compact(); }// 手动删除 selectionKeySetIter.remove(); } } }}
客户端
代码实现
import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;import java.util.concurrent.TimeUnit;
/** * @author binbin.hou * @since 1.0.0 */public class NioTcpClient {
/** * 监听的端口号 */ private static final int PORT = 18888;
public static void main(String[] args) throws IOException, InterruptedException { //1. 设置为非阻塞 SocketChannel socketChannel = SocketChannel.open(); socketChannel.configureBlocking(false); socketChannel.connect(new InetSocketAddress(PORT));
//2. 连接中... while (!socketChannel.finishConnect()) { System.out.println("."); TimeUnit.SECONDS.sleep(1); } System.out.println("\n");
//3. 写入/读取信息 String info = "hello nio test"; ByteBuffer readBuffer = ByteBuffer.allocate(info.length()); ByteBuffer writeBuffer = ByteBuffer.wrap(info.getBytes());
int totalReceivedBytes = 0; int receivedBytes = 0;
while (totalReceivedBytes // 循环写入 while (writeBuffer.hasRemaining()) { socketChannel.write(writeBuffer); }
receivedBytes = socketChannel.read(readBuffer); // 说明服务端中断 if(receivedBytes == -1) { throw new RuntimeException("Server has been shut done."); } totalReceivedBytes += receivedBytes; }
System.out.println("Client received from server: " + new String(readBuffer.array())); socketChannel.close(); }}
测试
运行服务端
服务端
Server started listen on: 18888
运行客户端
客户端
Client received from server: hello nio test
服务端
...Server read: hello nio test ...
JDK AIO
jdk7中新增了一些与文件(网络)I/O相关的一些api。这些API被称为NIO.2,或称为AIO(Asynchronous I/O)。
AIO最大的一个特性就是异步能力,这种能力对socket与文件I/O都起作用。
实现方式
Future 方式
即提交一个 I/O 操作请求(accept/read/write),返回一个 Future。
然后您可以对 Future 进行检查(调用get(timeout)),确定它是否完成,或者阻塞 IO 操作直到操作正常完成或者超时异常。
使用 Future 方式很简单,需要注意的是,因为Future.get()是同步的,所以如果不仔细考虑使用场合,使用 Future 方式可能很容易进入完全同步的编程模式,从而使得异步操作成为一个摆设。
如果这样,那么原来旧版本的 Socket API 便可以完全胜任,大可不必使用异步 I/O.
Callback 方式
即提交一个 I/O 操作请求,并且指定一个 CompletionHandler。
当异步 I/O 操作完成时,便发送一个通知,此时这个 CompletionHandler 对象的 completed 或者 failed 方法将会被调用。
性能
因为AIO的实施需充分调用OS参与,IO需要操作系统支持、并发也同样需要操作系统的支持,所以性能方面不同操作系统差异会比较明显。
Future 实现方式
Server
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.charset.Charset;import java.util.concurrent.ExecutionException;import java.util.concurrent.Future;import java.util.concurrent.TimeUnit;import java.util.concurrent.TimeoutException;
public class AioFutureServer {
private static final int DEFAULT_PORT = 12345;
private AsynchronousServerSocketChannel serverSocketChannel;
public AioFutureServer() throws IOException { serverSocketChannel = AsynchronousServerSocketChannel.open(); serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT)); System.out.println("Server listen on port: " + DEFAULT_PORT); }
public void startWithFuture() throws InterruptedException, ExecutionException, TimeoutException { while (true) { // 循环接收客户端请求 Future future = serverSocketChannel.accept();// get() 是为了确保 accept 到一个连接 AsynchronousSocketChannel socket = future.get(); handleWithFuture(socket); } }/** * 处理未来的信息 * @param channel 异步客户端 */private void handleWithFuture(AsynchronousSocketChannel channel) throws InterruptedException, ExecutionException, TimeoutException { ByteBuffer readBuf = ByteBuffer.allocate(8); readBuf.clear();// 一次可能读不完while (true) {//get 是为了确保 read 完成,超时时间可以有效避免DOS攻击,如果客户端一直不发送数据,则进行超时处理 Integer integer = channel.read(readBuf).get(10, TimeUnit.SECONDS); System.out.println("read: " + integer);if (integer == -1) {break; } readBuf.flip(); System.out.println("received: " + Charset.forName("UTF-8").decode(readBuf)); readBuf.clear(); } }public static void main(String[] args) throws IOException, InterruptedException, ExecutionException, TimeoutException {new AioFutureServer().startWithFuture(); }}
客户端
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.charset.Charset;import java.util.concurrent.ExecutionException;import java.util.concurrent.Future;import java.util.concurrent.TimeUnit;import java.util.concurrent.TimeoutException;
public class AioClient {
private static final int DEFAULT_PORT = 12345;
public static void main(String[] args) throws IOException, ExecutionException, InterruptedException { AsynchronousSocketChannel client = AsynchronousSocketChannel.open(); client.connect(new InetSocketAddress("localhost", DEFAULT_PORT)).get(); client.write(ByteBuffer.wrap("123456789".getBytes())); }
}
测试
启动服务端
Server listen on port: 12345
启动客户端
服务端日志
read: 8received: 12345678read: 1received: 9Exception in thread "main" java.util.concurrent.ExecutionException: java.io.IOException: 指定的网络名不再可用。
Callback 模式
服务端
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;import java.nio.charset.Charset;import java.util.concurrent.TimeUnit;
public class AioCompletionServer {
private static final int DEFAULT_PORT = 12345;
private AsynchronousServerSocketChannel serverSocketChannel;
public AioCompletionServer() throws IOException { serverSocketChannel = AsynchronousServerSocketChannel.open(); serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT)); System.out.println("Server listen on port: " + DEFAULT_PORT); }
/** * 使用回调的方式 */ public void startWithCompletionHandler() { serverSocketChannel.accept(null, new CompletionHandler() {@Overridepublic void completed(AsynchronousSocketChannel result, Object attachment) {// 再此接收客户端连接 serverSocketChannel.accept(null, this);// 处理结果 handleWithCompletionHandler(result); }@Overridepublic void failed(Throwable exc, Object attachment) { exc.printStackTrace(); } }); }/** * 处理异步的结果 * @param channel 客户端信道 */private void handleWithCompletionHandler(final AsynchronousSocketChannel channel) {try {final long timeout = 10L;final ByteBuffer buffer = ByteBuffer.allocate(8);// 再次读取,还是一种回调的方式。 channel.read(buffer, timeout, TimeUnit.SECONDS, null, new CompletionHandler() {@Overridepublic void completed(Integer result, Object attachment) { System.out.println("read:" + result);if (result == -1) {try { channel.close(); } catch (IOException e) { e.printStackTrace(); }return; } buffer.flip(); System.out.println("received message:" + Charset.forName("UTF-8").decode(buffer)); buffer.clear();// 递归调用,直到结束为止。 channel.read(buffer, timeout, TimeUnit.SECONDS, null, this); }@Overridepublic void failed(Throwable exc, Object attachment) { exc.printStackTrace(); } }); } catch (Exception e) { e.printStackTrace(); } }public static void main(String[] args) throws IOException, InterruptedException {new AioCompletionServer().startWithCompletionHandler();// 沉睡等待处理。 TimeUnit.SECONDS.sleep(100); }}
客户端
同上
小结
本文讲述了 jdk 实现的 bio/nio/aio 的方式,你是否会感觉 jdk 中的 api 设计过于复杂呢?
下一节我们将通过 netty 框架实现上述功能,并讲述我们为什么要选择 netty 作为网络开发的基本工具。
希望本文对你有所帮助,如果喜欢,欢迎点赞收藏转发一波。
我是老马,期待与你的下次相遇。
bio阻塞的缺点_java 中的 BIO/NIO/AIO 详解相关推荐
- java mod %区别_Java中 % 与Math.floorMod() 区别详解
%为取余(rem),Math.floorMod()为取模(mod) 取余取模有什么区别呢? 对于整型数a,b来说,取模运算或者取余运算的方法都是: 1.求 整数商: c = a/b; 2.计算模或者余 ...
- java 静态 编译_Java中的动态和静态编译实例详解
Java中的动态和静态编译实例详解 首先,我们来说说动态和静态编译的问题. Q: java和javascript有什么区别? 总结了一下:有以下几点吧: 1.首先从运行环境来说java代码是在JVM上 ...
- java中迭代器要导包吗_java 中迭代器的使用方法详解
java 中迭代器的使用方法详解 前言: 迭代器模式将一个集合给封装起来,主要是为用户提供了一种遍历其内部元素的方式.迭代器模式有两个优点:①提供给用户一个遍历的方式,而没有暴露其内部实现细节:②把元 ...
- java中math的方法_Java中Math类常用方法代码详解
近期用到四舍五入想到以前整理了一点,就顺便重新整理好经常见到的一些四舍五入,后续遇到常用也会直接在这篇文章更新... public class Demo{ public static void mai ...
- java foreach标签_Java中Velocity foreach循环标签详解
Java中Velocity foreach循环标签详解 Java Velocity中foreach循环可以很容易的遍历数组或者集合. 定义 #foreach( $elem in $allElems) ...
- java中 enum什么意思_Java中枚举Enum的使用详解
在某些情况下,一个类的对象时有限且固定的,如季节类,它只有春夏秋冬4个对象这种实例有限且固定的类,在 Java 中被称为枚举类: 在 Java 中使用 enum 关键字来定义枚举类,其地位与 clas ...
- java io bio nio aio 详解
BIO.NIO.AIO的区别: BIO就是基于Thread per Request的传统server/client实现模式, NIO通常采用Reactor模式, AIO通常采用Proactor模式, ...
- java ==和===_java中==和equals的区别详解
分析前基础了解: 一)JVM把内存划分成两种:一种是栈内存,一种是堆内存. ①在函数中定义的一些基本类型的变量和对象的引用变量(变量名)都在函数的栈内存中分配. ②当在一段代码块定义一个变量时,Jav ...
- java的日期操作_java中对时间的操作详解
代码演示: //数据库中去的日期 Date s = list.get(0).getSdate(); System.out.println(s);// Tue Apr 28 00:00:00 CST 2 ...
最新文章
- windbg 调试技巧
- 网络推广外包——网络推广外包专员浅析网站流量应该如何提升?
- cmd oracle 连接实例_基于winserver的Oracle数据库跨版本下的rman备份恢复
- 如何将本地代码使用Git上传至Github
- 检测python进程是否存活
- 雷赛运动控制卡能不能用c语言_基于PMAC控制卡的三坐标测量机控制系统
- Spring Boot 如何解决多个定时任务阻塞问题?
- java怎么看具体被挂起的线程_Java知多少(65)线程的挂起、恢复和终止
- rhel6.x版本和rehel7.x版本破解密码及恢复损坏的文件分区
- mac中的echo颜色输出
- 中国移动通信研究院笔试题2
- (自用)Linux系统彻底卸载MySQL数据库
- talfta---动态故障树分析软件产品介绍
- CCNet_Criss-Cross Attention for Semantic Segmentation
- graphpad多条不同的曲线_GraphPad Prism
- H3C交换机MPLS配置
- 创业手记 Mr.Hua
- Overleaf使用Latex输入中文的两种方法
- qq屏蔽怎么知道对方信息(qq屏蔽怎么让对方知道)
- C++病毒-----------混乱鼠标