BIO、NIO、AIO学习
李羽秋
2022年06月19日 · 阅读 1,314
BIO、NIO、AIO学习
1. Java 的 I/O演进之路
1.1 I/O模型基本说明
I/O 模型 就是用什么样的通道或者说是通信格式和架构进行数据的传输和接收，很大程度上决定了程序通信的性能，Java共支持3种网络编程的I/O模型:BIO、NIO、AlO。实际通信需求下，要根据不同的业务场景和性能需求决定选择不同的I/O模型。
1.2 Java BIO
同步并阻塞(传统阻塞型)，服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个线程进行处理，如果这个连接不做任何事情会造成不必要的线程开销。
1.3 Java NIO
Java NIO： 同步非阻塞，服务器实现模式为一个线程处理多个请求，即客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有I/O请求就进行处理。
1.4 Java AIO
Java AIO : 异步非阻塞，服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程进行处理，一般适用于连接数多且连接时间较长的应用。
1.5 BIO、NIO、AIO适用场景分析
- BIO方式适用于连接数目比较小且固定的架构，这种方式对服务器资源要求比较高，并发局限于应用中，JDK1.4以前的唯一选择，但程序简单易理解。
- NIO方式适用于连接数目多且连接比较短（轻操作）的架构，比如聊天服务器，弹幕系统，服务器间通讯等。编程比较复杂，JDK1.4开始支持。
- AlO方式使用于连接数目多且连接比较长（重操作）的架构，比如相册服务器，充分调用OS参与并发操作，编程比较复杂，JDK7开始支持。
2. BIO 讲解
2.1 基本介绍
- ·Java BIO 就是传统的java io 编程，其相关的类和接口在java.io
- BlO(blocking l/O):同步阻塞，服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个线程进行处理，如果这个连接不做任何事情会造成不必要的线程开销，可以通过线程池机制改善(实现多个客户连接服务器)
2.2 工作机制
2.3 传统BIO编程实例
网络编程的基本模型是Client/Server模型，也就是两个进程之间进行相互通信，其中服务端提供位置信息（绑定ip和端口），客户端通过连接操作向服务端监听的端口地址发起连接请求，基于tcp协议下进行三次握手连接，连接成功后，双方通过网络套接字（Socket)进行通信。
传统的同步阻塞模型开发中，服务端ServerSocket负责绑定ip地址，启动监听端口，客户端Socket负责发起连接操作。连接成功之后，双方通过输入和输出流进行同步阻塞式通信。
客户端
/***
* 客户端
*/
public class Client {
public static void main(String[] args) throws IOException {
// 1.创建socket对象请求服务器链接
Socket socket = new Socket("127.0.0.1",9999);
// 2. 从socket对象中获取一个字节输出流
OutputStream os = socket.getOutputStream();
// 3. 把字节输出流包装成一个打印流
PrintStream ps = new PrintStream(os);
ps.println("hello world!服务器，您好");
ps.flush();
}
}
服务端
/***
* 目标: 客户端发送信息，服务端接收信息
*/
public class Server {
public static void main(String[] args) {
try {
// 1. 定义一个ServerSocket 对象进行服务端端口注册
ServerSocket ss = new ServerSocket(9999);
// 2. 监听客户端的Socket链接请求
Socket socket = ss.accept();
// 3. 从socket 管道中得到一个字节输入流对象
InputStream is = socket.getInputStream();
// 4. 把字节输入流包装成一个缓冲字符输入流
BufferedReader br = new BufferedReader(new InputStreamReader(is));
String msg;
if ((msg = br.readLine()) != null) {
System.out.println("服务端接收到:"+msg);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
小结
- 在以上通信中，服务端会一直等待客户端信息，如果客户端没有进行消息的发送，服务端将一直进入阻塞状态。
- 同时服务端是按照行获取消息的，这意味着客户端也必须按照行进行消息的发送，否则服务端将进入等待消息的阻塞状态。
2.4 BIO模式下多发和多收消息
在上面的案例只能实现客户端发送消息，服务端接收消息，并不能实现反复地接收消息和反复地发送消息，我们只需要在客户端案例中，加上反复按照行发送消息的逻辑即可。
客户端
/***
* 客户端
*/
public class Client {
public static void main(String[] args) throws IOException {
// 1.创建socket对象请求服务器链接
Socket socket = new Socket("127.0.0.1",9999);
// 2. 从socket对象中获取一个字节输出流
OutputStream os = socket.getOutputStream();
// 3. 把字节输出流包装成一个打印流
PrintStream ps = new PrintStream(os);
Scanner sc = new Scanner(System.in);
while (true){
System.out.println("请说");
String msg = sc.nextLine();
if(msg.equals("q")) break;
ps.println(msg);
ps.flush();
}
}
}
服务端
/***
* 目标: 服务端可以反复接收消息，客户端反复接收消息
*/
public class Server {
public static void main(String[] args) {
try {
// 1. 定义一个ServerSocket 对象进行服务端端口注册
ServerSocket ss = new ServerSocket(9999);
// 2. 监听客户端的Socket链接请求
Socket socket = ss.accept();
// 3. 从socket 管道中得到一个字节输入流对象
InputStream is = socket.getInputStream();
// 4. 把字节输入流包装成一个缓冲字符输入流
BufferedReader br = new BufferedReader(new InputStreamReader(is));
String msg;
while ((msg = br.readLine()) != null) {
System.out.println("服务端接收到:"+msg);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
2.5 BIO模式下接收多个客户端
概述
在上述案例中，一个服务端只能接收一个客户端的请求，那么如果服务端需要处理多个客户端的消息通信请求应该如何处理呢，此时我们就需要在服务端引入线程，即客户端每发起一个请求，服务端就创建一个新的线程来处理这个客户端的请求，这样就实现了一个客户端一个线程的模型。
客户端
public class Client {
public static void main(String[] args) {
try {
// 1. 请求与服务端的socket对象连接
Socket socket = new Socket("127.0.0.1",9999);
// 2. 得到一个打印流
PrintStream ps = new PrintStream(socket.getOutputStream());
//3. 使用循环不断发送消息给服务端接收
Scanner sc = new Scanner(System.in);
while (true){
System.out.println("请说:");
String s = sc.nextLine();
ps.println(s);
ps.flush();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
服务端
public class Server {
public static void main(String[] args) {
try {
// 1.注册端口
ServerSocket ss = new ServerSocket(9999);
//2. 定义一个死循环，负责不断接收客户端的socket的连接请求
while (true){
Socket socket = ss.accept();
// 3. 创建一个独立的线程来处理与这个客户端的socket通信需求
new ServerThreadReader(socket).start();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
线程类
public class ServerThreadReader extends Thread{
private Socket socket;
public ServerThreadReader(Socket socket){
this.socket = socket;
}
@Override
public void run() {
try {
//从socket对象得到一个字节输入流
InputStream is = socket.getInputStream();
//使用缓存字符输入流包装字节输入流
BufferedReader br = new BufferedReader(new InputStreamReader(is));
String msg;
while ((msg = br.readLine())!=null){
System.out.println(msg);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
小结
- 每个socket接收到，都会创建一个线程，线程的竞争、切换上下文影响性能；
- 每个线程都会占用栈空间和CPU资源;
- 客户端的并发访问增加，服务端将呈现1：1的线程开销，访问量越大，系统将发生线程溢出，线程创建失败，最终导致进程宕机或者僵死，从而不能对外提供服务。
2.6 伪异步I/O编程
概述
在上述案例中，客户端的并发访问增加时，服务端将呈现1：1的线程开销，访问量越大，系统将发生线程栈溢出，线程创建失败，最终导致进程宕机或者僵死，从而不能对外提供服务。
接下来我们采用一个伪异步I/O的通信框架，采用线程池和任务队列实现，当客户端接入时，将客户端的socket封装成一个Task交给后端的线程池进行处理。JDK的线程池维护一个消息队列和N个活跃的线程，对消息队列中Socket任务进行处理，由于线程池可以设置消息队列的大小和最大线程数，因此，它的资源占用是可控的，无论是多少个客户端并发访问，都不会导致资源的耗尽和宕机。
客户端
public class Client {
public static void main(String[] args) {
try {
// 1. 请求与服务端的socket对象连接
Socket socket = new Socket("127.0.0.1",9999);
// 2. 得到一个打印流
PrintStream ps = new PrintStream(socket.getOutputStream());
//3. 使用循环不断发送消息给客户端接收
Scanner sc = new Scanner(System.in);
while (true){
System.out.println("请说:");
String msg = sc.nextLine();
ps.println(msg);
ps.flush();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
线程池处理类
public class HandlerSocketServerPool {
// 1.创建一个线程池的成员变量用于存储一个线程池对象
private ExecutorService executorService;
public HandlerSocketServerPool(int maxThreadNum,int queueSize){
executorService = new ThreadPoolExecutor(3,maxThreadNum,120, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(queueSize));
}
public void execute(Runnable target){
executorService.execute(target);
}
}
socket任务类
public class ServerRunnableTarget implements Runnable{
private Socket socket;
public ServerRunnableTarget(Socket socket){
this.socket = socket;
}
@Override
public void run() {
// 处理接收到的客户端socket通信需求
try {
// 1. 从socket管道得到一个字节输入流对象
InputStream is = socket.getInputStream();
// 2. 把字节输入流包装成一个缓存字符输入流
BufferedReader br = new BufferedReader((new InputStreamReader(is)));
String msg;
while ((msg = br.readLine())!=null){
System.out.println("服务端收到:"+msg);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
服务端
public class Server {
public static void main(String[] args) {
try {
//1. 注册端口
ServerSocket ss = new ServerSocket(9999);
// 2。定义一个死循环，负责不断接收客户端的socket的连接请求
HandlerSocketServerPool pool = new HandlerSocketServerPool(3,10);
while (true){
Socket socket = ss.accept();
//3. 把socket对象交给线程池处理
Runnable target = new ServerRunnableTarget(socket);
pool.execute(target);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
小结
- 伪异步采用了线程池实现，因此避免了为每个请求创建一个独立线程造成线程资源耗尽的问题，但由于底层依然是采用的同步阻塞模型，因此无法从根上解决问题。
- 如果单个消息处理很慢，或者服务器线程池中的全部线程都被阻塞，那么后续的socket的I/O消息都将在队列中排队。新的socket请求将被拒绝，客户端会发生大量连接超时。
2.7 基于BIO形式下的文件上传
支持任意类型文件形式的上传
客户端
public class Client {
public static void main(String[] args) {
try {
FileInputStream is = new FileInputStream("C:\\Users\\liyuqiu\\Desktop\\data\\1.png");
//1. 请求与服务端的socket连接
Socket socket = new Socket("127.0.0.1", 8888);
//2. 把字节输出流包装成一个数据输出流
DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
// 3.先发送上传文件的后缀给服务器
dos.writeUTF(".png");
//4. 把文件数据发送给服务端进行接收
byte[] buffer = new byte[1024];
int len;
while((len = is.read(buffer)) > 0){
dos.write(buffer,0,len);
}
dos.flush();
socket.shutdownOutput();
} catch (IOException e) {
e.printStackTrace();
}
}
}
服务端
public class Server {
public static void main(String[] args) {
try {
ServerSocket ss = new ServerSocket(8888);
while (true){
Socket socket = ss.accept();
new ServerReadThread(socket).start();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
线程类
public class ServerReadThread extends Thread{
private Socket socket;
public ServerReadThread(Socket socket){
this.socket = socket;
}
@Override
public void run() {
try {
//1. 得到一个数据输入流来读取客户端发送过来的数据
DataInputStream dis = new DataInputStream(socket.getInputStream());
//2. 读取客户端发送过来的文件类型
String suffix = dis.readUTF();
System.out.println("服务端已经成功接收到了文件类型:"+suffix);
//3.定义一个字节输出管道，负责把客户端发来的文件数据写出去
FileOutputStream fis = new FileOutputStream("D:\\" + UUID.randomUUID().toString() + suffix);
//4. 从数据输入流中读取文件数据，写到字节输出流中
byte[] buffer = new byte[1024];
int len;
while ((len = dis.read(buffer)) >0){
fis.write(buffer,0,len);
}
fis.close();
System.out.println("服务器保存成功!");
} catch (IOException e) {
e.printStackTrace();
}
}
}
小结
- 同步阻塞模式下，客户端怎么发，服务端就怎么接收，如客户端使用的是DataOutputStream,那么服务端就该用DataInputStream,客户端dos.writeUTF(“.jpg”);服务端就该String suffix = dis.readUTF();
- 客户端发完数据后必须通知服务端自己发完socket.shutdownOutput(),否则服务端会一直等待。
2.8 Java BIO 模式下的端口转发思想
需求：需要实现一个客户端的消息可以发送到所有的客户端去接收
客户端
public class Client {
public static void main(String[] args) {
try {
// 1.请求与服务端的socket对象连接
Socket socket = new Socket("127.0.0.1", 8888);
// 2. 接收消息
new ClientReaderThread(socket).start();
while(true){
// 发送消息
OutputStream os = socket.getOutputStream();
PrintStream ps = new PrintStream(os);
//3. 使用循环不断地发送消息给服务端接收
Scanner sc = new Scanner(System.in);
String msg = sc.nextLine();
ps.println(msg);
ps.flush();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
客户端线程类
public class ClientReaderThread extends Thread{
private Socket socket;
public ClientReaderThread(Socket socket){
this.socket = socket;
}
@Override
public void run() {
try {
InputStream is = socket.getInputStream();
// 把字节流输入流转为缓存字符输入流
BufferedReader br = new BufferedReader(new InputStreamReader(is));
String msg;
while ((msg = br.readLine()) != null){
System.out.println(msg);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
服务端
public class Server {
public static List<Socket> allSocketOnline = new ArrayList<>();
public static void main(String[] args) {
try {
ServerSocket ss = new ServerSocket(8888);
while (true){
Socket socket = ss.accept();
//把登录的客户端socket存入一个在线集合中
allSocketOnline.add(socket);
// 为当前登录成功的socket分配一个独立的线程来处理与之通信
new ServerReaderThread(socket).start();
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
服务端线程类
public class ServerReaderThread extends Thread {
private Socket socket;
public ServerReaderThread(Socket socket){
this.socket = socket;
}
@Override
public void run() {
try {
// 1.从socket中去获取当前客户端的输入流
BufferedReader br = new BufferedReader(new InputStreamReader(socket.getInputStream()));
String msg;
while ((msg = br.readLine()) !=null){
System.out.println("服务端接收消息:"+msg);
sendMsgToAllClient(msg,socket);
}
} catch (IOException e) {
e.printStackTrace();
System.out.println("当前有人下线了!");
Server.allSocketOnline.remove(socket);
}
}
private void sendMsgToAllClient(String msg,Socket socket) throws IOException {
for(Socket sk : Server.allSocketOnline){
//只发送给自己以外的其他客户端
if(socket !=sk){
PrintStream printStream = new PrintStream(sk.getOutputStream());
printStream.println(msg);
printStream.flush();
}
}
}
}
3. NIO深入剖析
3.1 基本介绍
- java NIO 也有人称之为java non-blocking IO是java1,4版本开始引入的一个新的IO API，可以代替标准的java IO API。NIO与原来的IO有同样的作用和目的，但是使用的方式完全不同，NIO支持面向缓冲区的、基于通道的IO操作。NIO将以更加高效的方式进行文件的读写操作。NIO可以理解为非阻塞IO，传统的IO的read和write只能阻塞执行，线程在读写期间不能干其他事情，比如调用socket.read()时，如果服务器一直没有数据传输过来，线程就一直阻塞，而NIO可以理解为非阻塞IO，传统的IO的read和write只能阻塞执行，线程在读写期间不能干其他事情，比如调用socket.read()时，如果服务器一直没有数据传输过来，线程就一直阻塞，而NIO中可以配置socket为非阻塞模式。
- NIO相关类都被放在java.nio包及子包下，并且对原java.io包中很多类进行改写
- NIO有三大核心部分：Channel(通道)，Buffer(缓冲区),Selectot(选择器)
- Java NIO 的非阻塞模式，使一个线程从某通道发送请求或者读取数据，但是它仅能得到目前可用的数据；如果目前没有数据可用时，就什么都不会获取，而不是保持线程阻塞，所以直到数据变得可以读取之前，但不需要等待它完全写入，这个线程同时可以去做别得事情。
- 通俗理解：NIO是可以做到用一个线程来处理多个操作，假设有10000个请求过来，根据实际情况，可以分配20或80个线程来处理。
3.2 NIO和BIO的比较
- BIO以流的方式处理数据，而NIO以块的方式处理数据，块I/O的效率比流I/O高很多
- BIO是阻塞的，NIO是非阻塞的
- BIO基于字节流和字符流进行操作，而NIO基于Channel(通道)和Buffer(缓冲区)进行操作，数据总是从通道读取到缓冲区中，或者从缓冲区写入到通道中。Selcetor(选择器)用于监听多个通道的事件，因此使用单个线程就可以监听多个客户端通道。
3.3 NIO三大核心原理示意图
NIO有三大核心部分: Channel(通道)、Buffer(缓冲区)、Selector（选择器）
Buffer缓存区
缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象，并提供一组方法，用来方便访问该块内存，相比较直接对数组的操作，Buffer API更加容易操作和管理。
Channel通道
Java NIO 的通道类似流，但又有些不同：既可以从通道中读取数据，又可以写数据到通道中。但流的读写通常是单向的。通道可以非阻塞读取和写入通道，通道可以支持读取或写入缓冲区，也支持异步读写。
Selector选择器
Selector是一个java NIO组件，可以检查一个或多个NIO通道，并确定哪些通道已经准备好进行读取或写入。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接，提高效率。
- 每个channel都会对应一个Buffer
- 一个线程对应Selector,一个Selector对应多个channel（连接）
- 程序切换到哪个channel是由事件决定的
- Selector会根据不同的事件，在各个通道上切换
- Buffer就是一个内存块，底层是一个数组
- 数据的读取写入时通过Buffer完成的，BIO要么是输入流，或者输出流，不能双向，但是NIO的buffer是可以读也可以写的。
- Java NIO 系统的在于：通道（channel)和缓冲区（Buffer)。通道表示打开到IO设备的连接。若需要使用NIO系统，需要获取用于连接IO设备的通道以及用于容纳数据的缓冲区。然后操作缓冲区，对数据进行处理。简而言之，Channel负责传输，Buffer负责存取数据。
3.4 NIO核心一：缓存区
缓冲区
一个用于特定基本数据类型的容器。由java.nio包定义的，所有缓冲区都是Buffer抽象类的子类。Java NIO 中的Buffer主要用于与NIO通道进行交互，数据从通道读入缓冲区，从缓冲区写入通道中。
Buffer类及其子类
Buffer就像一个数组，可以保存多个相同类型的数据。根据数据类型不同，有以下Buffer常用子类:
- ByteBuffer
- CharBuffer
- ShortBuffer
- IntBuffer
- LongBuffer
- FloatBuffer
- DoubleBuffer
上述Buffer类 他们都采用相似的方法进行管理数据，只是各自管理的数据类型不同而已。都是通过如下方法获取一个Buffer对象:
static XxxBuffer allocate(int capacity):创建一个容量为capacity的XxxBuffer对象
缓存区的基本属性
Buffer中的重要概念：
-
容量（capacity):作为一个内存块，Buffer具有一定的固定大小，也称为“容量”，缓冲区不能为负，并且创建后不能更改
-
限制: 表示缓冲区可以操作数据的大小（limit后数据不能进行读写）。缓冲区的限制不能为负，并且不能大于其容量。写入模式，限制等于buffer的容量。读取模式下，limit等于写入的数据量。
-
位置:下一个要读取或写入的数据，缓冲区的位置不能为负，并且不能大于其限制
-
标记与重置：标记一个索引，通过buffer的mark()方法指定Buffer中一个特定的position，之后可以通过调用reset()方法恢复到这个position。标记、位置、限制、容量遵守以T不变式:0<=mark<=position<=limit<=capacity
Buffer常见方法
Buffer clear(）清空缓冲区并返回对缓冲区的引用 Buffer flip(）为将缓冲区的界限设置为当前位置，并将当前位置重置为0 int capacity(）返回Buffer的capacity大小 boolean hasRemaining(）判断缓冲区中是否还有元素 int limit(）返回Buffer的界限（limit）的位置 Buffer limit(int n）将设置缓冲区界限为n，并返回一个具有新limit的缓冲区对象 Buffer mark(）对缓冲区设置标记 int position(）返回缓冲区的当前位置position Buffer position(int n）将设置缓冲区的当前位置为n，并返回修改后的Buffer对象 int remaining()返回position和limit之间的元素个数 Buffer reset(）将位置position转到以前设置的mark所在的位置 Buffer rewind();将位置设为为0．取消设置的mark缓冲区的数据操作
Buffer 所有子类提供了两个用于数据操作的方法：get() put()方法 取获取Buffer中的数据 get()：读取单个字节 get (byte［〕dst)：批量读取多个字节到dst中 get(int index)：读取指定索引位置的字节（不会移动position) 放到 入数据到Buffer中 put(byte b):将给定单个字节写入缓冲区的当前位置 put (byte[] src):将src中的字节写入缓存区的当前位置 put(int index,byte b)L将指定字节写入缓存区的索引位置（不会移动position）使用Buffer读写数据一般遵守以下四个步骤:
- 写入数据到Buffer
- 调用flip()方法，转换为读取模式
- 从buffer中读取数据
- 调用buffer.clear()方法或则buffer.compact()方法清除缓存区
直接与非直接缓存区
什么是直接内存与非直接内存
根据官方描述：
byte buffer。可以是两种类型，一种基于直接内存，另一种是非直接内存（也就是堆内存）。对于直接内存来说，JVM将会在IO操作上居于更高的性能，因为它直接作用于本地系统的IO操作。而非直接内存。对于直接内存来说，JVM将会在IO操作上具有更高的性能，因为它直接作用于本地系统的IO操作。而非直接内存，也就是堆内存的数据，如果要做IO操作，会先从本进程内存复制带直接内存，再利用本地IO处理。
从数据流的角度，非直接内存是下面这样的作用链
本地IO ---> 直接内存 ---> 非直接内存 --->直接内存 --->本地IO而直接内存是:
本地IO ---> 直接内存 ---> 本地IO很明显，在做IO处理时，比如网络发送大量数据时，直接内存会具有更高的效率。直接内存使用allocateDirect创建，但是它比申请普通的堆内存需要耗费更高的性能。不过，这部分的数据是在JVM之外，因此它不会占用应用内存。所以呢，当你有很大的数据要缓存，并且它的生命周期又很长，那么就比较适合使用直接内存。只是一般来说，如果不是能带来很明显的性能提升，还是推荐直接使用堆内存。字节缓冲区是直接缓冲区还是非直接缓冲区，可通过调用isDirect()方法来确定。
public class BufferTest03 { public static void main(String[] args) { ByteBuffer buffer = ByteBuffer.allocate(1024); System.out.println(buffer.isDirect()); System.out.println("--------------"); ByteBuffer buffer2 = ByteBuffer.allocateDirect(1024); System.out.println(buffer2.isDirect()); } }使用场景
-
有很大的数据需要存储，他的生命周期又很长
-
适合频繁的IO操作，比如网络并发场景
3.5 NIO核心二:通道
通道：由java.nio.channels包定义的。Channel 表示IO源与目标打开的连接。Channel类似于传统的“流”。只不过Channel本身不能直接访问数据，Channel只能与Buffer进行交互。
-
NIO的通道类类似于流，但有些区别如下:
- 通道可以同时进行读写，而流只能读或者只能写
- 通道可以实现异步读写数据
- 通道可以从缓冲读数据，也可以写数据到缓冲
-
BIO的stream是单向的，例如FileInputStream对象只能进行读取数据的操作，而NIO中的通道是双向的，可以读操作，也可以写操作。
-
Channel在NIO中是一个接口
public interface Channel extends Closeable()常用的channel实现类
- FileChannel: 用于读取、写入、映射和操作文件的通道
- DatagramChannel: 通过UDP读写网络中的数据通道
- SocketChannel: 通过TCP读写网络中数据
- ServerSocketChannel: 可以监听新进来的TCP连接，对每一个新进来的连接都会创建一个SocketChannel
FileChannel类
获取通道的一种方式是对支持通道的对象调用getChannel方法，支持通道的类如下:
- FileInputStream
- FileOutputStream
- RandomAccessFile
- DatagramSocket
- Socket
- ServerSocket
获取通道的其他方式是使用Files类的静态方法newByteChannel()获取字节通道，或通过通道的静态方法open（）打开并返回指定通道。
FileChannel的常用方法
int read(ByteBuffer dst) 从Channel当中读取数据至ByteBuffer
long read(ByteBuffer[] dsts）将channel当中的数据“分散”至ByteBuffer[]
int write(Bytesuffer src)将ByteBuffer当中的数据写入到Channel
long write(ByteBuffer[] srcs）将Bytesuffer[]当中的数据“聚集”到Channel
long position(）返回此通道的文件位置
FileChannel position(long p）设置此通道的文件位置
long size(）返回此通道的文件的当前大小
FileChannel truncate(long s）将此通道的文件截取为给定大小
void force(boolean metaData）强制将所有对此通道的文件更新写入到存储设备中
案例1-本地文件写数据
public class ChanellTest {
public static void main(String[] args) throws IOException {
//1. 字节输出流通向目标文件
FileOutputStream fos = new FileOutputStream("data01_txt");
//2.得到字节输出流对应的通道Channel
FileChannel channel = fos.getChannel();
//3. 分配缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put("hello,i am liyuqiu!".getBytes());
// 4. 把缓冲区切换为写模式
buffer.flip();
channel.write(buffer);
channel.close();
System.out.println("写数据到文件中!");
}
}
案例2-本地文件读数据
public class ChannelTestTwo {
public static void main(String[] args) throws IOException {
//1, 定义一个文件字节输入流与源文件接通
FileInputStream is = new FileInputStream("data01_txt");
//2. 需要得到文件字节输入流的文件通道
FileChannel channel = is.getChannel();
//3. 定义一个缓存区
ByteBuffer buffer = ByteBuffer.allocate(1024);
//4. 读取数据到缓存区
channel.read(buffer);
buffer.flip();
// 5. 读取出缓存区的数据并输出即可
String s = new String(buffer.array(), 0, buffer.remaining());
System.out.println(s);
}
}
案例3-使用Buffer完成文件复制
public class ChannelTestThree {
public static void main(String[] args) throws IOException {
//源文件
File srcFile = new File("D:\\1.pdf");
File destFile = new File("D:\\data\\2.pdf");
// 得到一个字节输出流、字节输入流
FileInputStream fis = new FileInputStream(srcFile);
FileOutputStream fos = new FileOutputStream(destFile);
//得到文件通道
FileChannel fisChannel = fis.getChannel();
FileChannel fosChannel = fos.getChannel();
//分配缓存区
ByteBuffer buffer = ByteBuffer.allocate(1024);
while (true){
//必须先清空缓存区然后再写入数据缓存区
buffer.clear();
//开始读取一次数据
int flag = fisChannel.read(buffer);
//开始读取一次数据
if (flag == -1){
break;
}
//已经读取了数据，把缓存区的模式切换为可读模式
buffer.flip();
// 把数据写出到
fosChannel.write(buffer);
}
fisChannel.close();
fosChannel.close();
System.out.println("复制完成");
}
}
案例4-分散和聚集
分散读取是指把Channel通道的数据读取到多个缓存区中去
聚集写入是指把多个Buffer中的数据聚集到Channel。
public class test {
public static void main(String[] args) throws IOException {
//1. 字节输入管道
FileInputStream is = new FileInputStream("data01_txt");
FileChannel isChannel = is.getChannel();
//2.字节输出管道
FileOutputStream os = new FileOutputStream("data02_txt");
FileChannel osChannel = os.getChannel();
//3. 定义多个缓存区做数据分散
ByteBuffer buffer1 = ByteBuffer.allocate(4);
ByteBuffer buffer2 = ByteBuffer.allocate(1024);
ByteBuffer[] buffers = {buffer1, buffer2};
//4. 从通道中读取数据分散到各个缓存区
isChannel.read(buffers);
//5. 从美国缓存区中查询是否由数据读取到
for (ByteBuffer buffer : buffers){
buffer.flip();
System.out.println(new String(buffer.array(),0,buffer.remaining()));
}
//6. 聚集写入到通道
osChannel.write(buffers);
isChannel.close();
osChannel.close();
}
}
transferFrom()
从目标通道中复制原通道数据
public class test2 {
public static void main(String[] args) throws IOException {
//1. 字节输入管道
FileInputStream is = new FileInputStream("data01_txt");
FileChannel isChannel = is.getChannel();
//2. 字节输出管道
FileOutputStream os = new FileOutputStream("data03_txt");
FileChannel osChannel = os.getChannel();
//3.复制数据
osChannel.transferFrom(isChannel, isChannel.position(),isChannel.size());
isChannel.close();
osChannel.close();
System.out.println("复制完成");
}
}
transferTo()
把原通道数据复制到目标通道
public class test3 {
public static void main(String[] args) throws IOException {
//1.字节输入管道
FileInputStream is = new FileInputStream("data01_txt");
FileChannel isChannel = is.getChannel();
//2.字节输出管道
FileOutputStream os = new FileOutputStream("data04_txt");
FileChannel osChannel = os.getChannel();
//3.复制数据
isChannel.transferTo(isChannel.position(), isChannel.size(),osChannel);
isChannel.close();
osChannel.close();
System.out.println("复制完成!");
}
}
3.6 NIO核心三：选择器
选择器是SelectableChannel 对象的多路复用器，Selector可以同时监控多个SelectableChannel的IO状况，也就是说selector可使一个单独的线程管理多个Channel。Selector是非阻塞IO的核心。
- java的NIO，用非阻塞的IO方式，可以用一个线程，处理多个客户端连接，就会使用到Selector（选择器）
- Selector能够检测多个注册的通道上是否有事件发生，如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道，也就是管理多个连接和请求。
- 只有在连接/通道真正有读写事件发生时，才会进行读写，就大大减少了系统开销，并且不必为每个连接都创建一个线程，不用维护多个线程，避免了多线程之间的上下文切换导致的开销。
选择器的应用
创建Selector：通过Selector.open()方法创建一个Selector。
Selector selector = Selector.open();
向选择器注册通道: SelectableChannel.register(Selector sel,int ops)
// 1.获取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
//2, 切换非阻塞模式
ssChannel.configureBlocking(false);
//3.绑定连接
ssChannel.bind(new InetSocketAddress(9898));
//4. 获取选择器
Selector selector = Selector.open();
//5. 将通道注册到选择器上，并且指定"监听接收事件"
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
当调用register(Selector sel,mt ops)将通道注册选择器时，选择器对通道的监听事件，需要通过第二个参数。ops可以指定监听事件类型
- 读: SelectionKey.OP_READ
- 写:SelectionKey.OP_WRITE
- 连接: SelectionKeyOP_CONNECT
- 接收:SelectionKey.OP_ACCEPT
- 若注册时不止监听一个事件，则可以使用“位或”操作符连接
int interest = selectionKey.OP_READ | SelectionKey.OP_WRITE
3.7 NIO非阻塞式网络通信原理分析
Selector示意图和特点说明
Selector可以实现: 一个I/O线程可以并发处理N个客户端连接和读写操作，这从根本上解决了传统同步阻塞I/O一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。
服务端流程
1.当客户端连接服务端时，服务端会通过ServerSocketChannel得到SocketChannel,即获取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
2.切换非阻塞模式
ssChannel.configureBlocking(false);
3.绑定连接
ssChannel.bind(new InetSocketAddress(9898));
4.获取选择器
Selector selector = Selector.open();
5.将通道注册到选择器上，并且指定“监听接收事件”
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
6.轮询式获取选择器上已经“准备就绪"的事件
//轮询式获取选择器上已经“准备就绪”的事件
while(selector.select() > 0){
System.out.println("轮一轮");
//7. 获取当前选择器中所有注册的“选择键”（已就绪的监听事件）
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()) {
//8.获取准备“就绪”事件
SelectionKey sk = it.next();
//9.判断具体使什么事件准备就绪
if(sk.isAcceptable()){
//10. 若“接收就绪”，获取客户端连接
SocketChannel sChannel = ssChannel.accept();
// 11.切换非阻塞模式
sChannel.configureBlocking(false);
//12. 将该通道注册到选择器上
sChannel.register(selector,SelectionKey.OP_READ);
}else if(sk.isReadable()){
//13. 获取当前选择器上“读就绪”状态的通道
SocketChannel sChannel = (SocketChannel) sk.channel();
//14. 读取数据
ByteBuffer buf = ByteBuffer.allocate(1024);
int len =0;
while ((len = sChannel.read(buf)) >0){
buf.flip();
System.out.println(new String(buf.array(), 0, len));
buf.clear();
}
}
it.remove();
}
}
客户端流程
1.获取通道
SocketChannel sChannel = SocketChannel.open(new
InetSocketAddress("127.0.0.1",9898));
2.切换非阻塞模式
sChannel.configureBlocking(false);
3.分配指定大小的缓存区
ByteBuffer buf = ByteBuffer.allocate(1024);
4.发送数据给客户端
Scanner scan = new Scanner(System.in);
while(scan.hasNext()){
String str = scan.nextLine();
buf.put((new SimpleDateFormat("yyyy/MM/dd
HH:mm:ss").format(System.currentTimeMillis()) + "\n" + str).getByte());
buf.flip();
sChannel.write(buf);
buf.clear();
}
//关闭通道
sChannel.close();
3.8 NIO非阻塞式网络通信入门案例
需求: 服务端接收客户端的连接请求，并接收多个客户端发送过来的事件
客户端
public class Client {
public static void main(String[] args) throws IOException {
//1. 获取通道
SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1",8888));
//2.切换为非阻塞模式
sChannel.configureBlocking(false);
//3.分配指定缓存区大小
ByteBuffer buf = ByteBuffer.allocate(1024);
//4.发送数据给服务端
Scanner scanner = new Scanner(System.in);
while (true){
System.out.println("请输入");
String msg = scanner.nextLine();
buf.put(msg.getBytes(StandardCharsets.UTF_8));
buf.flip();
sChannel.write(buf);
buf.clear();
}
}
}
服务端
public class Server {
public static void main(String[] args) throws IOException {
System.out.println("----服务端已启动-----");
//1.获取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
//2.切换为阻塞模式
ssChannel.configureBlocking(false);
//3.绑定连接的端口
ssChannel.bind(new InetSocketAddress(8888));
//4.获取选择器
Selector selector = Selector.open();
//5.将通道都注册到选择器上，并且开始指定监听接收事件
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
//6. 使用Selector选择器轮询已经就绪好的事件
while (selector.select() > 0){
System.out.println("开始一轮事件处理~~~~~");
//7. 获取选择器中的所有注册的通道已经就绪好的事件
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()){
SelectionKey sk = it.next();
if(sk.isAcceptable()){
//8.直接获取当前接入的客户端通道
SocketChannel schannel = ssChannel.accept();
//9.将客户端通道设置为非阻塞的
schannel.configureBlocking(false);
//10 将客户端通道也注册选择器Selector上
schannel.register(selector,SelectionKey.OP_READ);
}else if(sk.isReadable()){
//11. 获取当前选择器上的”就绪事件“
SocketChannel sChannel = (SocketChannel) sk.channel();
//12 开始读取数据
ByteBuffer buf = ByteBuffer.allocate(1024);
int len;
while ((len = sChannel.read(buf))>0){
buf.flip();
System.out.println(new String(buf.array(),0,len));
buf.clear();
}
}
it.remove();
}
}
}
}
3.9 NIO网络编程应用实例-群聊系统
需求
- 编写一个NIO群聊系统，实现客户端与客户端的通信需求（非阻塞）
- 服务端：可以检测用户上限，离线，并实现消息转发功能
- 客户端：通过channel可以无阻塞发送消息给其他所有客户端用户，同时可以接受其他客户端用户通过服务端转发来的消息。
服务端
public class Server {
private Selector selector;
private ServerSocketChannel ssChannel;
private static final int PORT = 8888;
public Server() {
try {
// 1. 创建构造器
selector = Selector.open();
// 2.获取通道
ssChannel = ServerSocketChannel.open();
// 3.切换为非阻塞模式
ssChannel.configureBlocking(false);
// 4.绑定连接的端口
ssChannel.bind(new InetSocketAddress(PORT));
// 5. 将通道都注册到选择器上去，并且开始指定监听接收事件
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
} catch (IOException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException {
//创建服务端对象
Server server = new Server();
//开始监听客户端各种消息事件：连接、群聊消息、离线消息
server.listen();
}
// 监听
public void listen() throws IOException {
while (selector.select() > 0) {
// 获取选择器在所有注册通道的就绪事件
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
// 开始遍历这个事件
while (it.hasNext()) {
SelectionKey sk = it.next();
if (sk.isAcceptable()) {
// 客户端接入请求
// 获取当前客户端通道
SocketChannel schannel = ssChannel.accept();
// 注册成非阻塞模式
schannel.configureBlocking(false);
// 注册给选择器，监听读数据的事件
schannel.register(selector, SelectionKey.OP_READ);
} else if (sk.isReadable()) {
readClientData(sk);
}
it.remove();
}
}
}
// 接收当前客户端的消息，转发给其他全部客户端通道
private void readClientData(SelectionKey sk) throws IOException {
System.out.println("进来了");
SocketChannel sChannel = null;
// 直接得到当前客户端通道
sChannel = ((SocketChannel) sk.channel());
// 创建缓存区对象，开始接收客户端通道的数据
ByteBuffer buffer = ByteBuffer.allocate(1024);
int count = sChannel.read(buffer);
if (count > 0) {
buffer.flip();
// 提取读到的信息
String msg = new String(buffer.array(), 0, buffer.remaining());
System.out.println("接收到客户端的消息:" + msg);
// 把这个消息推送给全部客户端接收
sendMsgToAllClient(msg, sChannel);
}
}
// 把当前客户端的消息推送给当前全部在线注册的channel
private void sendMsgToAllClient(String msg, SocketChannel sChannel) throws IOException {
System.out.println("服务端开始转发这个消息。当前处理的线程" + Thread.currentThread().getName());
for (SelectionKey key : selector.keys()) {
Channel channel = key.channel();
// 不要把消息发给自己
if (channel instanceof SocketChannel && channel != sChannel) {
ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes(StandardCharsets.UTF_8));
((SocketChannel) channel).write(buffer);
}
}
}
}
客户端
public class Client {
//1.定义客户端相关属性
private Selector selector;
private static int PORT = 8888;
private SocketChannel socketChannel;
//2. 初始化客户端消息
public Client() throws IOException {
//创建选择器
selector = Selector.open();
// 连接服务器
socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1",PORT));
// 设置非阻塞模式
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
System.out.println("当前客户端准备完成");
}
public static void main(String[] args) throws IOException {
Client client = new Client();
new Thread(new Runnable() {
@Override
public void run() {
try {
client.readInfo();
} catch (IOException e) {
e.printStackTrace();
}
// 发送消息
Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) {
System.out.println("----------------");
String s = scanner.nextLine();
try {
client.sendToServer(s);
} catch (IOException e) {
e.printStackTrace();
}
}
}
}).start();;
}
private void sendToServer(String s) throws IOException {
socketChannel.write(ByteBuffer.wrap(s.getBytes(StandardCharsets.UTF_8)));
}
private void readInfo() throws IOException{
while (selector.select() > 0) {
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey sk = it.next();
if (sk.isReadable()) {
SocketChannel sc = ((SocketChannel) sk.channel());
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.flip();
sc.read(buffer);
System.out.println(new String(buffer.array()).trim());
System.out.println("----------------------");
}
it.remove();
}
}
}
}
4. AIO深入剖析
-
Java AIO 是异步非阻塞的，服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程进行处理。
-
AIO是异步非阻塞，基于NIO，可以称之为NIO2.0
BIO NIO AIO Socket SocketChannel AsynchronousSocketChannel ServerSocket ServerSocketChannel AsynchronousServerSocketChannel
与NIO不同，当进行读写操作时，只需直接调用API的read或者write方法即可，这两种方法均为异步的，对于读操作而言，当有流可读时，操作系统会将可读的流传入read方法的缓冲区，对于写操作而言，当操作系统将write方法传递的流写入完毕时，操作系统主动通知应用程序。
即可理解为，read/write方法都是异步的，完成后会主动调用回调函数。在JDK1.7中，这部分内容被称作NIO.2，主要在java.nio.channel包下增加了下面四个异步通道:
- AsynchronousSocketChannel
- AsynchronousFileChannel
- AsynchronousFileChannel
- AsynchronousDatagramChannel
分类:
io
标签:
无