java NIO 前言
我从大二开始学习Java,一直偏重于J2EE领域,写多了SSH、SSM代码之后,Java让我失去了新鲜感,以为调调接口就完事了。笔者一度开始拥抱Go语言,直到我知道“JAVA NIO”这回事,才发现,JAVA能做的有很多。比如在多线程(java.util.concurrent)及网络领域(java.nio),老树开新花。
io即输入输出,输入输出的源头与目的地主要是网络和文件,我们先从比较简单的文件IO说起。
文件IO
以读取文件为例:
// FileInputStream或BufferedInputStream
Byte[] b = new byte[1024]; // 开启1m的缓冲区
while(in.read(b) != -1){
Xxx
}
# nio方式
FileInputStream fin = new FileInputStream( "readandshow.txt" );
FileChannel fc = fin.getChannel();
ByteBuffer buffer = ByteBuffer.allocate(1024);
fc.read(buffer);
传统的io代码,这个缓冲区(byte数组)是程序员约定俗成的行为。在jdk较新版本中,传统io类实际是用nio类实现的。
在java nio中,这个缓冲区被固定下来,数据直接被读取到缓冲区中。或者说,原先的Stream是Channel和Buffer的组合。
实际上,内核在读取java文件时,会将文件的部分内容拷贝到缓冲区块中。进行网络通信时,网络数据会最新到达tcp接收缓冲区中。Channel用于在字节缓冲区和位于通道另一侧实体(文件或套接字)的字节缓冲区之间有效地传输数据。使用buffer则使read()和write()调用得到了极大的简化,因为许多工作细节都由缓冲区完成了。clear()和flip()方法用于让缓冲区在读和写之间切换。
网络IO
在阻塞IO模式下,从网络中读取数据的过程是
InputeStream in = socket.getInputStream();
in.read();
这相当于用户线程主动去查询是否收到数据,读写事件的发起者是用户线程。java的nio框架主要基于Reactor模式(nio中的学名叫selector),有一个专门的reactor线程去监听注册的套接字,当发生连接建立和读写事件时,通知用户线程处理。此时,事件的发起者是Reactor线程。
而Reactor模式又与观察者模式类似,我们先从观察者模式开始。
从观察者模式讲起
假设有一个目标对象(Objector)和一个观察者对象(Observer),Observer想要知道Objector状态,有两种办法:
- 轮询(可以理解为pull的方式)
-
Objector状态发生变化时,主动通知Observer(可以理解为push的方式,事件源主动告知Observer)
class Objector{ List<Observer> observers; change(){ // 变化 // 通知观察者 for(Observer observer : observers){ observer.xxxx } } }
我们将这个模型简单的拓展一下,假设Observer要观察两个Objector的变化(Objector1和Objector2),那么我们可以:
class Objector1{
List<Observer> observers;
change(){
// 变化1
// 通知观察者
for(Observer observer : observers){
observer.xxxx
}
}
change2(){
// 变化2
// 通知观察者
for(Observer observer : observers){
observer.xxxx
}
}
}
class Objector2{
List<Observer> observers;
change(){
变化
for(Observer observer : observers){
observer.xxxx
}
}
}
但这样没有什么扩展性,比如无法支持“change2的只有个别Observer才关心”这样的需求。
计算机里的所有问题,可以试试加个中间层解决。于是,我们将Objector1中保存观察者的容器,以及通知观察者的代码提取出来,Objector有什么事就告诉中间件,Observer关心什么事也告诉中间件,然后由中间件负责准确的通知。
class Objector1{
中间件 middleware
change(){
middleware.func(变化信息)
}
}
class 中间件{
Map<感兴趣的事件,观察者列表>
func(){
While(true){
查询object1和object2有没有变化
有,就根据映射拿到观察者,调用观察者处理
}
}
}
这便是Reactor模式的一个简要雏形。观察者模式与单个事件源关联,而反应器模式则与多个事件源关联。
换个方式写WebServer
以一个web服务器为例,最简单的例子是这样的
class SingleThreadWebServer{
public static void main(String[] args) throws IOException{
ServerSocket socket = new ServerSocket();
while(true){
Socket connection = socket.accept(); //看到这个变量名,我好像明白,为什么叫"连接"池了
handleRequest(connection);
}
}
}
上述代码有两个阻塞的点, socket.accept()
和connection.read()
,很明显一个线程忙不过来,so
class ThreadPerTaskWebServer{
public static void main(String[] args) throws IOException{
ServerSocket socket = new ServerSocket();
while(true){
final Socket connection = socket.accept();
Runnable task = new Runnable(){
public void run(){
handleRequest(connection);
}
}
new Thread(task).start();
}
}
}
如果请求过多,这种方式会无限制创建线程。当然,我们可以使用Executor来执行task。但共同点都是,一个work线程处理一个connection。如果请求不是很繁忙,主线程会阻塞在socket.accept()
上,work线程会阻塞在connection.read()
上。
使用nio方式
public class NIOServer{
public static void main(String[] args) throws IOException{
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.socket().bind(new InetSocketAddress(80));
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while(true){
selector.select(1000);
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> it = selectedKeys.iterator();
SelectionKey key = null;
while (it.hasNext()) {
key = it.next();
it.remove();
handleInput(key);
}
}
}
public static void handleInput(SelectionKey key) throws IOException{
if(key.isAcceptable()) {
// Accept the new connection
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);
// Add the new connection to the selector
sc.register(selector, SelectionKey.OP_READ);
} else if (key.isReadable()) {
SocketChannel sc = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
// handle buffer