本次我们主要来说一下我们的IO阻塞模型,只是不多,但是一定要理解,对于后面理解netty很重要的
IO模型精讲
IO模型就是说用什么样的通道进行数据的发送和接收,Java共支持3种网络编程IO模式:BIO,NIO,AIO。
BIO
BIO(Blocking IO) 同步阻塞模型,一个客户端连接对应一个处理线程。也是我们熟悉的同步阻塞模型,先别管那个同步的概念,我们先来看一下什么是阻塞,简单来一段代码。
服务端:
package com.xiaocai.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class SocketServer {
public static void main(String[] args) throws IOException {
ServerSocket serverSocket = new ServerSocket(9000);
while (true) {
System.out.println("等待连接。。");
//阻塞方法
Socket socket = serverSocket.accept();
System.out.println("有客户端连接了。。");
handler(socket);
}
}
private static void handler(Socket socket) throws IOException {
System.out.println("thread id = " + Thread.currentThread().getId());
byte[] bytes = new byte[1024];
System.out.println("准备read。。");
//接收客户端的数据,阻塞方法,没有数据可读时就阻塞
int read = socket.getInputStream().read(bytes);
System.out.println("read完毕。。");
if (read != -1) {
System.out.println("接收到客户端的数据:" + new String(bytes, 0, read));
System.out.println("thread id = " + Thread.currentThread().getId());
}
socket.getOutputStream().write("HelloClient".getBytes());
socket.getOutputStream().flush();
}
}
客户端
package com.xiaocai.bio;
import java.io.IOException;
import java.net.Socket;
public class SocketClient {
public static void main(String[] args) throws IOException {
Socket socket = new Socket("127.0.0.1", 9000);
//向服务端发送数据
socket.getOutputStream().write("HelloServer".getBytes());
socket.getOutputStream().flush();
System.out.println("向服务端发送数据结束");
byte[] bytes = new byte[1024];
//接收服务端回传的数据
socket.getInputStream().read(bytes);
System.out.println("接收到服务端的数据:" + new String(bytes));
socket.close();
}
}
这个就是一个简单的BIO服务端代码,就是要准备接受线程访问的代码段。这一个单线程版本什么意思呢?
我们先开启一个端口为9000的socket服务,然后运行Socket socket = serverSocket.accept();意思就是等待线程的出现,我们来接收客户端的请求,这个方法时阻塞的,也是只有在阻塞状态才可以接收到我们的请求。当有请求进来时,运行handler(socket);方法,中间是打印线程ID的方法不解释,int read = socket.getInputStream().read(bytes);准备读取我们的客户端发送数据。read和write可能会混淆,我画个图来说一下。
我们也可以看到我们的客户端也是先拿到socket连接(Socket socket = new Socket("127.0.0.1", 9000)),然后要往服务端写入数据(socket.getOutputStream().write("HelloServer".getBytes());)以byte字节形式写入。这时我们的服务端等待read我们的客户端weite的数据,会进入阻塞状态,如果我们的客户端迟迟不写数据,我们的客户端一直是阻塞状态,也就无法接收到新的请求,因为阻塞了,没法回到我们的Socket socket = serverSocket.accept();去等待客户端请求,只要在serverSocket.accept阻塞时才可以接收新的请求。于是我们采取了多线程的方式来解决这个问题,我们来看一下代码。
package com.xiaocai.bio;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
public class SocketServer {
public static void main(String[] args) throws IOException {
ServerSocket serverSocket = new ServerSocket(9000);
while (true) {
System.out.println("等待连接。。");
//阻塞方法
Socket socket = serverSocket.accept();
System.out.println("有客户端连接了。。");
new Thread(new Runnable() {
@Override
public void run() {
try {
handler(socket);
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
}
}
private static void handler(Socket socket) throws IOException {
System.out.println("thread id = " + Thread.currentThread().getId());
byte[] bytes = new byte[1024];
System.out.println("准备read。。");
//接收客户端的数据,阻塞方法,没有数据可读时就阻塞
int read = socket.getInputStream().read(bytes);
System.out.println("read完毕。。");
if (read != -1) {
System.out.println("接收到客户端的数据:" + new String(bytes, 0, read));
System.out.println("thread id = " + Thread.currentThread().getId());
}
socket.getOutputStream().write("HelloClient".getBytes());
socket.getOutputStream().flush();
}
}
我们这时每次有客户端来新的请求时,我们就会开启一个线程来处理这个请求,及时你的客户端没有及时的write数据,虽然我们的服务端read进行了阻塞,也只是阻塞了你自己的线程,不会造成其它请求无法接收到。
这样的处理方式貌似好了很多很多,其实不然,想一个实例,我们的看小妹直播时,一句欢迎榜一大哥,弹幕很多,加入一次性来了100弹幕还好,我们开启100个线程来处理,如果一起来了十万弹幕呢?难道你要开启十万个线程来处理这些弹幕嘛?很显然BIO还是有弊端的,BIO还是有优点的(代码少,不容易出错)。
NIO
NIO(Non Blocking IO) 同步非阻塞,服务器实现模式为一个线程可以处理多个请求(连接),客户端发送的连接请求都会注册到多路复用器selector上,多路复用器轮询到连接有IO请求就进行处理。 可能概念太抽象了,我来举个例子吧,现在有两个小区都有很多的房子出租,BIO小区和NIO小区,都有一个门卫,BIO小区,来了一个租客,门卫大爷就拿着钥匙,带这个租客去看房子了,后面来的租客都暂时无法看房子了,尴尬...想同时多人看房子,必须增加门卫大爷的数量,而我们的NIO小区就很聪明,还是一个门卫大妈,来了一个租客要看房子,门卫大妈,给了那个租客一把钥匙,并且告诉他哪房间是空的,你自己进去看吧,及时这个租客看房子慢,耽误了很多时间也不怕了,因为门卫大妈一直在门卫室,即使又来了新的租客,门卫大妈也是如此,只给钥匙和空房间地址就可以了。这个例子反正我记得很清楚,也觉得很贴切,这里提到了一个钥匙的概念,一会告诉你们是做什么的,我们先看一下代码。
服务端
package com.xiaocai.nio;
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;
public class NIOServer {
//public static ExecutorService pool = Executors.newFixedThreadPool(10);
public static void main(String[] args) throws IOException {
// 创建一个在本地端口进行监听的服务Socket通道.并设置为非阻塞方式
ServerSocketChannel ssc = ServerSocketChannel.open();
//必须配置为非阻塞才能往selector上注册,否则会报错,selector模式本身就是非阻塞模式
ssc.configureBlocking(false);
ssc.socket().bind(new InetSocketAddress(9000));
// 创建一个选择器selector
Selector selector = Selector.open();
// 把ServerSocketChannel注册到selector上,并且selector对客户端accept连接操作感兴趣
ssc.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
System.out.println("等待事件发生。。");
// 轮询监听channel里的key,select是阻塞的,accept()也是阻塞的
int select = selector.select();
System.out.println("有事件发生了。。");
// 有客户端请求,被轮询监听到
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey key = it.next();
//删除本次已处理的key,防止下次select重复处理
it.remove();
handle(key);
}
}
}
private static void handle(SelectionKey key) throws IOException {
if (key.isAcceptable()) {
System.out.println("有客户端连接事件发生了。。");
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
//NIO非阻塞体现:此处accept方法是阻塞的,但是这里因为是发生了连接事件,所以这个方法会马上执行完,不会阻塞
//处理完连接请求不会继续等待客户端的数据发送
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
//通过Selector监听Channel时对读事件感兴趣
sc.register(key.selector(), SelectionKey.OP_READ);
} else if (key.isReadable()) {
System.out.println("有客户端数据可读事件发生了。。");
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
//NIO非阻塞体现:首先read方法不会阻塞,其次这种事件响应模型,当调用到read方法时肯定是发生了客户端发送数据的事件
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);
key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
} else if (key.isWritable()) {
SocketChannel sc = (SocketChannel) key.channel();
System.out.println("write事件");
// NIO事件触发是水平触发
// 使用Java的NIO编程的时候,在没有数据可以往外写的时候要取消写事件,
// 在有数据往外写的时候再注册写事件
key.interestOps(SelectionKey.OP_READ);
//sc.close();
}
}
}
客户端
package com.xiaocai.nio;
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.util.Iterator;
public class NioClient {
//通道管理器
private Selector selector;
/**
* 启动客户端测试
*
* @throws IOException
*/
public static void main(String[] args) throws IOException {
NioClient client = new NioClient();
client.initClient("127.0.0.1", 9000);
client.connect();
}
/**
* 获得一个Socket通道,并对该通道做一些初始化的工作
*
* @param ip 连接的服务器的ip
* @param port 连接的服务器的端口号
* @throws IOException
*/
public void initClient(String ip, int port) throws IOException {
// 获得一个Socket通道
SocketChannel channel = SocketChannel.open();
// 设置通道为非阻塞
channel.configureBlocking(false);
// 获得一个通道管理器
this.selector = Selector.open();
// 客户端连接服务器,其实方法执行并没有实现连接,需要在listen()方法中调
//用channel.finishConnect() 才能完成连接
channel.connect(new InetSocketAddress(ip, port));
//将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_CONNECT事件。
channel.register(selector, SelectionKey.OP_CONNECT);
}
/**
* 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
*
* @throws IOException
*/
public void connect() throws IOException {
// 轮询访问selector
while (true) {
selector.select();
// 获得selector中选中的项的迭代器
Iterator<SelectionKey> it = this.selector.selectedKeys().iterator();
while (it.hasNext()) {
SelectionKey key = (SelectionKey) it.next();
// 删除已选的key,以防重复处理
it.remove();
// 连接事件发生
if (key.isConnectable()) {
SocketChannel channel = (SocketChannel) key.channel();
// 如果正在连接,则完成连接
if (channel.isConnectionPending()) {
channel.finishConnect();
}
// 设置成非阻塞
channel.configureBlocking(false);
//在这里可以给服务端发送信息哦
ByteBuffer buffer = ByteBuffer.wrap("HelloServer".getBytes());
channel.write(buffer);
//在和服务端连接成功之后,为了可以接收到服务端的信息,需要给通道设置读的权限。
channel.register(this.selector, SelectionKey.OP_READ); // 获得了可读的事件
} else if (key.isReadable()) {
read(key);
}
}
}
}
/**
* 处理读取服务端发来的信息 的事件
*
* @param key
* @throws IOException
*/
public void read(SelectionKey key) throws IOException {
//和服务端的read方法一样
// 服务器可读取消息:得到事件发生的Socket通道
SocketChannel channel = (SocketChannel) key.channel();
// 创建读取的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
int len = channel.read(buffer);
if (len != -1) {
System.out.println("客户端收到信息:" + new String(buffer.array(), 0, len));
}
}
}
代码看到了很多很多,我来解释一下大概什么意思吧,这个NIO超级重要后面的netty就是基于这个写的,一定要搞懂,首先我们创建了一个ServerSocketChannel和一个选择器selector,设置为非阻塞的(固定写法,没有为什么),将我们的 selector绑定到我们的ServerSocketChannel上,然后运行selector.select();进入阻塞状态,别担心,这个阻塞没影响,为我们提供了接收客户端的请求,你没有请求,我阻塞着,不会耽误你们什么的。
回到我们的客户端,还是差不多的样子,拿到我们的NioClient开始连接我们的服务端,这个时候,我们的服务端接收到了我们的客户端请求,阻塞状态的selector.select()继续运行,并且给予了一个SelectionKey(Iterator<SelectionKey> it = selector.selectedKeys().iterator())也就是我们刚才的小例子中提到的钥匙,key=钥匙,还算是靠谱吧~!开始运行我们的handle方法,有个if else,这个是说,你是第一次请求要建立通道,还是要写数据,还是要读取数据,记住啊,读写都是相对的,自己多琢磨几次就可以转过圈来了,就是我上面画图说的read和write。拿我们的建立通道来说,通过我们的钥匙key你就可以得到ServerSocketChannel,然后进行设置下次可能会发生的读写事件,然后看我们的读事件,我们看到了int len = sc.read(buffer)这个读在我们的BIO中是阻塞的,而我们的NIO这个方法不是阻塞的,这也就体现出来了我们的BIO同步阻塞和NIO同步非阻塞,阻塞和非阻塞的区别也就说完了。画个图,我们来看一下我们的NIO模型。
NIO 有三大核心组件: Channel(通道), Buffer(缓冲区),Selector(选择器)
这里我们的Buffer没有去说,到netty会说的, Channel(通道), Buffer(缓冲区)都是双向的,现在回过头来想想我举的小例子,selector门卫大妈,SelectionKey钥匙。对于NIO有了一些理解了吧,NIO看着很棒的,但是你有想过写上述代码的痛苦吗?
AIO
AIO(NIO 2.0) 异步非阻塞, 由操作系统完成后回调通知服务端程序启动线程去处理, 一般适用于连接数较多且连接时间较长的应用。其实AIO就是对于NIO的二次封装,要不怎么叫做NIO2.0呢,我们来简单看一下代码。
服务端:
package com.xiaocai.aio;
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 AIOServer {
public static void main(String[] args) throws Exception {
final AsynchronousServerSocketChannel serverChannel =
AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(9000));
serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
@Override
public void completed(AsynchronousSocketChannel socketChannel, Object attachment) {
try {
// 再此接收客户端连接,如果不写这行代码后面的客户端连接连不上服务端
serverChannel.accept(attachment, this);
System.out.println(socketChannel.getRemoteAddress());
ByteBuffer buffer = ByteBuffer.allocate(1024);
socketChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
buffer.flip();
System.out.println(new String(buffer.array(), 0, result));
socketChannel.write(ByteBuffer.wrap("HelloClient".getBytes()));
}
@Override
public void failed(Throwable exc, ByteBuffer buffer) {
exc.printStackTrace();
}
});
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, Object attachment) {
exc.printStackTrace();
}
});
Thread.sleep(Integer.MAX_VALUE);
}
}
客户端:
package com.xiaocai.aio;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
public class AIOClient {
public static void main(String... args) throws Exception {
AsynchronousSocketChannel socketChannel = AsynchronousSocketChannel.open();
socketChannel.connect(new InetSocketAddress("127.0.0.1", 9000)).get();
socketChannel.write(ByteBuffer.wrap("HelloServer".getBytes()));
ByteBuffer buffer = ByteBuffer.allocate(512);
Integer len = socketChannel.read(buffer).get();
if (len != -1) {
System.out.println("客户端收到信息:" + new String(buffer.array(), 0, len));
}
}
}
阻塞非阻塞都明白了,这里来解释一下同步,我们看到我们的AIO在serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {}直接开启了线程,也就是说accept直接以后,我不再需要考虑阻塞情况,可以继续运行下面的代码了,也就是我们说到的异步执行,内部还是我们的NIO,不要觉得AIO多么的6B,内部就是封装了我们的NIO,性能和NIO其实差不多的,可能有些时候还不如NIO(未实测)。
遗漏一个知识点,NIO的多路复用器是如何工作的,在我们的JDK1.5以前的,多路复用器是数组和链表的方式来遍历的,到了我们的JDK1.5采用hash来回调的。
总结:
我们这次主要说了BIO、NIO、AIO三个网络编程IO模式,最重要的就是我们的NIO,一张图来总结一下三个IO的差别吧。