一、简介
NIO
一种同步非阻塞的I/O。
AIO
异步非阻塞I/O。
BIO
同步阻塞IO操作。
二、名词解释
阻塞和非阻塞
当线程执行阻塞操作时，是只能等待，而不能执行其他事情的。
非阻赛是不需要等待，直接返回，继续执行下一个操作。
同步和异步
同步异步是运行机制，当我们进行同步操作时，后续的任务是等待当前调用返回，才会进行下一步。
异步则相反，其他任务不需要等待当前调用返回，通常依靠事件、回调等机制来实现任务间次序关系。
IO分类
按操作数据分为：字节流（Reader、Writer）和字符流（InputStream、OutputStream）
按流向分：输入流（Reader、InputStream）和输出流（Writer、OutputStream）
三、实战演示
BIO
服务端代码：
import java.io.IOException;
import java.io.InputStream;
import java.net.ServerSocket;
import java.net.Socket;
/**
* @ProjectName: onereader
* @Package: com.onereader.webblog.common.bio
* @ClassName: BIOServer
* @Author: onereader
* @Description: ${description}
* @Date: 2019/9/1 14:30
* @Version: 1.0
*/
public class BIOServer {
ServerSocket server;
//服务器
public BIOServer(int port){
try {
//把Socket服务端启动
server = new ServerSocket(port);
System.out.println("BIO服务已启动，监听端口是：" + port);
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 开始监听，并处理逻辑
* @throws IOException
*/
public void listener() throws IOException{
//死循环监听
while(true){
//等待客户端连接，阻塞方法
Socket client = server.accept();
//获取输入流
InputStream is = client.getInputStream();
//定义数组，接收字节流
byte [] buff = new byte[1024];
int len = is.read(buff);
//只要一直有数据写入，len就会一直大于0
if(len > 0){
String msg = new String(buff,0,len);
System.out.println("收到" + msg);
}
}
}
public static void main(String[] args) throws IOException {
new BIOServer(9999).listener();
}
}
客户端代码：
import java.io.OutputStream;
import java.net.Socket;
import java.util.concurrent.CountDownLatch;
/**
* @ProjectName: onereader
* @Package: com.onereader.webblog.common.bio
* @ClassName: BIOClient
* @Author: onereader
* @Description: ${description}
* @Date: 2019/9/1 14:31
* @Version: 1.0
*/
public class BIOClient {
public static void main(String[] args){
int count = 10;
//计数器，模拟10个线程
final CountDownLatch latch = new CountDownLatch(count);
for(int i = 0 ; i < count; i ++){
new Thread(){
@Override
public void run() {
try{
//等待，保证线程一个一个创建连接
latch.await();
//创建socket,连接服务端
Socket client = new Socket("localhost", 9999);
//获取输出流
OutputStream os = client.getOutputStream();
//获取当前线程名
String name = "客户端线程："+Thread.currentThread().getName();
//发送到服务端
os.write(name.getBytes());
//关闭输入流
os.close();
//关闭socket连接
client.close();
}catch(Exception e){
}
}
}.start();
//计数器减1
latch.countDown();
}
}
}
结果：
收到客户端线程：Thread-1
收到客户端线程：Thread-4
收到客户端线程：Thread-0
收到客户端线程：Thread-3
收到客户端线程：Thread-2
收到客户端线程：Thread-6
收到客户端线程：Thread-8
收到客户端线程：Thread-7
收到客户端线程：Thread-5
收到客户端线程：Thread-9
简单来说明以下，启动服务端后，服务端就会一直处于阻塞等待状态，等待客户端连接后，才能继续执行读取客户端发送的信息，然后在进入循环阻塞，等待新的连接处理新的信息数据。上面，我们创建了10个线程模拟是个客户端连接服务端发送信息，服务端在循环阻塞方式，接收到客户端传过来的信息。
NIO
NIO的三个主要组成部分：
Channel（通道）、Buffer（缓冲区）、Selector（选择器）
NIO执行流程：
当客户端在使用NIO连接到服务端，先会打开Channel（通道），然后把数据写入到Buffer（缓冲区），并把Buffer数据通过通道发送，最后向选择器中注册一个事件。
服务端则会先打开Channel（通道），然后监听选择器中的事件，如果发现有事件后，就从渠道中获取Buffer，然后从Buffer中读取数据。如果没有，则一直循环监听下去。
Channel（通道）
Channel是一个对象，可以通过它读取和写入数据。
Channel是双向的，既可以读又可以写。
Channel可以进行异步的读写。
Channel的读写必须通过buffer对象。
在Java NIO中的Channel主要有如下几种类型：
FileChannel：从文件读取数据的
DatagramChannel：读写UDP网络协议数据
SocketChannel：读写TCP网络协议数据
ServerSocketChannel：可以监听TCP连接
Buffer（缓存区）
Buffer是一个对象，它包含一些要写入或者读到Stream对象的。应用程序不能直接对 Channel 进行读写操作，而必须通过 Buffer 来进行，即 Channel 是通过 Buffer 来读写数据的。
在NIO中，所有的数据都是用Buffer处理的，它是NIO读写数据的中转池。Buffer实质上是一个数组，通常是一个字节数据，但也可以是其他类型的数组。但一个缓冲区不仅仅是一个数组，重要的是它提供了对数据的结构化访问，而且还可以跟踪系统的读写进程。
Buffer属性介绍：
容量（Capacity）：缓冲区能够容纳的数据元素的最大数量。这一个容量在缓冲区创建时被设定，并且永远不能改变。
上界(Limit)：缓冲区的第一个不能被读或写的元素。或者说,缓冲区中现存元素的计数。
位置(Position)：下一个要被读或写的元素的索引。位置会自动由相应的 get( )和 put( )函数更新。
标记(Mark)：下一个要被读或写的元素的索引。位置会自动由相应的 get( )和 put( )函数更新。
读写数据步骤：
1.写入数据到 Buffer；
2.调用 flip() 方法；
3.从 Buffer 中读取数据；
4.调用 clear() 方法或者 compact() 方法。
当向 Buffer 写入数据时，Buffer 会记录下写了多少数据。一旦要读取数据，需要通过 flip() 方法将 Buffer 从写模式切换到读模式。在读模式下，可以读取之前写入到 Buffer 的所有数据。
一旦读完了所有的数据，就需要清空缓冲区，让它可以再次被写入。有两种方式能清空缓冲区：调用 clear() 或 compact() 方法。clear() 方法会清空整个缓冲区。compact() 方法只会清除已经读过的数据。任何未读的数据都被移到缓冲区的起始处，新写入的数据将放到缓冲区未读数据的后面。
Buffer主要有如下几种：
ByteBuffer
CharBuffer
DoubleBuffer
FloatBuffer
IntBuffer
LongBuffer
ShortBuffer
Selector（选择器）
无论客户端还是服务端都可以向Selector中注册事件，然后监听事件，最后根据不同的事件做不同的处理。
下面看下代码
客户端：
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.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
/**
* @ProjectName: onereader
* @Package: com.onereader.webblog.common.nio
* @ClassName: NIOClient
* @Author: onereader
* @Description: ${description}
* @Date: 2019/9/1 13:41
* @Version: 1.0
*/
public class NIOClient {
private final InetSocketAddress serverAdrress = new InetSocketAddress("localhost", 9999);
private Selector selector = null;
private SocketChannel client = null;
private Charset charset = Charset.forName("UTF-8");
public NIOClient() throws IOException{
//1.连接远程主机的IP和端口
client = SocketChannel.open(serverAdrress);
client.configureBlocking(false);
//打开选择器，注册读事件
selector = Selector.open();
client.register(selector, SelectionKey.OP_READ);
}
public void session(){
//开辟一个新线程从服务器端读数据
new Reader().start();
//开辟一个新线程往服务器端写数据
new Writer().start();
}
/**
* 写数据线程
*/
private class Writer extends Thread{
@Override
public void run() {
try{
//在主线程中 从键盘读取数据输入到服务器端
Scanner scan = new Scanner(System.in);
while(scan.hasNextLine()){
String line = scan.nextLine();
if("".equals(line)){
//不允许发空消息
continue;
}
//当前渠道是共用的，发送当前输入数据
client.write(charset.encode(line));
}
scan.close();
}catch(Exception e){
}
}
}
/**
* 读数据线程
*/
private class Reader extends Thread {
@Override
public void run() {
try {
//循环检测
while(true) {
int readyChannels = selector.select();
if(readyChannels == 0){
continue;
}
//获取selected所有的事件
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
while(keyIterator.hasNext()) {
SelectionKey key = (SelectionKey) keyIterator.next();
keyIterator.remove();
process(key);
}
}
}
catch (IOException io){
}
}
/**
* 根据事件的不同，做不同的处理
* @param key
* @throws IOException
*/
private void process(SelectionKey key) throws IOException {
//读就绪事件
if(key.isReadable()){
//通过key找到对应的通道
SocketChannel sc = (SocketChannel)key.channel();
//创建缓存区
ByteBuffer buff = ByteBuffer.allocate(1024);
String content = "";
//读数据
while(sc.read(buff) > 0){
buff.flip();
content += charset.decode(buff);
}
//打印内容
System.out.println(content);
//设置当前为读就绪
key.interestOps(SelectionKey.OP_READ);
}
}
}
public static void main(String[] args) throws IOException {
new NIOClient().session();
}
}
服务端：
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;
/**
* @ProjectName: onereader
* @Package: com.onereader.webblog.common.nio
* @ClassName: NIOServer
* @Author: onereader
* @Description: ${description}
* @Date: 2019/9/1 13:23
* @Version: 1.0
*/
public class NIOServer {
private int port = 9999;
private Charset charset = Charset.forName("UTF-8");
private Selector selector = null;
public NIOServer(int port) throws IOException{
this.port = port;
//1.打开通道
ServerSocketChannel server = ServerSocketChannel.open();
//设置服务端口
server.bind(new InetSocketAddress(this.port));
server.configureBlocking(false);
//2.打开选择器
selector = Selector.open();
//注册等待事件
server.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("服务已启动，监听端口是：" + this.port);
}
/**
* 监听事件
* @throws IOException
*/
public void listener() throws IOException{
//死循环，这里不会阻塞
while(true) {
//1.在轮询获取待处理的事件
int wait = selector.select();
System.out.println("当前等待处理的事件："+wait+"个");
if(wait == 0){
//如果没有可处理的事件，则跳过
continue;
}
//获取所有待处理的事件
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> iterator = keys.iterator();
//遍历
while(iterator.hasNext()) {
SelectionKey key = (SelectionKey) iterator.next();
//处理前，关闭选在择器中的事件
iterator.remove();
//处理事件
process(key);
System.out.println("事件Readable："+key.isReadable());
System.out.println("事件Acceptable："+key.isAcceptable());
}
}
}
/**
* 根据事件类型，做处理
* @param key
* @throws IOException
*/
public void process(SelectionKey key) throws IOException {
//连接就绪
if(key.isAcceptable()){
//获取通道
ServerSocketChannel server = (ServerSocketChannel)key.channel();
//进入服务端等待
SocketChannel client = server.accept();
//非阻塞模式
client.configureBlocking(false);
//注册选择器，并设置为读取模式，收到一个连接请求，
// 然后起一个SocketChannel，并注册到selector上，
// 之后这个连接的数据，就由这个SocketChannel处理
client.register(selector, SelectionKey.OP_READ);
//将此对应的channel设置为准备接受其他客户端请求
key.interestOps(SelectionKey.OP_ACCEPT);
client.write(charset.encode("来自服务端的慰问"));
}
//读就绪
if(key.isReadable()){
//返回该SelectionKey对应的 Channel，其中有数据需要读取
SocketChannel client = (SocketChannel)key.channel();
//往缓冲区读数据
ByteBuffer buff = ByteBuffer.allocate(1024);
StringBuilder content = new StringBuilder();
try{
while(client.read(buff) > 0){
buff.flip();
content.append(charset.decode(buff));
}
System.out.println("接收到客户端："+content.toString());
//将此对应的channel设置为准备下一次接受数据
key.interestOps(SelectionKey.OP_READ);
}catch (IOException io){
key.cancel();
if(key.channel() != null){
key.channel().close();
}
}
}
}
public static void main(String[] args) throws IOException {
new NIOServer(9999).listener();
}
}
结果：
客户端：
来自服务端的慰问
请输入要发送服务端的信息：
你好呀
服务端：
服务已启动，监听端口是：9999
当前等待处理的事件：1个
事件Readable：false
事件Acceptable：true
当前等待处理的事件：1个
接收到客户端：你好呀
事件Readable：true
事件Acceptable：false
这里配合上面的基础知识介绍，然后看代码就能明白。整个过程了~
AIO
客户端：
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;
/**
* @ProjectName: onereader
* @Package: com.onereader.webblog.common.aio
* @ClassName: AIOClient
* @Author: onereader
* @Description: ${description}
* @Date: 2019/9/1 14:03
* @Version: 1.0
*/
public class AIOClient {
private final AsynchronousSocketChannel client ;
public AIOClient() throws Exception{
client = AsynchronousSocketChannel.open();
}
public void connect(String host,int port)throws Exception{
//连接服务端
client.connect(new InetSocketAddress(host,port),null,new CompletionHandler<Void,Void>() {
/**
* 成功操作
* @param result
* @param attachment
*/
@Override
public void completed(Void result, Void attachment) {
try {
client.write(ByteBuffer.wrap(("客户端线程：" + Thread.currentThread().getName()+"请求服务端").getBytes())).get();
System.out.println("已发送至服务器");
} catch (Exception ex) {
ex.printStackTrace();
}
}
/**
* 失败操作
* @param exc
* @param attachment
*/
@Override
public void failed(Throwable exc, Void attachment) {
exc.printStackTrace();
}
});
//读取数据
final ByteBuffer bb = ByteBuffer.allocate(1024);
client.read(bb, null, new CompletionHandler<Integer,Object>(){
/**
* 成功操作
* @param result
* @param attachment
*/
@Override
public void completed(Integer result, Object attachment) {
System.out.println("获取反馈结果：" + new String(bb.array()));
}
/**
* 失败操作
* @param exc
* @param attachment
*/
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
}
}
);
try {
Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException ex) {
System.out.println(ex);
}
}
public static void main(String args[])throws Exception{
int count = 10;
final CountDownLatch latch = new CountDownLatch(count);
for (int i = 0; i < count; i ++) {
latch.countDown();
new Thread(){
@Override
public void run(){
try{
latch.await();
new AIOClient().connect("localhost",9999);
}catch(Exception e){
}
}
}.start();
}
Thread.sleep(1000 * 60 * 10);
}
}
服务端：
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.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import static java.util.concurrent.Executors.*;
/**
* @ProjectName: onereader
* @Package: com.onereader.webblog.common.aio
* @ClassName: AIOServer
* @Author: onereader
* @Description: ${description}
* @Date: 2019/9/1 14:04
* @Version: 1.0
*/
public class AIOServer {
private final int port;
public static void main(String args[]) {
int port = 9999;
new AIOServer(port);
}
/**
* 注册一个端口，用来给客户端连接
* @param port
*/
public AIOServer(int port) {
this.port = port;
listen();
}
//侦听方法
private void listen() {
try {
//线程缓冲池，为了体现异步
ExecutorService executorService = newCachedThreadPool();
//给线程池初始化一个线程
AsynchronousChannelGroup threadGroup = AsynchronousChannelGroup.withCachedThreadPool(executorService, 1);
//Asynchronous异步
final AsynchronousServerSocketChannel server = AsynchronousServerSocketChannel.open(threadGroup);
//启动监听
server.bind(new InetSocketAddress(port));
System.out.println("服务已启动，监听端口" + port);
final Map<String,Integer> count = new ConcurrentHashMap<String, Integer>();
count.put("count", 0);
//开始等待客户端连接
//实现一个CompletionHandler 的接口，匿名的实现类
server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
//缓存区
final ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
//实现IO操作完成的方法
@Override
public void completed(AsynchronousSocketChannel result, Object attachment) {
count.put("count", count.get("count") + 1);
System.out.println(count.get("count"));
try {
//清空缓存标记
buffer.clear();
//读取缓存内容
result.read(buffer).get();
//写模式转换成读模式
buffer.flip();
result.write(buffer);
buffer.flip();
} catch (Exception e) {
System.out.println(e.toString());
} finally {
try {
result.close();
server.accept(null, this);
} catch (Exception e) {
System.out.println(e.toString());
}
}
}
//实现IO操作失败的方法
@Override
public void failed(Throwable exc, Object attachment) {
System.out.println("IO操作是失败: " + exc);
}
});
try {
Thread.sleep(Integer.MAX_VALUE);
} catch (InterruptedException ex) {
System.out.println(ex);
}
} catch (IOException e) {
System.out.println(e);
}
}
}
结果：
客户端
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
已发送至服务器
获取反馈结果：客户端线程：Thread-22请求服务端
获取反馈结果：客户端线程：Thread-21请求服务端
获取反馈结果：客户端线程：Thread-20请求服务端
获取反馈结果：客户端线程：Thread-19请求服务端
获取反馈结果：客户端线程：Thread-18请求服务端
获取反馈结果：客户端线程：Thread-17请求服务端
获取反馈结果：客户端线程：Thread-16请求服务端
获取反馈结果：客户端线程：Thread-15请求服务端
获取反馈结果：客户端线程：Thread-14请求服务端
获取反馈结果：客户端线程：Thread-13请求服务端
服务端
服务已启动，监听端口9999
1
2
3
4
5
6
7
8
9
10
总结
看看代码实现，是不是比看一堆文字要舒服多了，我这里也不想过多熬诉了，大家自己看代码吧~
以上就是我本次总结的，说实话看了很多大佬写的东西。实在是难以通过文字就明白这个东西是咋回事，最后还是亲手实践操作以下，才能理解。