NIO Socket编程实例

1.阻塞模式实例　　

　　NIOUtil类，用来通过SOcket获取BufferedReader和PrintWriter。

package IO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.Socket;

public class NIOUtil {
    public static PrintWriter getPrintWriter(Socket socket) throws IOException {
        OutputStream outputStream = socket.getOutputStream();
        return new PrintWriter(outputStream, true);
    }

    public static BufferedReader getBufferedReader(Socket socket)
            throws IOException {
        InputStream inputStream = socket.getInputStream();
        return new BufferedReader(new InputStreamReader(inputStream));
    }
}

　　使用ServerSocketChannel创建阻塞服务器端程序：

package IO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BlockingNIOServer {
    private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private ExecutorService executorService = null;
    private static int DEFAULT_POOI_SIZE = 4;

    public BlockingNIOServer() throws IOException {
        super();
        this.executorService = Executors.newFixedThreadPool(DEFAULT_POOI_SIZE
                * Runtime.getRuntime().availableProcessors());
        this.serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().setReuseAddress(true);
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        this.executorService = executorService;
    }

    public void service() {
        while (true) {
            SocketChannel channel = null;
            try {
                channel = serverSocketChannel.accept();
                executorService.execute(new Handler(channel));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        new BlockingNIOServer().service();
    }

    private class Handler implements Runnable {
        private SocketChannel channel;
        public Handler(SocketChannel channel) {
            super();
            this.channel = channel;
        }

        @Override
        public void run() {
            handler(channel);
        }

        public void handler(SocketChannel channel) {
            Socket socket = null;
            try {
                socket = channel.socket();
                System.out.println("接收到来自：" + socket.getInetAddress() + " 端口："
                        + socket.getPort() + "的请求");
                BufferedReader bufferedReader = NIOUtil
                        .getBufferedReader(socket);
                PrintWriter printWriter = NIOUtil.getPrintWriter(socket);
                String msg = null;

                while ((msg = bufferedReader.readLine()) != null) {
                    System.out.println(msg);
                    printWriter.println(Echo(msg));
                    if ("bye".equalsIgnoreCase(msg))
                        break;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (channel != null) {
                    try {
                        channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }

        }

        private String Echo(String msg) {
            return "ECHO:" + msg;
        }
    }
}

　　使用SocketChannel创建阻塞Socket客户端：

package IO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BlockingNIOServer {
    private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private ExecutorService executorService = null;
    private static int DEFAULT_POOI_SIZE = 4;

    public BlockingNIOServer() throws IOException {
        super();
        this.executorService = Executors.newFixedThreadPool(DEFAULT_POOI_SIZE
                * Runtime.getRuntime().availableProcessors());
        this.serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().setReuseAddress(true);
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        this.executorService = executorService;
    }

    public void service() {
        while (true) {
            SocketChannel channel = null;
            try {
                channel = serverSocketChannel.accept();
                executorService.execute(new Handler(channel));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        new BlockingNIOServer().service();
    }

    private class Handler implements Runnable {
        private SocketChannel channel;
        public Handler(SocketChannel channel) {
            super();
            this.channel = channel;
        }

        @Override
        public void run() {
            handler(channel);
        }

        public void handler(SocketChannel channel) {
            Socket socket = null;
            try {
                socket = channel.socket();
                System.out.println("接收到来自：" + socket.getInetAddress() + " 端口："
                        + socket.getPort() + "的请求");
                BufferedReader bufferedReader = NIOUtil
                        .getBufferedReader(socket);
                PrintWriter printWriter = NIOUtil.getPrintWriter(socket);
                String msg = null;

                while ((msg = bufferedReader.readLine()) != null) {
                    System.out.println(msg);
                    printWriter.println(Echo(msg));
                    if ("bye".equalsIgnoreCase(msg))
                        break;
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (channel != null) {
                    try {
                        channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            }

        }

        private String Echo(String msg) {
            return "ECHO:" + msg;
        }
    }
}

2.非阻塞模式实例

　　Charset类，主要用于decode（）和encode()

package IO;

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;

public class CharSetUtil {
    public static String decode(ByteBuffer buffer, String charsetName) {
        Charset charset = Charset.forName(charsetName);
        CharBuffer msg = charset.decode(buffer);
        return msg.toString();
    }

    public static ByteBuffer encode(String msg, String charsetName) {
        Charset charset = Charset.forName(charsetName);
        ByteBuffer byteBuffer = charset.encode(msg);
        return byteBuffer;
    }
}

　　非阻塞的服务器端

package IO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.net.Socket;
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.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NoBlockingServer {
    private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private Selector selector = null;

    public NoBlockingServer() throws IOException {
        super();
        selector = Selector.open();
        this.serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().setReuseAddress(true);
        serverSocketChannel.configureBlocking(false);//设置为非阻塞
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        System.out.println("服务器启动成功");
    }

    public void service() throws IOException {
        //给serverSocketChannel注册OP_ACCEPT事件
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        //注意selector.select()将会阻塞
        while (selector.select() > 0) {
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = null;
                try {
                    selectionKey = (SelectionKey) iterator.next();
                    iterator.remove();

                    if (selectionKey.isAcceptable()) {
                        dealWithAcceptable(selectionKey);
                    }
                    if (selectionKey.isReadable()) {
                        dealWithReadable(selectionKey);
                    }
                    if (selectionKey.isWritable()) {
                        dealWithWritable(selectionKey);
                    }
                } catch (Exception e) {
                    if (selectionKey != null) {
                        selectionKey.cancel();
                        selectionKey.channel().close();
                    }
                }
            }
        }
    }

    private void dealWithAcceptable(SelectionKey selectionKey)
            throws IOException {
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey
                .channel();
        SocketChannel socketChannel = serverSocketChannel.accept();
        System.out.println("接收到来自：" + socketChannel.socket().getInetAddress()
                + " 端口" + socketChannel.socket().getPort() + "的请求");
        socketChannel.configureBlocking(false);
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ
                | SelectionKey.OP_WRITE, buffer);
    }

    private void dealWithReadable(SelectionKey selectionKey) throws IOException{
        ByteBuffer buffer=(ByteBuffer) selectionKey.attachment();
        SocketChannel channel=(SocketChannel) selectionKey.channel();
        ByteBuffer readBuffer=ByteBuffer.allocate(32);
        channel.read(readBuffer);
        readBuffer.flip();

        buffer.limit(buffer.capacity());
        buffer.put(readBuffer);
    }

    private void dealWithWritable(SelectionKey selectionKey) throws IOException{
        ByteBuffer buffer=(ByteBuffer) selectionKey.attachment();
        SocketChannel channel=(SocketChannel) selectionKey.channel();
        buffer.flip();

        String msg=CharSetUtil.decode(buffer, "UTF-8");

        if(msg.indexOf("\r\n")==-1){
            return;
        }

        String outPutData=msg.substring(0, msg.indexOf("\n")+1);
        System.out.println("接收来自客户端的数据："+outPutData);

        ByteBuffer outbyteBuffer=CharSetUtil.encode("echo:"+outPutData, "UTF-8");
        while (outbyteBuffer.hasRemaining()) {
            channel.write(outbyteBuffer);
        }

        ByteBuffer tmp=CharSetUtil.encode(outPutData, "UTF-8");
        buffer.position(tmp.limit());
        buffer.compact();
        if("bye\r\n".equalsIgnoreCase(outPutData)){
            selectionKey.cancel();
            channel.close();
            System.out.println("关闭与客户端的连接");
        }
    }

    public static void main(String[] args) throws Exception {
        new NoBlockingServer().service();
    }
}

　　非阻塞的客户端

package IO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
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.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class NoBlockingClient {
    private SocketChannel channel = null;
    private ByteBuffer send = ByteBuffer.allocate(1024);
    private ByteBuffer rece = ByteBuffer.allocate(1024);
    private Selector selector;

    public NoBlockingClient() throws IOException {
        super();
        channel = SocketChannel.open();
        channel.socket().connect(new InetSocketAddress("localhost", 8000));
        channel.configureBlocking(false);

        System.out.println("与服务器建立连接成功");
        selector = Selector.open();
    }

    public void talk() throws IOException {
        channel.register(selector, SelectionKey.OP_READ|SelectionKey.OP_WRITE);
        while (selector.select() > 0) {
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            Iterator iterator = selectionKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = null;
                try {
                    selectionKey = (SelectionKey) iterator.next();
                    iterator.remove();

                    if (selectionKey.isReadable()) {
                        dealWithReadable(selectionKey);
                    }
                    if (selectionKey.isWritable()) {
                        dealWithWritable(selectionKey);
                    }
                } catch (Exception e) {
                    if (selectionKey != null) {
                        selectionKey.cancel();
                        try {
                            selectionKey.channel().close();
                        } catch (IOException e1) {
                            e1.printStackTrace();
                        }
                    }
                }
            }
        }
    }

    private void receFromUser() throws IOException{
        BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(System.in));
        String msg=null;
        while ((msg=bufferedReader.readLine())!=null) {
            synchronized (send) {
                send.put(CharSetUtil.encode(msg+"\r\n", "UTF-8"));
            }
            if ("bye".equalsIgnoreCase(msg)) {
                break;
            }
        }
    }

    private void dealWithWritable(SelectionKey selectionKey) throws IOException{
        SocketChannel channel=(SocketChannel) selectionKey.channel();
        synchronized (send) {
            send.flip();
            channel.write(send);
            send.compact();
        }
    }

    private void dealWithReadable(SelectionKey selectionKey) throws IOException{
        SocketChannel channel=(SocketChannel) selectionKey.channel();
        channel.read(rece);
        rece.flip();
        String msg=CharSetUtil.decode(rece, "UTF-8");

        if(msg.indexOf("\r\n")==-1){
            return;
        }

        String outPutData=msg.substring(0, msg.indexOf("\n")+1);
        System.out.println(outPutData);

        if("echo:bye\r\n".equalsIgnoreCase(outPutData)){
            selectionKey.cancel();
            channel.close();
            selector.close();
            System.out.println("关闭与客户端的连接");
        }

        ByteBuffer tmp=CharSetUtil.encode(outPutData, "UTF-8");
        rece.position(tmp.limit());
        rece.compact();
    }

    public static void main(String[] args) throws IOException {
           System.out.println(System.getProperty("file.encoding"));
        final NoBlockingClient noBlockingClient=new NoBlockingClient();
        Thread thread=new Thread(){
            public void run() {
                try {
                    noBlockingClient.receFromUser();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            };
        };

        thread.start();
        noBlockingClient.talk();
    }
}

3.阻塞和非阻塞编程实例

　　服务器端使用阻塞和非阻塞模式，f负责接收客户端连接的线程按照阻塞模式工作，负责接收和发送数据的线程按照非阻塞模式工作。

package IO;

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 NOBlockingAndBolckingServer {
    private int port = 8000;
    private ServerSocketChannel serverSocketChannel = null;
    private Selector selector = null;
    private Object gate = new Object();

    public NOBlockingAndBolckingServer() throws IOException {
        super();
        selector = Selector.open();
        this.serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        System.out.println("服务器启动成功");
    }

    public void accept() {
        for (;;) {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                System.out.println("接收到来自："
                        + socketChannel.socket().getInetAddress() + " 端口"
                        + socketChannel.socket().getPort() + "的请求");
                socketChannel.configureBlocking(false);

                synchronized (gate) {
                    selector.wakeup();
                    socketChannel.register(selector, SelectionKey.OP_READ
                            | SelectionKey.OP_WRITE);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public void service() throws IOException {
        // 给serverSocketChannel注册OP_ACCEPT事件
        for (;;) {
            synchronized (gate) {
                int n = selector.select();
                if (n == 0)
                    continue;
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey selectionKey = null;
                    try {
                        selectionKey = (SelectionKey) iterator.next();
                        iterator.remove();

                        if (selectionKey.isReadable()) {
                            dealWithReadable(selectionKey);
                        }
                        if (selectionKey.isWritable()) {
                            dealWithWritable(selectionKey);
                        }
                    } catch (Exception e) {
                        if (selectionKey != null) {
                            selectionKey.cancel();
                            selectionKey.channel().close();
                        }
                    }
                }
            }
        }
    }

    private void dealWithReadable(SelectionKey selectionKey) throws IOException {
        ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
        SocketChannel channel = (SocketChannel) selectionKey.channel();
        ByteBuffer readBuffer = ByteBuffer.allocate(32);
        channel.read(readBuffer);
        readBuffer.flip();

        buffer.limit(buffer.capacity());
        buffer.put(readBuffer);
    }

    private void dealWithWritable(SelectionKey selectionKey) throws IOException {
        ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
        SocketChannel channel = (SocketChannel) selectionKey.channel();
        buffer.flip();

        String msg = CharSetUtil.decode(buffer, "UTF-8");

        if (msg.indexOf("\r\n") == -1) {
            return;
        }

        String outPutData = msg.substring(0, msg.indexOf("\n") + 1);
        System.out.println("接收来自客户端的数据：" + outPutData);

        ByteBuffer outbyteBuffer = CharSetUtil.encode("echo:" + outPutData,
                "UTF-8");
        while (outbyteBuffer.hasRemaining()) {
            channel.write(outbyteBuffer);
        }

        ByteBuffer tmp = CharSetUtil.encode(outPutData, "UTF-8");
        buffer.position(tmp.limit());
        buffer.compact();
        if ("bye\r\n".equalsIgnoreCase(outPutData)) {
            selectionKey.cancel();
            channel.close();
            System.out.println("关闭与客户端的连接");
        }
    }

    public static void main(String[] args) throws Exception {
        final NOBlockingAndBolckingServer server=new NOBlockingAndBolckingServer();
        Thread thread=new Thread(){
            public void run() {
                server.accept();
            };
        };
        thread.start();
        server.service();
    }
}

　　客户端和服务器端创建多个连接。

package IO;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.LinkedList;

class Target { // 表示一项任务
    InetSocketAddress address;
    SocketChannel channel;
    Exception failure;
    long connectStart; // 开始连接时的时间
    long connectFinish = 0; // 连接成功时的时间
    boolean shown = false; // 该任务是否已经打印

    Target(String host) {
        try {
            address = new InetSocketAddress(InetAddress.getByName(host), 80);
        } catch (IOException x) {
            failure = x;
        }
    }

    void show() { // 打印任务执行的结果
        String result;
        if (connectFinish != 0)
            result = Long.toString(connectFinish - connectStart) + "ms";
        else if (failure != null)
            result = failure.toString();
        else
            result = "Timed out";
        System.out.println(address + " : " + result);
        shown = true;
    }
}

public class PingClient {
    private Selector selector;
    // 存放用户新提交的任务
    private LinkedList targets = new LinkedList();
    // 存放已经完成的需要打印的任务
    private LinkedList finishedTargets = new LinkedList();

    public PingClient() throws IOException {
        selector = Selector.open();
        Connector connector = new Connector();
        Printer printer = new Printer();
        connector.start();
        printer.start();
        receiveTarget();
    }

    public static void main(String args[]) throws IOException {
        new PingClient();
    }

    public void addTarget(Target target) {
        // 向targets队列中加入一个任务
        SocketChannel socketChannel = null;
        try {
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
            socketChannel.connect(target.address);

            target.channel = socketChannel;
            target.connectStart = System.currentTimeMillis();

            synchronized (targets) {
                targets.add(target);
            }
            selector.wakeup();
        } catch (Exception x) {
            if (socketChannel != null) {
                try {
                    socketChannel.close();
                } catch (IOException xx) {
                }
            }
            target.failure = x;
            addFinishedTarget(target);
        }
    }

    public void addFinishedTarget(Target target) {
        // 向finishedTargets队列中加入一个任务
        synchronized (finishedTargets) {
            finishedTargets.notify();
            finishedTargets.add(target);
        }
    }

    public void printFinishedTargets() {
        // 打印finisedTargets队列中的任务
        try {
            for (;;) {
                Target target = null;
                synchronized (finishedTargets) {
                    while (finishedTargets.size() == 0)
                        finishedTargets.wait();
                    target = (Target) finishedTargets.removeFirst();
                }
                target.show();
            }
        } catch (InterruptedException x) {
            return;
        }
    }

    public void registerTargets() {
        // 取出targets队列中的任务，向Selector注册连接就绪事件
        synchronized (targets) {
            while (targets.size() > 0) {
                Target target = (Target) targets.removeFirst();

                try {
                    target.channel.register(selector, SelectionKey.OP_CONNECT,
                            target);
                } catch (IOException x) {
                    try {
                        target.channel.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    target.failure = x;
                    addFinishedTarget(target);
                }
            }
        }
    }

    public void processSelectedKeys() throws IOException {
        // 处理连接就绪事件
        for (Iterator it = selector.selectedKeys().iterator(); it.hasNext();) {
            SelectionKey selectionKey = (SelectionKey) it.next();
            it.remove();

            Target target = (Target) selectionKey.attachment();
            SocketChannel socketChannel = (SocketChannel) selectionKey
                    .channel();

            try {
                if (socketChannel.finishConnect()) {
                    selectionKey.cancel();
                    target.connectFinish = System.currentTimeMillis();
                    socketChannel.close();
                    addFinishedTarget(target);
                }
            } catch (IOException x) {
                socketChannel.close();
                target.failure = x;
                addFinishedTarget(target);
            }
        }
    }

    //接收用户输入的地址，向targets队列中加入任务
    public void receiveTarget() {
        try {
            BufferedReader localReader = new BufferedReader(
                    new InputStreamReader(System.in));
            String msg = null;
            while ((msg = localReader.readLine()) != null) {
                if (!msg.equals("bye")) {
                    Target target = new Target(msg);
                    addTarget(target);
                } else {
                    shutdown = true;
                    selector.wakeup();
                    break;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    boolean shutdown = false;

    public class Printer extends Thread {
        public Printer() {
            setDaemon(true);
        }

        public void run() {
            printFinishedTargets();
        }
    }

    public class Connector extends Thread {
        public void run() {
            while (!shutdown) {
                try {
                    registerTargets();
                    if (selector.select() > 0) {
                        processSelectedKeys();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}