Java 网络编程 —— 实现非阻塞式的服务器

public class EchoServer {
    
	private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private ExecutorService executorService; //线程池
    private static final int POOL_MULTIPLE = 4; //线程池中工作线程的数目
    
    public EchoServer() throws IOException {
        //创建一个线程池
        executorService = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors() * POOL_MULTIPLE);
        //创建一个ServerSocketChannel对象
        serverSocketChannel = ServerSocketChannel.open();
        //使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时，可以顺利绑定相同的端口
        serverSocketChannel.socket().setReuseAddress(true);
        //把服务器进程与一个本地端口绑定
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        System.out.println("服务器启动");
    }
    
    public void service() {
        while (true) {
            SocketChannel socketChannel = null;
            try {
                socketChannel = serverSocketChannel.accept();
                //处理客户连接
                executorService.execute(new Handler(socketChannel));
            } catch(IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String args[])throws IOException {
        new EchoServer().service();
    }
    
    //处理客户连按
    class Handler implements Runnable {

        private SocketChannel socketChannel;
		
        public Handler(SocketChannel socketChannel) {
            this.socketChannel = socketChannel;
        }
        
        public void run() {
            handle(socketChannel);
        }
        
        public void handle(SocketChannel socketChannel) {
            try {
                //获得与socketChannel关联的Socket对象
                Socket socket = socketChannel.socket();
                System.out.println("接收到客户连接，来自:" + socket.getInetAddress() + ":" + socket.getPort());
                
                BufferedReader br = getReader(socket);
                PrintWriter pw = getWriter(socket);
                
                String msg = null;
                while ((msg = br.readLine()) != null) {
                    System.out.println(msg);
                    pw.println(echo(msg));
                    if (msg.equals("bye")) {
                        break;
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    if(socketChannel != null) {
                        socketChannel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    } 
    
    private PrintWriter getWriter(Socket socket) throws IOException {
        OutputStream socketOut = socket.getOutputStream();
        return new PrintWriter(socketOut,true);
    }
    
    private BufferedReader getReader(Socket socket) throws IOException {
        InputStream socketIn = socket.getInputStream();
        return new BufferedReader(new InputStreamReader(socketIn));
    }
    
    public String echo(String msg) {
        return "echo:" + msg;
    }
}

// 创建一个Selector对象
selector = Selector.open();
//创建一个ServerSocketChannel对象
serverSocketChannel = ServerSocketChannel.open();
//使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时
//可以顺利绑定到相同的端口
serverSocketChannel.socket().setReuseAddress(true);
//使ServerSocketChannel工作于非阻塞模式
serverSocketChannel.configureBlocking(false):
//把服务器进程与一个本地端口绑定
serverSocketChannelsocket().bind(new InetSocketAddress(port));

public void service() throws IOException {
    serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
    //第1层while循环
    while(selector.select() > 0) {
        //获得Selector的selected-keys集合
        Set readyKeys = selector.selectedKeys();
        Iterator it = readyKeys.iterator();
        //第2层while循环
        while (it.hasNext()) {
            SelectionKey key = null;
            //处理SelectionKey
            try {
                //取出一个SelectionKey
                key = (SelectionKey) it.next();
                //把 SelectionKey从Selector 的selected-key 集合中删除
                it.remove();
                1f (key.isAcceptable()) { 处理接收连接就绪事件; }
                if (key.isReadable()) { 处理读就绪水件; }
                if (key.isWritable()) { 处理写就绪事件; }
            } catch(IOException e) {
                e.printStackTrace();
                try {
                    if(key != null) {
                        //使这个SelectionKey失效
                        key.cancel();
                        //关闭与这个SelectionKey关联的SocketChannel
                        key.channel().close();
                    }
                } catch(Exception ex) { 
                    e.printStackTrace();
                }
            }
        }
    }
}

if (key.isAcceptable()) {
    //获得与SelectionKey关联的ServerSocketChannel
    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
    //获得与客户连接的SocketChannel
    SocketChannel socketChannel = (SocketChannel) ssc.accept();
    //把Socketchannel设置为非阻塞模式
    socketChannel.configureBlocking(false);
    //创建一个用于存放用户发送来的数据的级冲区
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    //Socketchannel向Selector注册读就绪事件和写就绪事件
    socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
}

public void receive(SelectionKey key) throws IOException {
    //获得与SelectionKey关联的附件
    ByteBuffer buffer = (ByteBuffer) key.attachment();
    //获得与SelectionKey关联的Socketchannel
    SocketChannel socketChannel = (SocketChannel)key.channel();
    //创建一个ByteBuffer用于存放读到的数据
    ByteBuffer readBuff = ByteBuffer.allocate(32);
    socketChannel.read(readBuff);
    readBuff.flip();
    //把buffer的极限设为容量
    buffer.limit(buffer.capacity());
    //把readBuff中的内容拷贝到buffer
    buffer.put(readBuff);
}

public void send(SelectionKey key) throws IOException {
    //获得与SelectionKey关联的ByteBuffer
    ByteBuffer buffer = (ByteBuffer) key.attachment();
    //获得与SelectionKey关联的SocketChannel
    SocketChannel socketChannel = (SocketChannel) key.channel();
    buffer.flip();
    //按照GBK编码把buffer中的字节转换为字符串
    String data = decode(buffer);
    //如果还没有读到一行数据就返回
    if(data.indexOf("\r\n") == -1)
        return;
    //截取一行数据
    String outputData = data.substring(0, data.indexOf("\n") + 1);
    //把输出的字符串按照GBK编码转换为字节，把它放在outputBuffer中
    ByteBuffer outputBuffer = encode("echo:" + outputData);
    //输出outputBuffer的所有字节
    while(outputBuffer,hasRemaining())
        socketChannel.write(outputBuffer);
    //把outputData字符审按照GBK编码，转换为字节，把它放在ByteBuffer
    ByteBuffer temp = encode(outputData);
    //把buffer的位置设为temp的极限
    buffer.position(temp.limit()):
    //删除buffer已经处理的数据
    buffer.compact();
    //如果已经输出了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketChannel
    if(outputData.equals("bye\r\n")) {
        key.cancel();
        socketChannel.close();
    }
}

public class EchoServer {
    
	private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private Selector selector;
    private Charset charset = Charset.forName("GBK");

	public EchoServer() throws IOException {
        // 创建一个Selector对象
        selector = Selector.open();
        //创建一个ServerSocketChannel对象
        serverSocketChannel = ServerSocketChannel.open();
        //使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时
        //可以顺利绑定到相同的端口
        serverSocketChannel.socket().setReuseAddress(true);
        //使ServerSocketChannel工作于非阻塞模式
        serverSocketChannel.configureBlocking(false):
        //把服务器进程与一个本地端口绑定
        serverSocketChannelsocket().bind(new InetSocketAddress(port));
    }
    
    public void service() throws IOException {
        serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
        //第1层while循环
        while(selector.select() > 0) {
            //获得Selector的selected-keys集合
            Set readyKeys = selector.selectedKeys();
            Iterator it = readyKeys.iterator();
            //第2层while循环
            while (it.hasNext()) {
                SelectionKey key = null;
                //处理SelectionKey
                try {
                    //取出一个SelectionKey
                    key = (SelectionKey) it.next();
                    //把 SelectionKey从Selector 的selected-key 集合中删除
                    it.remove();
                    1f (key.isAcceptable()) {
                         //获得与SelectionKey关联的ServerSocketChannel
                        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                        //获得与客户连接的SocketChannel
                        SocketChannel socketChannel = (SocketChannel) ssc.accept();
                        //把Socketchannel设置为非阻塞模式
                        socketChannel.configureBlocking(false);
                        //创建一个用于存放用户发送来的数据的级冲区
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        //Socketchannel向Selector注册读就绪事件和写就绪事件
                        socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
                    }
                    if (key.isReadable()) { receive(key); }
                    if (key.isWritable()) { send(key); }
                } catch(IOException e) {
                    e.printStackTrace();
                    try {
                        if(key != null) {
                            //使这个SelectionKey失效
                            key.cancel();
                            //关闭与这个SelectionKey关联的SocketChannel
                            key.channel().close();
                        }
                    } catch(Exception ex) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    }
    
    public void receive(SelectionKey key) throws IOException {
        //获得与SelectionKey关联的附件
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        //获得与SelectionKey关联的Socketchannel
        SocketChannel socketChannel = (SocketChannel)key.channel();
        //创建一个ByteBuffer用于存放读到的数据
        ByteBuffer readBuff = ByteBuffer.allocate(32);
        socketChannel.read(readBuff);
        readBuff.flip();
        //把buffer的极限设为容量
        buffer.limit(buffer.capacity());
        //把readBuff中的内容拷贝到buffer
        buffer.put(readBuff);
    }
    
    public void send(SelectionKey key) throws IOException {
        //获得与SelectionKey关联的ByteBuffer
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        //获得与SelectionKey关联的SocketChannel
        SocketChannel socketChannel = (SocketChannel) key.channel();
        buffer.flip();
        //按照GBK编码把buffer中的字节转换为字符串
        String data = decode(buffer);
        //如果还没有读到一行数据就返回
        if(data.indexOf("\r\n") == -1)
            return;
        //截取一行数据
        String outputData = data.substring(0, data.indexOf("\n") + 1);
        //把输出的字符串按照GBK编码转换为字节，把它放在outputBuffer中
        ByteBuffer outputBuffer = encode("echo:" + outputData);
        //输出outputBuffer的所有字节
        while(outputBuffer,hasRemaining())
            socketChannel.write(outputBuffer);
        //把outputData字符审按照GBK编码，转换为字节，把它放在ByteBuffer
        ByteBuffer temp = encode(outputData);
        //把buffer的位置设为temp的极限
        buffer.position(temp.limit()):
        //删除buffer已经处理的数据
        buffer.compact();
        //如果已经输出了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketChannel
        if(outputData.equals("bye\r\n")) {
            key.cancel();
            socketChannel.close();
        }
    }
    
    //解码
    public String decode(ByteBuffer buffer) {
        CharBuffer charBuffer = charset.decode(buffer);
        return charBuffer.toStrinq();
    }
    
    //编码
    public ByteBuffer encode(String str) {
        return charset.encode(str);
    }
    
    public static void main(String args[])throws Exception {
        EchoServer server = new EchoServer();
        server.service();
    }
}

public class EchoServer {
    
	private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private Selector selector = null;
    private Charset charset = Charset.forName("GBK");

	public EchoServer() throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().setReuseAddress(true);
        serverSocketChannelsocket().bind(new InetSocketAddress(port));
    }
    
    public void accept() {
        while(true) {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                socketChannel.configureBlocking(false);
                
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                synchronized(gate) {
                    selector.wakeup();
                    socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
                }
            } catch(IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    private Object gate=new Object();
    
    public void service() throws IOException {
        while(true) {
            synchronized(gate){}
            int n = selector.select();
            if(n == 0) continue;
            Set readyKeys = selector.selectedKeys();
            Iterator it = readyKeys.iterator();
            while (it.hasNext()) {
                SelectionKey key = null;
                try {
    				it.remove();
                    if (key.isReadable()) {
                        receive(key);
                    }
                    if (key.isWritable()) {
                        send(key);
                    }
                } catch(IOException e) {
                    e.printStackTrace();
                    try {
                        if(key != null) {
                            key.cancel();
                            key.channel().close();
                        }
                    } catch(Exception ex) { e.printStackTrace(); }
                }
            }
        }
    }
    
    public void receive(SelectionKey key) throws IOException {
        ...
    }
    
    public void send(SelectionKey key) throws IOException {
        ...
    }
    
    public String decode(ByteBuffer buffer) {
        ...
    }
    
    public ByteBuffer encode(String str) {
        ...
    }
    
    public static void main(String args[])throws Exception {
        final EchoServer server = new EchoServer();
        Thread accept = new Thread() {
            public void run() {
                server.accept();
            }
        };
        accept.start();
		server.service();
    }
}