E81086713E446D36F62B2AA2A3502B5EB155

Java杂家

杂七杂八。。。一家之言

BlogJava 首页 新随笔 联系 聚合 管理
  141 Posts :: 1 Stories :: 174 Comments :: 0 Trackbacks

JDK7的NIO2特性或许是我最期待的,我一直想基于它写一个高性能的Java Http Server.现在这个想法终于可以实施了。
本人基于目前最新的JDK7 b76开发了一个HTTP Server性能确实不错。
在windows平台上NIO2采用AccpetEx来异步接受连接,并且读写全都关联到IOCP完成端口。不仅如此,为了方便开发者使用,连IOCP工作线程都封装好了,你只要提供线程池就OK。

但是要注意,IOCP工作线程的线程池必须是 Fix的,因为你发出的读写请求都关联到相应的线程上,如果线程死了,那读写完成情况是不知道的。

作为一个Http Server,传送文件是必不可少的功能,那一般文件的传送都是要把程序里的buffer拷贝到内核的buffer,由内核发送出去的。windows平台上为这种情况提供了很好的解决方案,使用TransmitFile接口

BOOL TransmitFile(
    SOCKET hSocket,
    HANDLE hFile,
    DWORD nNumberOfBytesToWrite,
    DWORD nNumberOfBytesPerSend,
    LPOVERLAPPED lpOverlapped,
    LPTRANSMIT_FILE_BUFFERS lpTransmitBuffers,
    DWORD dwFlags
);

你只要把文件句柄发送给内核就行了,内核帮你搞定其余的,真正做到Zero-Copy.
但是很不幸,NIO2里AsynchronousSocketChannel没有提供这样的支持。而为HTTP Server的性能考量,本人只好自己增加这个支持。

要无缝支持,这个必须得表现的跟 Read /Write一样,有完成的通知,通知传送多少数据,等等。

仔细读完sun的IOCP实现以后发现这部分工作他们封装得很好,基本只要往他们的框架里加东西就好了。
为了能访问他们的框架代码,我定义自己的TransmitFile支持类在sun.nio.ch包里,以获得最大的权限。

package sun.nio.ch;

import java.io.IOException;
import java.lang.reflect.Field;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.CompletionHandler;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.WritePendingException;
import java.util.concurrent.Future;


/**

 * 
@author Yvon
 * 
 
*/
public class WindowsTransmitFileSupport {
   
   //Sun's NIO2 channel  implementation class
    
private WindowsAsynchronousSocketChannelImpl channel;
   
    //nio2 framework core data structure
    PendingIoCache ioCache;

   //some field retrieve from sun channel implementation class
    
private Object writeLock;
    
private Field writingF;
    
private Field writeShutdownF;
    
private Field writeKilledF; // f

    WindowsTransmitFileSupport()
    {
        
//dummy one for JNI code
    }

    
/**
     * 
     
*/
    
public WindowsTransmitFileSupport(
            AsynchronousSocketChannel
             channel) {

        
this.channel = (WindowsAsynchronousSocketChannelImpl)channel;
        
try {
        // Initialize the fields
            Field f 
= WindowsAsynchronousSocketChannelImpl.class
                    .getDeclaredField(
"ioCache");
            f.setAccessible(
true);
            ioCache 
= (PendingIoCache) f.get(channel);
            f 
= AsynchronousSocketChannelImpl.class
                    .getDeclaredField(
"writeLock");
            f.setAccessible(
true);
            writeLock 
= f.get(channel);
            writingF 
= AsynchronousSocketChannelImpl.class
                    .getDeclaredField(
"writing");
            writingF.setAccessible(
true);

            writeShutdownF 
= AsynchronousSocketChannelImpl.class
                    .getDeclaredField(
"writeShutdown");
            writeShutdownF.setAccessible(
true);

            writeKilledF 
= AsynchronousSocketChannelImpl.class
                    .getDeclaredField(
"writeKilled");
            writeKilledF.setAccessible(
true);

        } 
catch (NoSuchFieldException e) {
            
// TODO Auto-generated catch block
            e.printStackTrace();
        } 
catch (SecurityException e) {
            
// TODO Auto-generated catch block
            e.printStackTrace();
        } 
catch (IllegalArgumentException e) {
            
// TODO Auto-generated catch block
            e.printStackTrace();
        } 
catch (IllegalAccessException e) {
            
// TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    
    
/**
     * Implements the task to initiate a write and the handler to consume the
     * result when the send file completes.
     
*/
    
private class SendFileTask<V, A> implements Runnable, Iocp.ResultHandler {
        
private final PendingFuture<V, A> result;
        
private final long file;//file is windows file HANDLE

        SendFileTask(
long file, PendingFuture<V, A> result) {
            
this.result = result;
            
this.file = file;
        }

    

        @Override
        
// @SuppressWarnings("unchecked")
        public void run() {
            
long overlapped = 0L;
            
boolean pending = false;
            
boolean shutdown = false;

            
try {
                channel.begin();

        

                
// get an OVERLAPPED structure (from the cache or allocate)
                overlapped = ioCache.add(result);
                
int n = transmitFile0(channel.handle, file, overlapped);
                
if (n == IOStatus.UNAVAILABLE) {
                    
// I/O is pending
                    pending = true;
                    
return;
                }
                
if (n == IOStatus.EOF) {
                    
// special case for shutdown output
                    shutdown = true;
                    
throw new ClosedChannelException();
                }
                
// write completed immediately
                throw new InternalError("Write completed immediately");
            } 
catch (Throwable x) {
                
// write failed. Enable writing before releasing waiters.
                channel.enableWriting();
                
if (!shutdown && (x instanceof ClosedChannelException))
                    x 
= new AsynchronousCloseException();
                
if (!(x instanceof IOException))
                    x 
= new IOException(x);
                result.setFailure(x);
            } 
finally {
                
// release resources if I/O not pending
                if (!pending) {
                    
if (overlapped != 0L)
                        ioCache.remove(overlapped);
                
                }
                channel.end();
            }

            
// invoke completion handler
            Invoker.invoke(result);
        }

        

        
/**
         * Executed when the I/O has completed
         
*/
        @Override
        @SuppressWarnings(
"unchecked")
        
public void completed(int bytesTransferred, boolean canInvokeDirect) {
    

            
// release waiters if not already released by timeout
            synchronized (result) {
                
if (result.isDone())
                    
return;
                channel.enableWriting();

                result.setResult((V) Integer.valueOf(bytesTransferred));

            }
            
if (canInvokeDirect) {
                Invoker.invokeUnchecked(result);
            } 
else {
                Invoker.invoke(result);
            }
        }

        @Override
        
public void failed(int error, IOException x) {
            
// return direct buffer to cache if substituted
        

            
// release waiters if not already released by timeout
            if (!channel.isOpen())
                x 
= new AsynchronousCloseException();

            
synchronized (result) {
                
if (result.isDone())
                    
return;
                channel.enableWriting();
                result.setFailure(x);
            }
            Invoker.invoke(result);
        }

    }

    
public <extends Number, A> Future<V> sendFile(long file, A att,
            CompletionHandler
<V, ? super A> handler) {

        
boolean closed = false;
        
if (channel.isOpen()) {
            
if (channel.remoteAddress == null)
                
throw new NotYetConnectedException();

            
            
// check and update state
            synchronized (writeLock) {
                
try{
                
if (writeKilledF.getBoolean(channel))
                    
throw new IllegalStateException(
                            
"Writing not allowed due to timeout or cancellation");
                
if (writingF.getBoolean(channel))
                    
throw new WritePendingException();
                
if (writeShutdownF.getBoolean(channel)) {
                    closed 
= true;
                } 
else {
                    writingF.setBoolean(channel, 
true);
                }
                }
catch(Exception e)
                {
                    IllegalStateException ise
=new IllegalStateException(" catch exception when write");
                    ise.initCause(e);
                    
throw ise;
                }
            }
        } 
else {
            closed 
= true;
        }

        
// channel is closed or shutdown for write
        if (closed) {
            Throwable e 
= new ClosedChannelException();
            
if (handler == null)
                
return CompletedFuture.withFailure(e);
            Invoker.invoke(channel, handler, att, 
null, e);
            
return null;
        }



        
return implSendFile(file,att,handler);
    }


    
<extends Number, A> Future<V> implSendFile(long file, A attachment,
            CompletionHandler
<V, ? super A> handler) {
        
// setup task
        PendingFuture<V, A> result = new PendingFuture<V, A>(channel, handler,
                attachment);
        SendFileTask
<V,A> sendTask=new SendFileTask<V,A>(file,result);
        result.setContext(sendTask);
        
// initiate I/O (can only be done from thread in thread pool)
        
// initiate I/O
        if (Iocp.supportsThreadAgnosticIo()) {
            sendTask.run();
        } 
else {
            Invoker.invokeOnThreadInThreadPool(channel, sendTask);
        }
        
return result;
    }
    
    
private native int transmitFile0(long handle, long file,
            
long overlapped);
    
}

这个操作跟默认实现的里的write操作是很像的,只是最后调用的本地方法不一样。。

接下来,我们怎么使用呢,这个类是定义在sun的包里的,直接用的话,会报IllegalAccessError,因为我们的类加载器跟初始化加载器是不一样的。
解决办法一个是通过启动参数-Xbootclasspath,让我们的包被初始加载器加载。我个人不喜欢这种办法,所以就采用JNI来定义我们的windows TransmitFile支持类。

这样我们的工作算是完成了,注意,发送文件的时候传得是文件句柄,这样做的好处是你可以更好的控制,一般是在发送前,打开文件句柄,完成后在回调通知方法里关闭文件句柄。



有兴趣的同学可以看看我的HTTP server项目:
http://code.google.com/p/jabhttpd/

目前基本功能实现得差不多,做了些简单的测试,性能比较满意。这个服务器不打算支持servlet api,基本是专门给做基于长连接模式通信的定做的。






posted on 2009-11-29 15:19 DoubleH 阅读(2569) 评论(2)  编辑  收藏

Feedback

# re: JDK7 NIO2 实践: 增加 TransmitFile支持 2009-12-01 12:14 创意礼品批发
非常好的文章,谢谢楼主分享!··  回复  更多评论
  

# re: JDK7 NIO2 实践: 增加 TransmitFile支持[未登录] 2011-12-19 10:19 VV
老师啊 jabhttpd 给一份,学习一下吧。。  回复  更多评论
  


只有注册用户登录后才能发表评论。


网站导航: