狂奔 lion

自强不息

NIO学习之Web服务器示例

1 根据cpu core数量确定selector数量
2 用一个selector服务accept，其他selector按照core-1分配线程数运行
3 accept selector作为生产者把获得的请求放入队列
4 某个selector作为消费者从blocking queue中取出请求socket channel，并向自己注册
5 当获得read信号时，selector建立工作任务线程worker，并提交给系统线程池
6 worker线程排队后在线程池中执行，把协议头读入缓冲区，然后解析，处理，响应，关闭连接

import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class NioServer {
    private static NioServer svr = new NioServer();
    private final int numProcessors = Runtime.getRuntime().availableProcessors();
    private static LinkedBlockingQueue<SocketChannel> queue = new LinkedBlockingQueue<SocketChannel>();
    private static ExecutorService es = Executors.newFixedThreadPool(3);
    
    private NioServer(){}
    
    public static NioServer getInstance(){return svr;}
    
    public void init(){
        try {
            for(int i=0;i<numProcessors - 1;i++){
                Thread tk = new Thread(new Talker());
                tk.start();
            }
            
            Selector acceptSelector = Selector.open();

            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);

            InetSocketAddress isa = new InetSocketAddress("127.0.0.1", 8080);
            ssc.socket().bind(isa);

            ssc.register(acceptSelector, SelectionKey.OP_ACCEPT);

            while (acceptSelector.select() > 0) {
                Set<SelectionKey> readyKeys = acceptSelector.selectedKeys();
                Iterator<SelectionKey> i = readyKeys.iterator();

                while (i.hasNext()) {
                    SelectionKey sk = i.next();
                    i.remove();
                    if (sk.isAcceptable()) {
                        ServerSocketChannel nextReady = (ServerSocketChannel) sk
                                .channel();
                        SocketChannel s = nextReady.accept();
                        s.configureBlocking(false);
                        queue.offer(s);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    private static class Talker implements Runnable{
        private Selector se = null;

        
        public Talker(){
            try {
                this.se = Selector.open();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        
        public void addChannelIfAvailable(){
            try {
                if(queue.isEmpty())
                    return;
                SocketChannel sc = queue.poll();
                sc.register(se, SelectionKey.OP_READ);
            } catch (ClosedChannelException e) {
                e.printStackTrace();
            }
        }
        
        public void run() {
            try {
                while (true) {
                    int skOps = se.select(20);
                    addChannelIfAvailable();
                    if(skOps <= 0){
                        continue;
                    }
                    
                    Set<SelectionKey> readyKeys = se.selectedKeys();
                    Iterator<SelectionKey> i = readyKeys.iterator();
                    while (i.hasNext()) {
                        SelectionKey sk = i.next();
                        i.remove();
                        
                        if (sk.isValid() && sk.isReadable()) {
                            SocketChannel sc = (SocketChannel) sk.channel();
                            sc.configureBlocking(false);
                            
                            Worker worker = new Worker(sc);
                            es.execute(worker);    
                        }
                    }
                    Thread.sleep(300);
                }
            } catch (IOException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    private static class Worker implements Runnable{
        ByteBuffer bb = ByteBuffer.allocateDirect(1024);
        SocketChannel sc = null;
        
        public Worker(SocketChannel sc){
            Thread.currentThread().setName(this.toString());
            this.sc = sc;
        }
        
        public void run() {
            try {
                try{
                    sc.read(bb);
                }catch(IOException e){
                    e.printStackTrace();
                    sc.finishConnect();
                    sc.close();
                    return;
                }
                bb.flip();
                byte[] bs = new byte[bb.limit()];
                bb.get(bs);
                bb.clear();
                
                StringBuilder sb = new StringBuilder();
                sb.append("HTTP/1.1 200 OK").append("\n").append("Date:" + new Date()).append("\n");
                sb.append("Server:NIO Server By Yi Yang\n\n");
                bb.put(sb.toString().getBytes());
                
                bb.flip();
                sc.write(bb);
                bb.clear();
                
                FileInputStream is = new FileInputStream("E:/test.html");
                is.getChannel().transferTo(0, 1024, sc);
                
                is.close();
                sc.finishConnect();
                sc.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args) throws Exception {
        NioServer server = NioServer.getInstance();
        server.init();
    }
}
===============06/27/10 
如何解析header？，以行为单位读取，按照header敏感的关键字进行匹配 对于首行取得对方调用的方法GET/POST 地址 和协议版本 
然后根据用户的配置，和解析地址请求，获得响应的servlet，并把通过反射+默认构造函数构造这个servlet，解析地址参数等设置到对象httpservletrequest和httpservletresponse中，然后通过反射invoke对应的get/post/put/delete等方法，并把封装的两个对象作为参数传进去，同时在response的header中传递一个cookie作为session的依据。


 @2008 杨一. 版权所有. 保留所有权利

posted on 2010-06-25 19:19 杨一 阅读(2004) 评论(0)  编辑  收藏 所属分类: Java SE


只有注册用户登录后才能发表评论。
网站导航:
 
<2010年6月>
日一二三四五六
303112345
6789101112
13141516171819
20212223242526
27282930123
45678910

导航

公告

本人在blogjava上发表的文章及随笔除特别声明外均为原创或翻译，作品受知识产权法保护并被授权遵从 知识分享协议:署名-非商业性使用-相同方式共享 欢迎转载,请在转载时注明作者姓名(杨一)及出处(www.blogjava.net/yangyi)
/////////////////////////////////////////
我的访问者

常用链接

留言簿(5)

随笔分类(55)

随笔档案(55)

相册

Java

其他技术

生活

最新随笔

搜索

积分与排名

最新评论

阅读排行榜

评论排行榜

自强不息


 用心 - 珍惜时间，勇于创造