import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
public class NioServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        // 打開Channel服務(wù)端并綁定監(jiān)聽一個(gè)端口
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(8459));
        ssc.configureBlocking(false);
        // 打開多路復(fù)用器 并注冊(cè)到 ServerSocketChannel 并監(jiān)聽連接事件
        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        System.out.println("服務(wù)器已啟動(dòng)...");
        while (true) {
            // 如果沒有事件發(fā)生 則select() 處于阻塞狀態(tài)
            selector.select();
            // 發(fā)生事件
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = selectionKeys.iterator();
            // 處理事件
            while (iterator.hasNext()) {
                // 拿到具體事件
                SelectionKey selectionKey = iterator.next();
                // 判斷事件的類型
                if (selectionKey.isAcceptable()) {
                    System.out.println("客戶端連接事件");
                    SocketChannel channel = ssc.accept();
                    channel.configureBlocking(false);
                    channel.register(selector, SelectionKey.OP_READ);
                    if (channel.finishConnect()) {
                        System.out.println("完成連接");
                    }
                } else if (selectionKey.isReadable()) {
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int read = sc.read(buffer);
                    System.out.println("收到的消息:" + new String(buffer.array(), 0, read));
                    // 響應(yīng)客戶端  這里可有可無
                    buffer.clear();
                    buffer.put("已收到消息".getBytes());
                    // 將緩沖區(qū)各標(biāo)志復(fù)位，因?yàn)橄蚶锩鎝ut了數(shù)據(jù)標(biāo)志被改變要想從中讀取數(shù)據(jù)發(fā)向服務(wù)器，就要復(fù)位
                    buffer.flip();
                    sc.write(buffer);
                    // 設(shè)置監(jiān)聽事件的集合 這里把寫入事件加入
                    selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_WRITE);
                    System.out.println("服務(wù)器向客戶端發(fā)送已確認(rèn)消息");
                } else if (selectionKey.isWritable()) {
                    System.out.println("觸發(fā)往客戶端寫入數(shù)據(jù)事件");
                    // 發(fā)送完了就取消監(jiān)聽寫事件，否則會(huì)無限循環(huán)觸發(fā)寫事件
                    selectionKey.interestOps(selectionKey.interestOps() & ~SelectionKey.OP_WRITE);
                }
                // 事件完成后，將其移除
                iterator.remove();
            }
        }
    }

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
import java.util.Set;
public class NioClient {
    public static void main(String[] args) throws IOException {
        //打開選擇器
        Selector selector = Selector.open();
        //打開通道
        SocketChannel socketChannel = SocketChannel.open();
        //配置非阻塞模型
        socketChannel.configureBlocking(false);
        //連接遠(yuǎn)程主機(jī)
        socketChannel.connect(new InetSocketAddress("127.0.0.1", 8459));
        //注冊(cè)事件
        socketChannel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ );
        //循環(huán)處理
        new Thread(() -> {
            while (true) {
                try {
                    selector.select();
                    Set<SelectionKey> keys = selector.selectedKeys();
                    Iterator<SelectionKey> iter = keys.iterator();
                    while (iter.hasNext()) {
                        SelectionKey key = iter.next();
                        if (key.isConnectable()) {
                            //連接建立或者連接建立不成功
                            SocketChannel channel = (SocketChannel) key.channel();
                            //完成連接的建立
                            if (channel.finishConnect()) {
                                System.out.println("完成連接");
                            }
                        } else if (key.isReadable()) {
                            SocketChannel channel = (SocketChannel) key.channel();
                            ByteBuffer buffer = ByteBuffer.allocate(1024);
                            buffer.clear();
                            channel.read(buffer);
                            System.out.println("客戶端收到消息:" + new String(buffer.array()));
                            key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
                        } else if (key.isWritable()) {
                            System.out.println("客戶端向服務(wù)端寫入數(shù)據(jù)");
                            // 設(shè)置監(jiān)聽事件的集合 這里把寫入事件加入
                            key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                        }
                        iter.remove();
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                    break;
                }
            }
        }).start();
        Scanner scanner = new Scanner(System.in);
        while (true) {
            System.out.println("請(qǐng)輸入要發(fā)送的字符串");
            String str = scanner.nextLine();
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            buffer.put(str.getBytes());
            buffer.flip();
            socketChannel.write(buffer);
        }
    }
}

// Reactor線程  
package server;  
import java.io.IOException;  
import java.net.InetSocketAddress;  
import java.nio.channels.SelectionKey;  
import java.nio.channels.Selector;  
import java.nio.channels.ServerSocketChannel;  
import java.util.Iterator;  
import java.util.Set;  
public class TCPReactor implements Runnable {  
    private final ServerSocketChannel ssc;  
    private final Selector selector;  
    public TCPReactor(int port) throws IOException {  
        selector = Selector.open();  
        ssc = ServerSocketChannel.open();  
        InetSocketAddress addr = new InetSocketAddress(port);  
        ssc.socket().bind(addr); // 在ServerSocketChannel綁定監(jiān)聽端口  
        ssc.configureBlocking(false); // 設(shè)置ServerSocketChannel為非阻塞  
        SelectionKey sk = ssc.register(selector, SelectionKey.OP_ACCEPT); // ServerSocketChannel向selector註冊(cè)一個(gè)OP_ACCEPT事件，然後返回該通道的key  
        sk.attach(new Acceptor(selector, ssc)); // 給定key一個(gè)附加的Acceptor對(duì)象  
    }  
    @Override  
    public void run() {  
        while (!Thread.interrupted()) { // 在線程被中斷前持續(xù)運(yùn)行  
            System.out.println("Waiting for new event on port: " + ssc.socket().getLocalPort() + "...");  
            try {  
                if (selector.select() == 0) // 若沒有事件就緒則不往下執(zhí)行  
                    continue;  
            } catch (IOException e) {  
                // TODO Auto-generated catch block  
                e.printStackTrace();  
            }  
            Set<SelectionKey> selectedKeys = selector.selectedKeys(); // 取得所有已就緒事件的key集合  
            Iterator<SelectionKey> it = selectedKeys.iterator();  
            while (it.hasNext()) {  
                dispatch((SelectionKey) (it.next())); // 根據(jù)事件的key進(jìn)行調(diào)度  
                it.remove();  
            }  
        }  
    }  
    /* 
     * name: dispatch(SelectionKey key) 
     * description: 調(diào)度方法，根據(jù)事件綁定的對(duì)象開新線程 
     */  
    private void dispatch(SelectionKey key) {  
        Runnable r = (Runnable) (key.attachment()); // 根據(jù)事件之key綁定的對(duì)象開新線程  
        if (r != null)  
            r.run();  
    }  
}  

// 接受連線請(qǐng)求線程  
package server;  
import java.io.IOException;  
import java.nio.channels.SelectionKey;  
import java.nio.channels.Selector;  
import java.nio.channels.ServerSocketChannel;  
import java.nio.channels.SocketChannel;  
public class Acceptor implements Runnable {  
    private final ServerSocketChannel ssc;  
    private final Selector selector;  
    public Acceptor(Selector selector, ServerSocketChannel ssc) {  
        this.ssc=ssc;  
        this.selector=selector;  
    }  
    @Override  
    public void run() {  
        try {  
            SocketChannel sc= ssc.accept(); // 接受client連線請(qǐng)求  
            System.out.println(sc.socket().getRemoteSocketAddress().toString() + " is connected.");  
            if(sc!=null) {  
                sc.configureBlocking(false); // 設(shè)置為非阻塞  
                SelectionKey sk = sc.register(selector, SelectionKey.OP_READ); // SocketChannel向selector註冊(cè)一個(gè)OP_READ事件，然後返回該通道的key  
                selector.wakeup(); // 使一個(gè)阻塞住的selector操作立即返回  
                sk.attach(new TCPHandler(sk, sc)); // 給定key一個(gè)附加的TCPHandler對(duì)象  
            }  
        } catch (IOException e) {  
            // TODO Auto-generated catch block  
            e.printStackTrace();  
        }  
    }  
}  

// Handler線程  
package server;  
import java.io.IOException;  
import java.nio.ByteBuffer;  
import java.nio.channels.SelectionKey;  
import java.nio.channels.SocketChannel;  
import java.util.concurrent.LinkedBlockingQueue;  
import java.util.concurrent.ThreadPoolExecutor;  
import java.util.concurrent.TimeUnit;  
public class TCPHandler implements Runnable {  
    private final SelectionKey sk;  
    private final SocketChannel sc;  
    int state;   
    public TCPHandler(SelectionKey sk, SocketChannel sc) {  
        this.sk = sk;  
        this.sc = sc;  
        state = 0; // 初始狀態(tài)設(shè)定為READING  
    }  
    @Override  
    public void run() {  
        try {  
            if (state == 0)  
                read(); // 讀取網(wǎng)絡(luò)數(shù)據(jù)  
            else  
                send(); // 發(fā)送網(wǎng)絡(luò)數(shù)據(jù)  
        } catch (IOException e) {  
            System.out.println("[Warning!] A client has been closed.");  
            closeChannel();  
        }  
    }  
    private void closeChannel() {  
        try {  
            sk.cancel();  
            sc.close();  
        } catch (IOException e1) {  
            e1.printStackTrace();  
        }  
    }  
    private synchronized void read() throws IOException {  
        // non-blocking下不可用Readers，因?yàn)镽eaders不支援non-blocking  
        byte[] arr = new byte[1024];  
        ByteBuffer buf = ByteBuffer.wrap(arr);  
        int numBytes = sc.read(buf); // 讀取字符串  
        if(numBytes == -1)  
        {  
            System.out.println("[Warning!] A client has been closed.");  
            closeChannel();  
            return;  
        }  
        String str = new String(arr); // 將讀取到的byte內(nèi)容轉(zhuǎn)為字符串型態(tài)  
        if ((str != null) && !str.equals(" ")) {  
            process(str); // 邏輯處理  
            System.out.println(sc.socket().getRemoteSocketAddress().toString()  
                    + " > " + str);  
            state = 1; // 改變狀態(tài)  
            sk.interestOps(SelectionKey.OP_WRITE); // 通過key改變通道註冊(cè)的事件  
            sk.selector().wakeup(); // 使一個(gè)阻塞住的selector操作立即返回  
        }  
    }  
    private void send() throws IOException  {  
        // get message from message queue  
        String str = "Your message has sent to "  
                + sc.socket().getLocalSocketAddress().toString() + "\r\n";  
        ByteBuffer buf = ByteBuffer.wrap(str.getBytes()); // wrap自動(dòng)把buf的position設(shè)為0，所以不需要再flip()  
        while (buf.hasRemaining()) {  
            sc.write(buf); // 回傳給client回應(yīng)字符串，發(fā)送buf的position位置 到limit位置為止之間的內(nèi)容  
        }  
        state = 0; // 改變狀態(tài)  
        sk.interestOps(SelectionKey.OP_READ); // 通過key改變通道註冊(cè)的事件  
        sk.selector().wakeup(); // 使一個(gè)阻塞住的selector操作立即返回  
    }  
    void process(String str) {  
        // do process(decode, logically process, encode)..  
        // ..  
    }  
}  

package server;  
import java.io.IOException;  
public class Main {  
    public static void main(String[] args) {  
        // TODO Auto-generated method stub  
        try {  
            TCPReactor reactor = new TCPReactor(1333);  
            reactor.run();  
        } catch (IOException e) {  
            // TODO Auto-generated catch block  
            e.printStackTrace();  
        }  
    }  
}  

package main.pkg;  
import java.io.BufferedReader;  
import java.io.IOException;  
import java.io.InputStreamReader;  
import java.io.PrintWriter;  
import java.net.Socket;  
import java.net.UnknownHostException;  
public class Client {  
    /** 
     * @param args 
     */  
    public static void main(String[] args) {  
        // TODO Auto-generated method stub  
        String hostname=args[0];  
        int port = Integer.parseInt(args[1]);  
        //String hostname="127.0.0.1";  
        //int port=1333;  
        System.out.println("Connecting to "+ hostname +":"+port);  
        try {  
            Socket client = new Socket(hostname, port); // 連接至目的地  
            System.out.println("Connected to "+ hostname);  
            PrintWriter out = new PrintWriter(client.getOutputStream());  
            BufferedReader in = new BufferedReader(new InputStreamReader(client.getInputStream()));  
            BufferedReader stdIn = new BufferedReader(new InputStreamReader(System.in));  
            String input;  
            while((input=stdIn.readLine()) != null) { // 讀取輸入  
                out.println(input); // 發(fā)送輸入的字符串  
                out.flush(); // 強(qiáng)制將緩衝區(qū)內(nèi)的數(shù)據(jù)輸出  
                if(input.equals("exit"))  
                {  
                    break;  
                }  
                System.out.println("server: "+in.readLine());  
            }  
            client.close();  
            System.out.println("client stop.");  
        } catch (UnknownHostException e) {  
            // TODO Auto-generated catch block  
            System.err.println("Don't know about host: " + hostname);  
        } catch (IOException e) {  
            // TODO Auto-generated catch block  
            System.err.println("Couldn't get I/O for the socket connection");  
        }  
    }  
}