最近学习时碰到事件驱动和非阻塞的相关知识，
随之想到了Java中的Reactor、io与nio的一些东西；
在前辈的博客上翻了翻、复习复习，在此记录一番。
实在找不到比较大点的东西，于是随便弄了个压缩包，大小在1G左右；
写个程序模拟一下下载，开两个客户端线程请求下载；
结果会是：一个请求会一直阻塞，直到一个文件下载完成后另一个文件才开始下载。
先看看服务端的代码:
class DownLoadServer implements Runnable {
@Override
public void run() {
try {
@SuppressWarnings("resource")
final ServerSocket ss = new ServerSocket(8989);
while (true) {
Socket server = ss.accept();
byte[] bfile = new byte[1024];
try {
FileInputStream fis = new FileInputStream("D:/doc_backup.rar");
OutputStream os = server.getOutputStream();
while (fis.read(bfile) > -1) {
os.write(bfile);
}
fis.close();
server.close();
} catch (IOException e) {
System.out.println("server线程输出流我的天");
}
}
} catch (Exception e) {
System.out.println("server线程 我的天~");
}
}
}
很简单，就是accept后开个inputStream和outputStream，边读边写。
接着再看看客户端的代码:
class DownlLoadClient implements Runnable {
@SuppressWarnings("resource")
@Override
public void run() {
try {
Socket client = new Socket("127.0.0.1", 8989);
InputStream is = client.getInputStream();
FileOutputStream fos = new FileOutputStream(
"E:/testfolder/langchao" + Thread.currentThread().getId()
+ ".txt");
byte[] fromServer = new byte[1024];
while (is.read(fromServer) > -1) {
fos.write(fromServer);
}
client.close();
} catch (IOException e) {
System.out.println("client线程我的天~");
}
}
}
输出的文件名是随便取的，也没什么特别，只是把读过来的输出去。
结果当然是这个样子的:
服务端只有一对inputStream和outputStream对象在受理请求，前面的没写完后面的别想写。
那如果有很多inputStream和outputStream对象受理请求呢？
想法不错，也就是说把服务端代码改成这样子:
class DownLoadServer implements Runnable {
@Override
public void run() {
try {
@SuppressWarnings("resource")
final ServerSocket ss = new ServerSocket(8989);
while (true) {
final Socket server = ss.accept();
Thread t = new Thread() {
@Override
public void run() {
super.run();
byte[] bfile = new byte[1024];
try {
FileInputStream fis = new FileInputStream("D:/doc_backup.rar");
OutputStream os = server.getOutputStream();
while (fis.read(bfile) > -1) {
os.write(bfile);
}
fis.close();
server.close();
} catch (IOException e) {
System.out.println("server线程输出流我的天");
}
}
};
t.start();
}
} catch (Exception e) {
System.out.println("server线程 我的天~");
}
}
}
大概就是这个意思，每accept到就为客户端提供"一对一特殊服务"；
嗯，或者也可以算一下获取了多少下载请求，每N次请求开1次"特殊服务"。
但无论如何都无法回避一个问题——"特殊服务"的成本很高，线程的切换和线程的资源都是开销。
如果继续按照这个方法做下去，也只能是弄个Thread Pool。
但如果请求数量超过了pool的maxActive数量，那问题又饶了一圈回来了。
我们追求低成本高效率，于是早在JDK1.4就有了java.nio；
nio怎么讲？有说是new io的、也有叫native io，或许叫non-block io...
概念上也就是channel、buffer、selector、selectionKey...
先看一下server代码:
System.out.println("server start...");
ServerSocketChannel serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8989));
serverChannel.configureBlocking(false);
Selector sel = Selector.open();
serverChannel.register(sel, SelectionKey.OP_ACCEPT);
File file = new File("D:/doc_backup.rar");
ByteBuffer buffer = ByteBuffer.allocate(100*1024);
CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
while(true){
sel.select();
Iterator selKeyItr = sel.selectedKeys().iterator();
while(selKeyItr.hasNext()){
SelectionKey key = selKeyItr.next();
selKeyItr.remove();
String outputFilePath=StringUtils.EMPTY;
if(key.isAcceptable()){
System.out.println("server acceptable");
SocketChannel channel = ((ServerSocketChannel)key.channel()).accept();
channel.configureBlocking(false);
channel.register(sel, SelectionKey.OP_READ);
}else if(key.isReadable()){
System.out.println("server readable");
SocketChannel channel = (SocketChannel) key.channel();
channel.configureBlocking(false);
channel.read(buffer);
buffer.flip();
CharBuffer clientBuffer = decoder.decode(buffer);
outputFilePath = clientBuffer.toString();
buffer.clear();
SelectionKey writeKey = channel.register(sel, SelectionKey.OP_WRITE);
}else if(key.isWritable()){
System.out.println("server writable");
SocketChannel channel =(SocketChannel) key.channel();
FileChannel fileChannel = new FileInputStream(file).getChannel();
ByteBuffer fileByte = ByteBuffer.allocate(1024*100);
while(fileChannel.read(fileByte)!=-1){
fileByte.flip();
channel.write(fileByte);
fileByte.clear();
}
channel.register(sel, SelectionKey.OP_READ);
}
}
}
代码贴出来有点乱，但也就开一个线程，监听与注册事件。
select()方法必须，不然client的send根本recv不到。
socketChannel将blocking设置为false，不然会在事件注册时出现java.nio.channels.IllegalBlockingModeException
Unchecked exception thrown when a blocking-mode-specific operation is invoked upon a channel in the incorrect blocking mode.
同样地，在write事件中把blocking设置为true或者使用阻塞的面向流的IO也会出现同样的异常。
client继承Thread，run method如下:
public void run() {
try {
System.out.println("client...");
SocketAddress addr = new InetSocketAddress(8989);
SocketChannel client = SocketChannel.open();
client.configureBlocking(false);
Selector sel = Selector.open();
client.register(sel, SelectionKey.OP_CONNECT);
CharsetEncoder encoder = Charset.forName("UTF-8").newEncoder();
client.connect(addr);
while (true) {
sel.select();
Iterator selKeyItr = sel.selectedKeys().iterator();
while (selKeyItr.hasNext()) {
SelectionKey key = selKeyItr.next();
selKeyItr.remove();
if (key.isConnectable()) {
System.out.println("client connectble");
SocketChannel channel = (SocketChannel) key.channel();
String filePath = "E:/testfolder/channelTest"+Thread.currentThread().getId()+".rar";
channel.finishConnect();
channel.write(encoder.encode(CharBuffer.wrap(filePath)));
channel.register(sel, SelectionKey.OP_READ).attach(filePath);
} else if (key.isReadable()) {
System.out.println("client readble...");
SocketChannel channel = (SocketChannel) key.channel();
if(key.attachment()!=null){
@SuppressWarnings("resource")
FileChannel fc = new FileOutputStream(key.attachment().toString()).getChannel();
ByteBuffer fileByte = ByteBuffer.allocate(1024*100);
while(channel.read(fileByte)!=-1){
fileByte.flip();
fc.write(fileByte);
fileByte.clear();
}
}
channel.register(sel, SelectionKey.OP_CONNECT);
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
在上面代码中的attach()并没有发挥太大用处，attach()可以为selectionKey对象添加任何一个object。
但仅限一个，若没添加，attachment()会取出null。
运行后发现事件都获取到了，但文件仍然是一个接一个的下载。
原因是server触发write事件后创建fileChannel并一次写完。
事件响应的执行体太大，影响后面的执行。
非阻塞嘛，要得就是立即返回。
解决方法是分多次事件去读写，每次事件继续读写上一次事件的缓冲。
我可以好好使用一下这个attach()了。
首先我加了一个resolver类，我打算把他的实例加到attachment中去:
class ChannelResolver{
private FileChannel channel;
private ByteBuffer buffer;
private FileInputStream fis;
public ChannelResolver(String filePath){
try {
this.fis = new FileInputStream(filePath);
this.channel = this.fis.getChannel();
buffer = ByteBuffer.allocate(1024*100);
} catch (FileNotFoundException e) {
e.printStackTrace();
}
}
ByteBuffer readInto(){
try {
buffer.clear();
int i = channel.read(buffer);
buffer.flip();
if(i<0){
return null;
}
} catch (IOException e) {
e.printStackTrace();
}
return buffer;
}
}
将channel注册write事件后在return的selectionKey上attach一个实例。
然后在write事件中获取attachment进行读写:
public void run() {
System.out.println("server start...");
ServerSocketChannel serverChannel;
try {
serverChannel = ServerSocketChannel.open();
serverChannel.bind(new InetSocketAddress(8989));
serverChannel.configureBlocking(false);
Selector sel = Selector.open();
serverChannel.register(sel, SelectionKey.OP_ACCEPT);
ByteBuffer buffer = ByteBuffer.allocate(100*1024);
CharsetDecoder decoder = Charset.forName("UTF-8").newDecoder();
while(true){
sel.select();
Iterator selKeyItr = sel.selectedKeys().iterator();
while(selKeyItr.hasNext()){
SelectionKey key = selKeyItr.next();
selKeyItr.remove();
if(key.isAcceptable()){
System.out.println("server acceptable");
SocketChannel channel = ((ServerSocketChannel)key.channel()).accept();
channel.configureBlocking(false);
channel.register(sel, SelectionKey.OP_READ);
}else if(key.isReadable()){
System.out.println("server readable"+Thread.currentThread().getName());
SocketChannel channel = (SocketChannel) key.channel();
if(channel.read(buffer)>0){
buffer.flip();
CharBuffer clientBuffer = decoder.decode(buffer);
System.out.println("from client::"+clientBuffer.toString());
buffer.clear();
}
channel.register(sel, SelectionKey.OP_WRITE).attach(new ChannelResolver("D:/doc_backup.rar"));
}else if(key.isWritable()){
SocketChannel channel =(SocketChannel) key.channel();
if(key.attachment()!=null){
ChannelResolver resolver = (ChannelResolver)key.attachment();
buffer = resolver.readInto();
if(buffer!=null){
channel.write(buffer);
}
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
在handler的readInto()中已经进行了flip()，在这里就不用再flip()了。
相应地，client的读取也要改一下:
public void run() {
try {
System.out.println("client...");
SocketAddress addr = new InetSocketAddress(8989);
SocketChannel client = SocketChannel.open();
client.configureBlocking(false);
Selector sel = Selector.open();
client.register(sel, SelectionKey.OP_CONNECT);
CharsetEncoder encoder = Charset.forName("UTF-8").newEncoder();
ByteBuffer buffer= ByteBuffer.allocate(1024*500);
client.connect(addr);
while (true) {
sel.select();
Iterator selKeyItr = sel.selectedKeys().iterator();
while (selKeyItr.hasNext()) {
SelectionKey key = selKeyItr.next();
selKeyItr.remove();
if (key.isConnectable()) {
System.out.println("client connectble");
SocketChannel channel = (SocketChannel) key.channel();
channel.configureBlocking(false);
channel.finishConnect();
channel.write(encoder.encode(CharBuffer.wrap("start download")));
channel.register(sel, SelectionKey.OP_READ);
} else if (key.isReadable()) {
SocketChannel channel = (SocketChannel) key.channel();
if(channel.read(buffer)>0){
buffer.flip();
fc.write(buffer);
buffer.clear();
}else{
channel.close();
}
}
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
下面引用书本上的一段话：
[是基于事件驱动思想的，实现上通常采用Reactor模式，从程序角度而言，当发起IO的读写操作时，是非阻塞的；当socket有流可读或可写入socket时，操作系统会相应地通知应用程序进行处理，应用再将流读取到缓冲区或写入操作系统。对于网络IO而言，主要有连接建立、流读取和流写入三种事件。
AIO同样基于事件驱动思想，实现上通常采用Proactor模式。从程序角度而言，和NIO不同，当进行读写操作时，只须直接调用API的read或write方法即可。这两种方法均为异步的，对于读操作而言，当有流可读取时，操作系统会将可读的流传入read方法的缓冲区，并通知应用程序；对于写操作而言，当操作系统将write方法传递的流写入完毕时，操作系统主动通知应用程序。
较之NIO而言，AIO一方面简化了程序的编写，流的读取和写入都由操作系统来代替完成；另一方面省去了NIO中程序要遍历事件通知队列(selector)的代价。windows基于iocp、Linux基于epoll。]