io感觉上和多线程并没有多大关系，但是nio改变了线程在应用层面使用的方式，也解决了一些实际的困难。而aio是异步io和前面的系列也有点关系。在此，为了学习和记录，也写一篇文章来介绍nio和aio。
1. 什么是nio
nio是new i/o的简称，与旧式的基于流的i/o方法相对，从名字看，它表示新的一套java i/o标 准。它是在java 1.4中被纳入到jdk中的，并具有以下特性：
所有的从通道中的读写操作，都要经过buffer，而通道就是io的抽象，通道的另一端就是操纵的文件。
2. buffer
java中buffer的实现。基本的数据类型都有它对应的buffer
buffer的简单使用例子：
我要评论
package test;
import java.io.file;
import java.io.fileinputstream;
import java.nio.bytebuffer;
import java.nio.channels.filechannel;
public class test {
public static void main(string[] args) throws exception {
fileinputstream fin = new fileinputstream(new file(
"d:\\temp_buffer.tmp"));
filechannel fc = fin.getchannel();
bytebuffer bytebuffer = bytebuffer.allocate(1024);
fc.read(bytebuffer);
fc.close();
bytebuffer.flip();//读写转换
}
}
总结下使用的步骤是：
1. 得到channel
2. 申请buffer
3. 建立channel和buffer的读/写关系
4. 关闭
下面的例子是使用nio来复制文件：
public static void niocopyfile(string resource, string destination)
throws ioexception {
fileinputstream fis = new fileinputstream(resource);
fileoutputstream fos = new fileoutputstream(destination);
filechannel readchannel = fis.getchannel(); // 读文件通道
filechannel writechannel = fos.getchannel(); // 写文件通道
bytebuffer buffer = bytebuffer.allocate(1024); // 读入数据缓存
while (true) {
buffer.clear();
int len = readchannel.read(buffer); // 读入数据
if (len == -1) {
break; // 读取完毕
}
buffer.flip();
writechannel.write(buffer); // 写入文件
}
readchannel.close();
writechannel.close();
}
buffer中有3个重要的参数：位置（position）、容量（capactiy）和上限（limit）
这里要区别下容量和上限，比如一个buffer有10kb，那么10kb就是容量，我将5kb的文件读到buffer中，那么上限就是5kb。
下面举个例子来理解下这3个重要的参数：
public static void main(string[] args) throws exception {
bytebuffer b = bytebuffer.allocate(15); // 15个字节大小的缓冲区
system.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
+ " position=" + b.position());
for (int i = 0; i < 10; i++) {
// 存入10个字节数据
b.put((byte) i);
}
system.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
+ " position=" + b.position());
b.flip(); // 重置position
system.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
+ " position=" + b.position());
for (int i = 0; i < 5; i++) {
system.out.print(b.get());
}
system.out.println();
system.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
+ " position=" + b.position());
b.flip();
system.out.println("limit=" + b.limit() + " capacity=" + b.capacity()
+ " position=" + b.position());
}
整个过程如图：
此时position从0到10，capactiy和limit不变。
该操作会重置position，通常，将buffer从写模式转换为读 模式时需要执行此方法 flip()操作不仅重置了当前的position为0，还将limit设置到当前position的位置 。
limit的意义在于，来确定哪些数据是有意义的，换句话说，从position到limit之间的数据才是有意义的数据，因为是上次操作的数据。所以flip操作往往是读写转换的意思。
意义同上。
而buffer中大多数的方法都是去改变这3个参数来达到某些功能的：
public final buffer rewind()
将position置零，并清除标志位（mark）
public final buffer clear()
将position置零，同时将limit设置为capacity的大小，并清除了标志mark
public final buffer flip()
先将limit设置到position所在位置，然后将position置零，并清除标志位mark，通常在读写转换时使用
文件映射到内存
public static void main(string[] args) throws exception {
randomaccessfile raf = new randomaccessfile("c:\\mapfile.txt", "rw");
filechannel fc = raf.getchannel();
// 将文件映射到内存中
mappedbytebuffer mbb = fc.map(filechannel.mapmode.read_write, 0,
raf.length());
while (mbb.hasremaining()) {
system.out.print((char) mbb.get());
}
mbb.put(0, (byte) 98); // 修改文件
raf.close();
}
对mappedbytebuffer的修改就相当于修改文件本身，这样操作的速度是很快的。
3. channel
多线程网络服务器的一般结构：
简单的多线程服务器：
public static void main(string[] args) throws exception {
serversocket echoserver = null;
socket clientsocket = null;
try {
echoserver = new serversocket(8000);
} catch (ioexception e) {
system.out.println(e);
}
while (true) {
try {
clientsocket = echoserver.accept();
system.out.println(clientsocket.getremotesocketaddress()
+ " connect!");
tp.execute(new handlemsg(clientsocket));
} catch (ioexception e) {
system.out.println(e);
}
}
}
功能就是服务器端读到什么数据，就向客户端回写什么数据。
这里的tp是一个线程池，handlemsg是处理消息的类。
static class handlemsg implements runnable{
省略部分信息
public void run(){
try {
is = new bufferedreader(new inputstreamreader(clientsocket.getinputstream()));
os = new printwriter(clientsocket.getoutputstream(), true);
// 从inputstream当中读取客户端所发送的数据
string inputline = null;
long b=system. currenttimemillis ();
while ((inputline = is.readline()) != null)
{
os.println(inputline);
}
long e=system. currenttimemillis ();
system. out.println ("spend:"+(e - b)+" ms ");
} catch (ioexception e) {
e.printstacktrace();
}finally
{
关闭资源
}
}
}
客户端：
public static void main(string[] args) throws exception {
socket client = null;
printwriter writer = null;
bufferedreader reader = null;
try {
client = new socket();
client.connect(new inetsocketaddress("localhost", 8000));
writer = new printwriter(client.getoutputstream(), true);
writer.println("hello!");
writer.flush();
reader = new bufferedreader(new inputstreamreader(
client.getinputstream()));
system.out.println("from server: " + reader.readline());
} catch (exception e) {
} finally {
// 省略资源关闭
}
}
以上的网络编程是很基本的，使用这种方式，会有一些问题：
为每一个客户端使用一个线程，如果客户端出现延时等异常，线程可能会被占用很长时间。因为数据的准备和读取都在这个线程中。此时，如果客户端数量众多，可能会消耗大量的系统资源。
解决方案:
使用非阻塞的nio （读取数据不等待，数据准备好了再工作）
为了体现nio使用的高效。
这里先模拟一个低效的客户端来模拟因网络而延时的情况：
private static executorservice tp= executors.newcachedthreadpool();
private static final int sleep_time=1000*1000*1000;
public static class echoclient implements runnable{
public void run(){
try {
client = new socket();
client.connect(new inetsocketaddress("localhost", 8000));
writer = new printwriter(client.getoutputstream(), true);
writer.print("h");
locksupport.parknanos(sleep_time);
writer.print("e");
locksupport.parknanos(sleep_time);
writer.print("l");
locksupport.parknanos(sleep_time);
writer.print("l");
locksupport.parknanos(sleep_time);
writer.print("o");
locksupport.parknanos(sleep_time);
writer.print("!");
locksupport.parknanos(sleep_time);
writer.println();
writer.flush();
}catch(exception e)
{
}
}
}
服务器端输出：
spend:6000ms
spend:6000ms
spend:6000ms
spend:6001ms
spend:6002ms
spend:6002ms
spend:6002ms
spend:6002ms
spend:6003ms
spend:6003ms
因为
while ((inputline = is.readline()) != null)
是阻塞的，所以时间都花在等待中。
如果用nio来处理这个问题会怎么做呢？
nio有一个很大的特点就是：把数据准备好了再通知我
而channel有点类似于流，一个channel可以和文件或者网络socket对应 。
selector是一个选择器，它可以选择某一个channel，然后做些事情。
一个线程可以对应一个selector，而一个selector可以轮询多个channel，而每个channel对应了一个socket。
与上面一个线程对应一个socket相比，使用nio后，一个线程可以轮询多个socket。
当selector调用select()时，会查看是否有客户端准备好了数据。当没有数据被准备好时，select()会阻塞。平时都说nio是非阻塞的，但是如果没有数据被准备好还是会有阻塞现象。
当有数据被准备好时，调用完select()后，会返回一个selectionkey，selectionkey表示在某个selector上的某个channel的数据已经被准备好了。
只有在数据准备好时，这个channel才会被选择。
这样nio实现了一个线程来监控多个客户端。
而刚刚模拟的网络延迟的客户端将不会影响nio下的线程，因为某个socket网络延迟时，数据还未被准备好，selector是不会选择它的，而会选择其他准备好的客户端。
selectnow()与select()的区别在于，selectnow()是不阻塞的，当没有客户端准备好数据时，selectnow()不会阻塞，将返回0，有客户端准备好数据时，selectnow()返回准备好的客户端的个数。
主要代码：
package test;
import java.net.inetaddress;
import java.net.inetsocketaddress;
import java.net.socket;
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.nio.channels.spi.abstractselector;
import java.nio.channels.spi.selectorprovider;
import java.util.hashmap;
import java.util.iterator;
import java.util.linkedlist;
import java.util.map;
import java.util.set;
import java.util.concurrent.executorservice;
import java.util.concurrent.executors;
public class multithreadnioechoserver {
public static map<socket, long> geym_time_stat = new hashmap<socket, long>();
class echoclient {
private linkedlist<bytebuffer> outq;
echoclient() {
outq = new linkedlist<bytebuffer>();
}
public linkedlist<bytebuffer> getoutputqueue() {
return outq;
}
public void enqueue(bytebuffer bb) {
outq.addfirst(bb);
}
}
class handlemsg implements runnable {
selectionkey sk;
bytebuffer bb;
public handlemsg(selectionkey sk, bytebuffer bb) {
super();
this.sk = sk;
this.bb = bb;
}
@override
public void run() {
// todo auto-generated method stub
echoclient echoclient = (echoclient) sk.attachment();
echoclient.enqueue(bb);
sk.interestops(selectionkey.op_read | selectionkey.op_write);
selector.wakeup();
}
}
private selector selector;
private executorservice tp = executors.newcachedthreadpool();
private void startserver() throws exception {
selector = selectorprovider.provider().openselector();
serversocketchannel ssc = serversocketchannel.open();
ssc.configureblocking(false);
inetsocketaddress isa = new inetsocketaddress(8000);
ssc.socket().bind(isa);
// 注册感兴趣的事件，此处对accpet事件感兴趣
selectionkey acceptkey = ssc.register(selector, selectionkey.op_accept);
for (;;) {
selector.select();
set readykeys = selector.selectedkeys();
iterator i = readykeys.iterator();
long e = 0;
while (i.hasnext()) {
selectionkey sk = (selectionkey) i.next();
i.remove();
if (sk.isacceptable()) {
doaccept(sk);
} else if (sk.isvalid() && sk.isreadable()) {
if (!geym_time_stat.containskey(((socketchannel) sk
.channel()).socket())) {
geym_time_stat.put(
((socketchannel) sk.channel()).socket(),
system.currenttimemillis());
}
doread(sk);
} else if (sk.isvalid() && sk.iswritable()) {
dowrite(sk);
e = system.currenttimemillis();
long b = geym_time_stat.remove(((socketchannel) sk
.channel()).socket());
system.out.println("spend:" + (e - b) + "ms");
}
}
}
}
private void dowrite(selectionkey sk) {
// todo auto-generated method stub
socketchannel channel = (socketchannel) sk.channel();
echoclient echoclient = (echoclient) sk.attachment();
linkedlist<bytebuffer> outq = echoclient.getoutputqueue();
bytebuffer bb = outq.getlast();
try {
int len = channel.write(bb);
if (len == -1) {
disconnect(sk);
return;
}
if (bb.remaining() == 0) {
outq.removelast();
}
} catch (exception e) {
// todo: handle exception
disconnect(sk);
}
if (outq.size() == 0) {
sk.interestops(selectionkey.op_read);
}
}
private void doread(selectionkey sk) {
// todo auto-generated method stub
socketchannel channel = (socketchannel) sk.channel();
bytebuffer bb = bytebuffer.allocate(8192);
int len;
try {
len = channel.read(bb);
if (len < 0) {
disconnect(sk);
return;
}
} catch (exception e) {
// todo: handle exception
disconnect(sk);
return;
}
bb.flip();
tp.execute(new handlemsg(sk, bb));
}
private void disconnect(selectionkey sk) {
// todo auto-generated method stub
//省略略干关闭操作
}
private void doaccept(selectionkey sk) {
// todo auto-generated method stub
serversocketchannel server = (serversocketchannel) sk.channel();
socketchannel clientchannel;
try {
clientchannel = server.accept();
clientchannel.configureblocking(false);
selectionkey clientkey = clientchannel.register(selector,
selectionkey.op_read);
echoclient echoclinet = new echoclient();
clientkey.attach(echoclinet);
inetaddress clientaddress = clientchannel.socket().getinetaddress();
system.out.println("accepted connection from "
+ clientaddress.gethostaddress());
} catch (exception e) {
// todo: handle exception
}
}
public static void main(string[] args) {
// todo auto-generated method stub
multithreadnioechoserver echoserver = new multithreadnioechoserver();
try {
echoserver.startserver();
} catch (exception e) {
// todo: handle exception
}
}
}
代码仅作参考，主要的特点是，对不同事件的感兴趣来做不同的事。
当用之前模拟的那个延迟的客户端时，这次的时间消耗就在2ms到11ms之间了。性能提升是很明显的。
总结：
1. nio会将数据准备好后，再交由应用进行处理，数据的读取/写入过程依然在应用线程中完成，只是将等待的时间剥离到单独的线程中去。
2. 节省数据准备时间（因为selector可以复用）
5. aio
aio的特点：
1. 读完了再通知我
2. 不会加快io，只是在读完后进行通知
3. 使用回调函数，进行业务处理
aio的相关代码：
asynchronousserversocketchannel
server = asynchronousserversocketchannel.open().bind( new inetsocketaddress (port));
使用server上的accept方法
public abstract <a> void accept(a attachment,completionhandler<asynchronoussocketchannel,? super a> handler);
completionhandler为回调接口，当有客户端accept之后，就做handler中的事情。
示例代码：
server.accept(null,
new completionhandler<asynchronoussocketchannel, object>() {
final bytebuffer buffer = bytebuffer.allocate(1024);
public void completed(asynchronoussocketchannel result,
object attachment) {
system.out.println(thread.currentthread().getname());
future<integer> writeresult = null;
try {
buffer.clear();
result.read(buffer).get(100, timeunit.seconds);
buffer.flip();
writeresult = result.write(buffer);
} catch (interruptedexception | executionexception e) {
e.printstacktrace();
} catch (timeoutexception e) {
e.printstacktrace();
} finally {
try {
server.accept(null, this);
writeresult.get();
result.close();
} catch (exception e) {
system.out.println(e.tostring());
}
}
}
@override
public void failed(throwable exc, object attachment) {
system.out.println("failed: " + exc);
}
});
这里使用了future来实现即时返回，关于future请参考上一篇
在理解了nio的基础上，看aio，区别在于aio是等读写过程完成后再去调用回调函数。
nio是同步非阻塞的
aio是异步非阻塞的
由于nio的读写过程依然在应用线程里完成，所以对于那些读写过程时间长的，nio就不太适合。
而aio的读写过程完成后才被通知，所以aio能够胜任那些重量级，读写过程长的任务。
如对本文有疑问， 点击进行留言回复！！
sap cloud platform destination的配置
springcloud中feign调用处理mybatis-plus Ipage反序列化问题。
Flume 史上最全面的大数据学习第十篇（一） 别再说不知道flume是什么了
网友评论