NIO Socket非阻塞模式

NIO主要原理和適用

NIO 有一個主要的類Selector,這個相似一個觀察者，只要咱們把須要探知的socketchannel告訴Selector,咱們接着作別的事情，當有 事件發生時，他會通知咱們，傳回一組SelectionKey,咱們讀取這些Key,就會得到咱們剛剛註冊過的socketchannel,而後，咱們從 這個Channel中讀取數據，放心，包準可以讀到，接着咱們能夠處理這些數據。

Selector內部原理實際是在作一個對所註冊的channel的輪詢訪問，不斷的輪詢(目前就這一個算法)，一旦輪詢到一個channel有所註冊的事情發生，好比數據來了，他就會站起來報告，交出一把鑰匙，讓咱們經過這把鑰匙來讀取這個channel的內容。

jdk供的無阻塞I/O（NIO）有效解決了多線程服務器存在的線程開銷問題，但在使用上略顯得複雜一些。在NIO中使用多線程，主要目的已不是爲了應對 每一個客戶端請求而分配獨立的服務線程，而是經過多線程充分使用用多個CPU的處理能力和處理中的等待時間，達到提升服務能力的目的。

這段時間在研究NIO，寫篇博客來記住學過的東西。仍是從最簡單的Hello World開始，client多線程請求server端，server接收client的名字，並返回Hello! ＋名字的字符格式給client。固然實際應用並不這麼簡單，實際多是訪問文件或者數據庫獲取信息返回給client。非阻塞的NIO有何神祕之處？

代 碼：

1）server端代碼

public class HelloWorldServer {

    static int BLOCK = 1024;
    static String name = "";
    protected Selector selector;
    protected ByteBuffer clientBuffer = ByteBuffer.allocate(BLOCK);
    protected CharsetDecoder decoder;
    static CharsetEncoder encoder = Charset.forName("GB2312").newEncoder();

    public HelloWorldServer(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();
                    process(key);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 處理事件
    protected void process(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()) { // 讀信息
            SocketChannel channel = (SocketChannel) key.channel();
            int count = channel.read(clientBuffer);
            if (count > 0) {
                clientBuffer.flip();
                CharBuffer charBuffer = decoder.decode(clientBuffer);
                name = charBuffer.toString();
                // System.out.println(name);
                SelectionKey sKey = channel.register(selector,
                        SelectionKey.OP_WRITE);
                sKey.attach(name);
            } else {
                channel.close();
            }

            clientBuffer.clear();
        } else if (key.isWritable()) { // 寫事件
            SocketChannel channel = (SocketChannel) key.channel();
            String name = (String) key.attachment();

            ByteBuffer block = encoder.encode(CharBuffer
                    .wrap("Hello !" + name));


            channel.write(block);

            //channel.close();

        }
    }

    public static void main(String[] args) {
        int port = 8888;
        try {
            HelloWorldServer server = new HelloWorldServer(port);
            System.out.println("listening on " + port);

            server.listen();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

2）client端代碼

public class HelloWorldClient {

    static int SIZE = 10;
    static InetSocketAddress ip = new InetSocketAddress("localhost", 8888);
    static CharsetEncoder encoder = Charset.forName("GB2312").newEncoder();

    static class Message implements Runnable {
        protected String name;
        String msg = "";

        public Message(String index) {
            this.name = index;
        }

        public void run() {
            try {
                long start = System.currentTimeMillis();
                //打開Socket通道
                SocketChannel client = SocketChannel.open();
                //設置爲非阻塞模式
                client.configureBlocking(false);
                //打開選擇器
                Selector selector = Selector.open();
                //註冊鏈接服務端socket動做
                client.register(selector, SelectionKey.OP_CONNECT);
                //鏈接
                client.connect(ip);
                //分配內存
                ByteBuffer buffer = ByteBuffer.allocate(8 * 1024);
                int total = 0;

                _FOR: for (;;) {
                    selector.select();
                    Iterator iter = selector.selectedKeys().iterator();

                    while (iter.hasNext()) {
                        SelectionKey key = (SelectionKey) iter.next();
                        iter.remove();
                        if (key.isConnectable()) {
                            SocketChannel channel = (SocketChannel) key
                                    .channel();
                            if (channel.isConnectionPending())
                                channel.finishConnect();
                            channel
                                    .write(encoder
                                            .encode(CharBuffer.wrap(name)));

                            channel.register(selector, SelectionKey.OP_READ);
                        } else if (key.isReadable()) {
                            SocketChannel channel = (SocketChannel) key
                                    .channel();
                            int count = channel.read(buffer);
                            if (count > 0) {
                                total += count;
                                buffer.flip();

                                while (buffer.remaining() > 0) {
                                    byte b = buffer.get();
                                    msg += (char) b;

                                }

                                buffer.clear();
                            } else {
                                client.close();
                                break _FOR;
                            }
                        }
                    }
                }
                double last = (System.currentTimeMillis() - start) * 1.0 / 1000;
                System.out.println(msg + "used time :" + last + "s.");
                msg = "";
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException {

        String names[] = new String[SIZE];

        for (int index = 0; index < SIZE; index++) {
            names[index] = "jeff[" + index + "]";
            new Thread(new Message(names[index])).start();
        }

    }
}