Java NIO原理及实例

NIO中的重要概念 通道、缓冲区、选择器

1 通道

类似于流，但是可以异步读写数据（流只能同步读写），通道是双向的，（流是单向的），通道的数据总是要先读到一个buffer 或者 从一个buffer写入，即通道与buffer进行数据交互。
通道类型：

• FileChannel：从文件中读写数据。
• DatagramChannel：能通过UDP读写网络中的数据。
• SocketChannel：能通过TCP读写网络中的数据。
• ServerSocketChannel：可以监听新进来的TCP连接，像Web服务器那样。对每一个新进来的连接都会创建一个SocketChannel。
  FileChannel比较特殊，它可以与通道进行数据交互， 不能切换到非阻塞模式，套接字通道可以切换到非阻塞模式。

2 缓冲区

本质上是一块可以存储数据的内存，被封装成了buffer对象而已。

缓冲区类型：

• ByteBuffer
• MappedByteBuffer
• CharBuffer
• DoubleBuffer
• FloatBuffer
• IntBuffer
• LongBuffer
• ShortBuffer　
  常用方法：
• allocate() - 分配一块缓冲区
• put() - 向缓冲区写数据
• get() - 向缓冲区读数据
• filp() - 将缓冲区从写模式切换到读模式
• clear() - 从读模式切换到写模式，不会清空数据，但后续写数据会覆盖原来的数据，即使有部分数据没有读，也会被遗忘；
• compact() - 从读数据切换到写模式，数据不会被清空，会将所有未读的数据copy到缓冲区头部，后续写数据不会覆盖，而是在这些数据之后写数据
• mark() - 对position做出标记，配合reset使用
• reset() - 将position置为标记值　
  　缓冲区的一些属性：
• capacity - 缓冲区大小，无论是读模式还是写模式，此属性值不会变；
• position - 写数据时，position表示当前写的位置，每写一个数据，会向下移动一个数据单元，初始为0；最大为capacity - 1，切换到读模式时，position会被置为0，表示当前读的位置
• limit - 写模式下，limit 相当于capacity 表示最多可以写多少数据，切换到读模式时，limit 等于原先的position，表示最多可以读多少数据。

3 选择器：

相当于一个观察者，用来监听通道感兴趣的事件，一个选择器可以绑定多个通道。

通道向选择器注册时，需要指定感兴趣的事件，选择器支持以下事件：

• SelectionKey.OP_CONNECT
• SelectionKey.OP_ACCEPT
• SelectionKey.OP_READ
• SelectionKey.OP_WRITE　　
  　如果你对不止一种事件感兴趣，那么可以用“位或”操作符将常量连接起来，如下：
  int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
  通道向选择器注册时，会返回一个 SelectionKey对象，具有如下属性
• interest集合
• ready集合
• Channel
• Selector
• 附加的对象（可选）　　
  　　用“位与”操作interest 集合和给定的SelectionKey常量，可以确定某个确定的事件是否在interest 集合

int interestSet = selectionKey.interestOps();  

boolean isInterestedInAccept  = interestSet & SelectionKey.OP_ACCEPT;
boolean isInterestedInConnect = interestSet & SelectionKey.OP_CONNECT;  
boolean isInterestedInRead    = interestSet & SelectionKey.OP_READ;  
boolean isInterestedInWrite   = interestSet & SelectionKey.OP_WRITE;  

ready 集合是通道已经准备就绪的操作的集合。在一次选择(Selection)之后，你会首先访问这个ready set。Selection将在下一小节进行解释。可以这样访问ready集合：
int readySet = selectionKey.readyOps();
也可以使用以下四个方法获取已就绪事件，返回值为boolean：

 selectionKey.isAcceptable();  
selectionKey.isConnectable();  
selectionKey.isReadable();  
selectionKey.isWritable();

可以将一个对象或者更多信息附着到SelectionKey上，即记录在附加对象上，方法如下：

selectionKey.attach(theObject);  
Object attachedObj = selectionKey.attachment();

可以通过选择器的select方法获取是否有就绪的通道；

• int select()
• int select(long timeout)
• int selectNow()
  返回值表示上次执行select之后，就绪通道的个数。　
  可以通过selectedKeySet获取已就绪的通道。返回值是SelectionKey 的集合，处理完相应的通道之后，需要removed 因为Selector不会自己removed.select阻塞后，可以用wakeup唤醒；执行wakeup时，如果没有阻塞的select 那么执行完wakeup后下一个执行select就会立即返回。调用close() 方法关闭selector。

package com.pt.nio;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
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;
import java.util.Set;

public class Reactor implements Runnable {
    public int id = 100001;
    public int bufferSize = 2048;
    @Override
    public void run() {
        // TODO Auto-generated method stub
        init();
    }

    public void init() {
        try {
            // 创建通道和选择器
            ServerSocketChannel socketChannel = ServerSocketChannel.open();
            Selector selector = Selector.open();
            InetSocketAddress inetSocketAddress = new InetSocketAddress(
                    InetAddress.getLocalHost(), 4700);
            socketChannel.socket().bind(inetSocketAddress);
            // 设置通道非阻塞 绑定选择器
            socketChannel.configureBlocking(false);
            socketChannel.register(selector, SelectionKey.OP_ACCEPT).attach(
                    id++);
            System.out.println("Server started .... port:4700");
            listener(selector);

        } catch (Exception e) {
            // TODO: handle exception
        }
    }

    public void listener(Selector in_selector) {
        try {
            while (true) {
                Thread.sleep(1*1000);
                in_selector.select(); // 阻塞 直到有就绪事件为止
                Set<SelectionKey> readySelectionKey = in_selector
                        .selectedKeys();
                Iterator<SelectionKey> it = readySelectionKey.iterator();
                while (it.hasNext()) {
                    SelectionKey selectionKey = it.next();
                    // 判断是哪个事件
                    if (selectionKey.isAcceptable()) {// 客户请求连接
                        System.out.println(selectionKey.attachment()
                                + " - 接受请求事件");
                        // 获取通道 接受连接,
                        // 设置非阻塞模式（必须），同时需要注册 读写数据的事件，这样有消息触发时才能捕获
                        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey
                                .channel();
                        serverSocketChannel
                                .accept()
                                .configureBlocking(false)
                                .register(
                                        in_selector,
                                        SelectionKey.OP_READ
                                                | SelectionKey.OP_WRITE).attach(id++);
                        System.out
                                .println(selectionKey.attachment() + " - 已连接");

                        // 下面这种写法是有问题的 不应该在serverSocketChannel上面注册
                        /*
                         * serverSocketChannel.configureBlocking(false);
                         * serverSocketChannel.register(in_selector,
                         * SelectionKey.OP_READ);
                         * serverSocketChannel.register(in_selector,
                         * SelectionKey.OP_WRITE);
                         */
                    }
                    if (selectionKey.isReadable()) {// 读数据
                        System.out.println(selectionKey.attachment()
                                + " - 读数据事件");
                        SocketChannel clientChannel=(SocketChannel)selectionKey.channel();
                        ByteBuffer receiveBuf = ByteBuffer.allocate(bufferSize);
                        clientChannel.read(receiveBuf);
                        System.out.println(selectionKey.attachment()
                                + " - 读取数据：" + getString(receiveBuf));
                    }
                    if (selectionKey.isWritable()) {// 写数据
                        System.out.println(selectionKey.attachment()
                                + " - 写数据事件");
                        SocketChannel clientChannel = (SocketChannel) selectionKey.channel();
                        ByteBuffer sendBuf = ByteBuffer.allocate(bufferSize);
                        String sendText = "hello\n";
                        sendBuf.put(sendText.getBytes());
                        sendBuf.flip();        //写完数据后调用此方法
                        clientChannel.write(sendBuf);
                    }
                    if (selectionKey.isConnectable()) {
                        System.out.println(selectionKey.attachment()
                                + " - 连接事件");
                    }
                    // 必须removed 否则会继续存在，下一次循环还会进来,
                    // 注意removed 的位置，针对一个.next() remove一次
                    it.remove(); 
                }
            }
        } catch (Exception e) {
            // TODO: handle exception
            System.out.println("Error - " + e.getMessage());
            e.printStackTrace();
        }

    }
    /**
     * ByteBuffer 转换 String
     * @param buffer
     * @return
     */
    public static String getString(ByteBuffer buffer)
    {
        String string = "";
        try
        {
            for(int i = 0; i<buffer.position();i++){
                string += (char)buffer.get(i);
            }
            return string;
        }
        catch (Exception ex)
        {
            ex.printStackTrace();
            return "";
        }
    }
}

NIO服务器端