NIO系列5：SocketChannel的理解

本文参考至：http://ifeve.com/socket-channel/
在NIO系列4中，采用了SocketChannel作为案例讲解Selector，当时我确实看不太懂。现在写一下SocketChannel的理解：

Java NIO中的SocketChannel是一个连接到TCP网络套接字的通道。可以通过以下2种方式创建SocketChannel：
1、打开一个SocketChannel并连接到互联网上的某台服务器。
2、一个新连接到达ServerSocketChannel时，会创建一个SocketChannel。

这里简要的介绍一下Channel的读写数据的方法，其实对于所有的Channel读写数据的方法都几乎一样，都是从Buffer中读或者写到Buffer中，下面举FileChannel和SocketChannel两个例子：

Reading from a FileChannel：

    ByteBuffer buf = ByteBuffer.allocate(48);
    int bytesRead = inChannel.read(buf);

Reading from a SocketChannel：

    ByteBuffer buf = ByteBuffer.allocate(48);
    int bytesRead = socketChannel.read(buf);

Writing to a SocketChannel：

    String newData = "New String to write to file..." + System.currentTimeMillis();
    ByteBuffer buf = ByteBuffer.allocate(48);
    buf.clear();
    buf.put(newData.getBytes());
    buf.flip();
    while(buf.hasRemaining()) {
        socketchannel.write(buf);
    }

Writing Data to a FileChannel:

String newData = "New String to write to file..." + System.currentTimeMillis();
    ByteBuffer buf = ByteBuffer.allocate(48);
    buf.clear();
    buf.put(newData.getBytes());
    buf.flip();
    while(buf.hasRemaining()) {
        fileChannel.write(buf);
    }

以下代码模拟了服务器和客户端：

服务器：

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.nio.charset.Charset;
import java.util.Iterator;

public class TCPServer {

    private static final int bufferSize = 1024;
    private static final long timeOut = 3000;// 超时时间
    private static final int listenPort = 1993;// 本地监听端口

    public static void main(String[] args) throws Exception {
        Selector selector = Selector.open();
        ServerSocketChannel listenerChannel = ServerSocketChannel.open();// 创建监听通道，专门用来监听指定的本地端口
        listenerChannel.socket().bind(new InetSocketAddress(listenPort));// 将listenerChannel的socket绑定为本地服务器（IP+prot）绑定
        listenerChannel.configureBlocking(false);
        // 将选择器绑定到监听信道,只有非阻塞信道才可以注册选择器.并在注册过程中指出该信道可以进行Accept操作
        listenerChannel.register(selector, SelectionKey.OP_ACCEPT);
        TCPProtocolImpl protocol = new TCPProtocolImpl(bufferSize);

        while (true) {
            if (selector.select(timeOut) == 0) {// 监听注册的通道，当其中有注册的IO时该函数返回（3000ms没有反应返回0），操作可以进行，并添加对应的SelectorKey
                System.out.println("It haven't I/O now, please wait!");
                continue;
            }

            Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
            while (keyIter.hasNext()) {
                try {
                    SelectionKey key = keyIter.next();
                    if (key.isAcceptable()) {
                        protocol.handleAccept(key);
                    }
                    if (key.isReadable()) {
                        protocol.handleRead(key);
                    }
                } catch (IOException e) {
                    keyIter.remove();
                    continue;
                }
                keyIter.remove();
            }
        }
    }
}

class TCPProtocolImpl{
    private int bufferSize;

    public TCPProtocolImpl() {
        super();
    }

    public TCPProtocolImpl(int bufferSize) {
        super();
        this.bufferSize = bufferSize;
    }

    public void handleAccept(SelectionKey key) throws IOException {
        // 返回创建此键的通道，接受客户端建立连接的请求，并返回SocketChannel对象
        SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
        clientChannel.configureBlocking(false);
        // 将clientChannel注册到服务端的selector中
        clientChannel.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(bufferSize));
    }

    public void handleRead(SelectionKey key) throws IOException {
        // 获取客户端通信的通道
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        buffer.clear();
        // 从客户端通道读取信息到buffer缓冲区中(并返回读到信息的字节数)
        long bytesRead = clientChannel.read(buffer);
        if (bytesRead == -1) {
            clientChannel.close();
        } else {
            buffer.flip();
            // 将字节转化为为UTF-8的字符串
            String receivedString = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
            System.out.println("接收到来自：" + clientChannel.socket().getRemoteSocketAddress() + "发来的信息：" + receivedString);
            String msgSendToClient = "已接收到你的信息：" + receivedString + "正在处理中";
            buffer = ByteBuffer.wrap(msgSendToClient.getBytes("UTF-8"));
            clientChannel.write(buffer);
            // 设置为下一次读取或是写入做准备
            key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        }
    }
}

客户端：

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.nio.charset.Charset;
import java.util.Scanner;

public class TCPClient {

    // 通道选择器，用于管理客户端的通道
    private Selector selector;

    // 与服务器通信的通道
    SocketChannel socketChannel;

    // 要连接的服务器的IP
    private String hostIp;

    // 要连接的远程服务器在监听的端口
    private int hostListenningPort;
    
    static TCPClient client;
    
    static boolean mFlag = true;

    public TCPClient(String hostIp, int hostPort) throws IOException {
        this.hostIp = hostIp;
        this.hostListenningPort = hostPort;
        init();
    }

    private void init() throws IOException {
        // 打开监听通道
        socketChannel = SocketChannel.open(new InetSocketAddress(hostIp, hostListenningPort));
        socketChannel.configureBlocking(false);
        
        // 创建选择器，并把通道注册到选择器中
        selector = Selector.open();
        socketChannel.register(selector, SelectionKey.OP_READ);
        
        new TCPClientReadThread(selector);
    }
    
    /**
     * 发送字符串到服务器
     * @param message
     * @throws IOException
     */
    public void sendMsg(String message) throws IOException{
        ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes("UTF-8"));
        socketChannel.write(writeBuffer);
    }
    
    public static void main(String[] args) throws IOException {
        client = new TCPClient("127.0.0.1", 1993);
        new Thread(){
            @Override
            public void run(){
                try{
                    client.sendMsg("test----~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
                    while(mFlag){
                        Scanner scan = new Scanner(System.in);
                        String string = scan.next();
                        client.sendMsg(string);
                    }
                }catch (Exception e) {
                    mFlag = false;
                }finally{
                    mFlag = false;
                }
                super.run();
            }
        }.start();
    }
} 

class TCPClientReadThread implements Runnable {
    private Selector selector;

    public TCPClientReadThread(Selector selector) {
        super();
        this.selector = selector;
        new Thread(this).start();
    }

    @Override
    public void run() {
        try {
            while (selector.select() > 0) {// select()方法只能使用一次，用了之后就会自动删除,每个连接到服务器的选择器都是独立的
                // 遍历每个有IO操作Channel对应的SelectionKey
                for (SelectionKey sk : selector.selectedKeys()) {
                    if (sk.isReadable()) {
                        // 使用NIO读取Channel中的数据
                        SocketChannel sc = (SocketChannel) sk.channel();
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        sc.read(buffer);
                        buffer.flip();
                        String receivedString = Charset.forName("UTF-8").newDecoder().decode(buffer).toString();
                        System.out.println("接收到来自服务器：" + sc.socket().getRemoteSocketAddress() + "的信息：" + receivedString);
                        sk.interestOps(SelectionKey.OP_READ);
                    }
                    selector.selectedKeys().remove(sk);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}