java学习——NIO篇2

  在前面写的nio学习中是采取的简单的demo，后来在网上看到一个demo，一个用来多线程下载的demo，在这个demo里才会体现nio的高性能之处。充分体现nio的优势。

场景：一个服务端，然后100个客户端同时连接服务端，然后下载一个文件下来。服务端只需要一个连接来处理这些来自客户端的连接。其中对文件的传递采用的是nio中的FileChannel 。很好用的一个东西。

 

忘记从哪里转过来的代码了。。。

 

服务端：

 

import java.io.FileInputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.util.Iterator;

public class NIOServer {
    static int BLOCK = 4096;

    // 处理与客户端的交互
    public class HandleClient {
        protected FileChannel channel;
        protected ByteBuffer buffer;

        public HandleClient() throws IOException {
            this.channel = new FileInputStream(filename).getChannel();
            this.buffer = ByteBuffer.allocate(BLOCK);
        }

        public ByteBuffer readBlock() {
            try {
                buffer.clear();
                int count = channel.read(buffer);
                buffer.flip();
                if (count <= 0)
                    return null;
            } catch (IOException e) {
                e.printStackTrace();
            }
            return buffer;
        }

        public void close() {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    protected Selector selector;
    protected String filename = "d:\\test.jpg"; // a big file
    protected ByteBuffer clientBuffer = ByteBuffer.allocate(BLOCK);
    protected CharsetDecoder decoder;

    public NIOServer(int port) throws IOException {
        selector = this.getSelector(port);
        Charset charset = Charset.forName("GB2312");
        decoder = charset.newDecoder();
    }

    // 获取Selector
    protected Selector getSelector(int port) throws IOException {
        ServerSocketChannel server = ServerSocketChannel.open();
        Selector sel = Selector.open();
        server.socket().bind(new InetSocketAddress(port));
        server.configureBlocking(false);
        server.register(sel, SelectionKey.OP_ACCEPT);
        return sel;
    }

    // 监听端口
    public void listen() {
        try {
            for (;;) {
                selector.select();
                Iterator iter = selector.selectedKeys().iterator();
                while (iter.hasNext()) {
                    SelectionKey key = (SelectionKey) iter.next();
                    iter.remove();
                    handleKey(key);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 处理事件
    protected void handleKey(SelectionKey key) throws IOException {
        if (key.isAcceptable()) { // 接收请求
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel channel = server.accept();
            channel.configureBlocking(false);
            channel.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) { // 读信息
            System.out.println("能读");
            SocketChannel channel = (SocketChannel) key.channel();
            int count = channel.read(clientBuffer);
            if (count > 0) {
                clientBuffer.flip();
                CharBuffer charBuffer = decoder.decode(clientBuffer);
                System.out.println("Client >>" + charBuffer.toString());
                SelectionKey wKey = channel.register(selector, SelectionKey.OP_WRITE);
                wKey.attach(new HandleClient());
            } else
                channel.close();
            clientBuffer.clear();
        } else if (key.isWritable()) { // 写事件
            System.out.println("能写");
            SocketChannel channel = (SocketChannel) key.channel();
            HandleClient handle = (HandleClient) key.attachment();
            ByteBuffer block = handle.readBlock();
            if (block != null)
                channel.write(block);
            else {
                handle.close();
                channel.close();
            }
        }
    }

    public static void main(String[] args) {
        int port = 12345;
        try {
            NIOServer server = new NIOServer(port);
            System.out.println("Listernint on " + port);
            while (true) {
                server.listen();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}

 

 

客户端：

 

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetEncoder;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NIOClient {
    static int SIZE = 10;
    static InetSocketAddress ip = new InetSocketAddress("localhost", 12345);
    static CharsetEncoder encoder = Charset.forName("GB2312").newEncoder();
    static FileChannel[] filechannel = new FileChannel[SIZE];

    static class Download implements Runnable {
        protected int index;

        public Download(int index) {
            this.index = index;
        }

        public void run() {
            try {
                long start = System.currentTimeMillis();
                SocketChannel client = SocketChannel.open();
                client.configureBlocking(false);
                Selector selector = Selector.open();
                client.register(selector, SelectionKey.OP_CONNECT);
                client.connect(ip);
                ByteBuffer buffer = ByteBuffer.allocate(8 * 1024);
                int total = 0;
                boolean flag = true;
                while (flag) {
                    selector.select();
                    Iterator iter = selector.selectedKeys().iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        iter.remove();
                        if (key.isConnectable()) {
                            SocketChannel channel = (SocketChannel) key.channel();
                            if (channel.isConnectionPending())
                                channel.finishConnect();
                            channel.write(encoder.encode(CharBuffer.wrap("Hello from " + index)));
                            channel.register(selector, SelectionKey.OP_READ);
                        } else if (key.isReadable()) {
                            SocketChannel channel = (SocketChannel) key.channel();
                            int count = channel.read(buffer);
                            if (count > 0) {
                                writeToFile(buffer, index);
                                total += count;
                                buffer.clear();
                            } else {
                                filechannel[index].close();
                                client.close();
                                flag = false;
                            }
                        }
                    }
                }
                double last = (System.currentTimeMillis() - start) * 1.0 / 1000;
                System.out.println("Thread " + index + " downloaded " + total + "bytes in " + last + "s.");
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static synchronized void writeToFile(ByteBuffer bb, int index) {
        try {
            if (filechannel[index] == null) {
                filechannel[index] = new FileOutputStream(new File("test" + index + ".jpg")).getChannel();
            }
            bb.flip();
            filechannel[index].write(bb);

        } catch (FileNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws IOException {
        ExecutorService exec = Executors.newFixedThreadPool(SIZE);
        for (int index = 0; index < SIZE; index++) {
            exec.execute(new Download(index));
        }
        exec.shutdown();
    }
}