Java NIO

# Java NIO #

Java NIO属于非阻塞IO，这是与传统IO最本质的区别。传统IO包括socket和文件IO都是阻塞式的，即一个动作的执行，必须等待先前的动作完成；非阻塞的IO在线程执行一个动作，不用等待动作执行完，可以去做别的事情，这是因为NIO是基于Channel的，而不是基于流的。每个线程可以同时监听多个注册到Selector上的Channel， NIO 是一种同步非阻塞的 IO 模型。同步是指线程不断轮询 IO 事件是否就绪，非阻塞是指线程在等待 IO 的时候，可以同时做其他任务。同步的核心就是 Selector，Selector 代替了线程本身轮询 IO 事件，避免了阻塞同时减少了不必要的线程消耗；非阻塞的核心就是通道和缓冲区，当 IO 事件就绪时，可以通过写到缓冲区，保证 IO 的成功，而无需线程阻塞式地等待。

## Buffer 缓冲区 ##

Java NIO中所有的缓冲区都继承于 Buffer这个抽象类。Buffer类似于一块可以被读写的区域，因此有读模式和写模式两种模式，在Buffer类中通过三个变量控制缓冲区的读和写position（下一个读或写的数组下标），limit（在读模式和写模式下有不同的含义），capacity（数组的容量）：

- 写模式：往数组中写入数据时候，将limit设置为capacity，表示最多可以写入capacity个元素，position就表示下一个写入的下标。通过Buffer的flip()方法可以从写模式切换到读模式。

public final Buffer flip() {

limit = position;

position = 0;

mark = -1;

return this;

}

- 读模式：从数组中读取数据的时候，将limit设置为position，表示最多只有这么多个元素可以读，position设置为0，表示从头开始读数据。通过调用Buffer的clear()方法或者compact()方法可以从读模式切换到写模式。

public final Buffer clear() {

position = 0;

limit = capacity;

mark = -1;

return this;

}

clear()方法是不保留数据的（即使有些数据还没有读完），因为直接将position设置为0，表示从头开始写入数据，limit重新设置capacity。

public ByteBuffer compact() {

System.arraycopy(hb, ix(position()), hb, ix(0), remaining());

position(remaining());

limit(capacity());

discardMark();

return this;

}

compact()方法不是Buffer类中的方式，是在Buffer的各个子类（ByteBuffer、CharBuffer等）中定义的抽象方法，并且在（ByteBuffer、CharBuffer等）的子类中提供了实现，compact()方法会保留还没读的数据，先将没读的数据拷贝到数组的最前面，然后设置position为下一个写入的下标，limit重写设置为capacity。

## ByteBuffer 字节缓冲区 ##

ByteBuffer继承于Buffer，ByteBuffer仍然是个抽象类，我们只能通过它的allocate(int capacity)方法获取一个非直接字节缓冲区（缓冲区不是直接在内存中的），也可以通过warp(byte[] b)方法创建一个非直接字节缓冲区。如果想要创建一个直接字节缓冲区，可以使用allocateDirect(int capacity)方法，直接缓冲区不是在堆上分配的，因此不受GC的管理，其创建和释放过程比较耗时，但是直接缓冲区上数据的读写比较快。

//在堆上创建一个字节缓冲区

public static ByteBuffer allocate(int capacity) {

if (capacity < 0)

throw new IllegalArgumentException();

return new HeapByteBuffer(capacity, capacity);

}

//创建一个直接字节缓冲区

public static ByteBuffer allocateDirect(int capacity) {

return new DirectByteBuffer(capacity);

}

**HeapByteBuffer：**HeapByteBuffer是 ByteBuffer的默认实现类，在堆上创建一个字节缓冲区.HeapByteBuffer主要有以下几个方面的功能：

- 读数据:

//获取position处的元素，并将position++

public byte get() {

return hb[ix(nextGetIndex())];

}

//获取指定下标处的元素，此时position是没有变的。

public byte get(int i) {

return hb[ix(checkIndex(i))];

}

//获取从position开始的length个元素，并拷贝到指定数组中，position会更新为position + length

public ByteBuffer get(byte[] dst, int offset, int length) {

checkBounds(offset, length, dst.length);

if (length > remaining())

throw new BufferUnderflowException();

System.arraycopy(hb, ix(position()), dst, offset, length);

position(position() + length);

return this;

}

- 写数据：

//将指定元素插入到position处，并将position++（如果满了将会抛出异常）

public ByteBuffer put(byte x) {

hb[ix(nextPutIndex())] = x;

return this;

}

//将指定元素插入到指定位置，position位置不会改变

public ByteBuffer put(int i, byte x) {

hb[ix(checkIndex(i))] = x;

return this;

}

//从position处开始写入指定数组中的元素，position会被更新为（position + length）

public ByteBuffer put(byte[] src, int offset, int length) {

checkBounds(offset, length, src.length);

if (length > remaining())

throw new BufferOverflowException();

System.arraycopy(src, offset, hb, ix(position()), length);

position(position() + length);

return this;

}

- 读写char, int ,short, double等其他基本数据类型

//获取一个字符，Bits类根据大端还是小端用不同的方式组合两个字节

public char getChar() {

return Bits.getChar(this, ix(nextGetIndex(2)), bigEndian);

}

//大端存储模式，高位字节保存在低字节部分，因此组合的时候低字节在前面，高字节在后面

static char getCharB(ByteBuffer bb, int bi) {

return makeChar(bb._get(bi ),

bb._get(bi + 1));

}

//小端存储模式，高位字节保存在高字节部分，因此组合的时候高字节在前面，低字节在后面

static char getCharL(ByteBuffer bb, int bi) {

return makeChar(bb._get(bi + 1),

bb._get(bi ));

}

//写入一个char字符的时候也是一样

public ByteBuffer putChar(char x) {

Bits.putChar(this, ix(nextPutIndex(2)), x, bigEndian);

return this;

}

- 用ByteBuffer包装成其他基本数据类型的缓冲区（CharBuffer、IntBuffer等）

//ByteBuffer的asCharBuffer()方法

public CharBuffer asCharBuffer() {

int size = this.remaining() >> 1;

int off = offset + position();

return (bigEndian

? (CharBuffer)(new ByteBufferAsCharBufferB(this,

-1,

size,

off))

: (CharBuffer)(new ByteBufferAsCharBufferL(this,

-1,

size,

off)));

}

asCharBuffer()方法也会根据机器是大端模式还是小端模式，创建不同的对象。如果是大端模式，将会创建ByteBufferAsCharBufferB类对象，由于ByteBufferAsCharBufferB对象是由ByteBuffer对象包装而来的，虽然ByteBufferAsCharBufferB的父类是CharBuffer，但是ByteBufferAsCharBufferB类中操作的并不是字符数组而是字节数组，所以ByteBufferAsCharBufferB类对象在读写的时候仍然借助Bits类完成，即先通过字节组成字符再读写。根据ByteBuffer得到其他类型的缓冲区也是一样的实现原理。

- 创建只读型缓冲区

public ByteBuffer asReadOnlyBuffer() {

return new HeapByteBufferR(hb,

this.markValue(),

this.position(),

this.limit(),

this.capacity(),

offset);

}

asReadOnlyBuffer()会创建HeapByteBufferR类对象，每种缓冲区都可以创建对应的只读型缓冲区，只读型缓冲区就直接继承创建它的这个类，如HeapByteBufferR继承于HeapByteBuffer，在HeapByteBufferR类中重写所有的put方法，所有的put方法都直接抛出异常，而get方法仍然使用父类的get方法。

**MappedByteBuffer**继承于ByteBuffer，是直接字节数组的抽象父类，在MappedByteBuffer类中只定义了三个和物理磁盘相关的方法：

- isLoaded()：用于判断这个字节数组是否存在物理磁盘上

- load()：将直接字节数组中的数据写到物理磁盘上

- force()：强制将直接字节数组中的数据写到物理磁盘上

**DirectByteBuffer**继承于MappedByteBuffer，用于创建直接字节数组的类。由于直接字节数组不是在堆上分配内存，因此不受GC控制，创建和释放过程比较繁琐，通过通过Unsafe类的allocateMemory()方法分配内存空间，而且DirectByteBuffer的所有get()和put()方法都是通过Unsafe类完成

//读取position处的元素

public byte get() {

return ((unsafe.getByte(ix(nextGetIndex()))));

}

//在position处写入元素

public ByteBuffer put(byte x) {

unsafe.putByte(ix(nextPutIndex()), ((x)));

return this;

}

## Channel ##

关于同步、异步、阻塞和非阻塞的区别：同步和异步说的是消息的通知机制，阻塞非阻塞说的是线程的状态。

- 同步阻塞IO：client在调用read()方法时，stream里没有数据可读，线程停止向下执行，直至stream有数据。

- 同步非阻塞IO：client在调用read()方法时，stream里没有数据可读，read()方法就返回了，线程可以去干别的事，但是需要有一个线程监听这stream中是否有数据准备好。

- 异步非阻塞IO：服务端调用read()方法，若stream中无数据则返回，程序继续向下执行。当stream中有数据时，操作系统会负责把数据拷贝到用户空间，然后通知这个线程，这里的消息通知机制就是异步！

NIO 是一种同步非阻塞的 IO 模型。同步是指线程不断轮询 IO 事件是否就绪，非阻塞是指线程在等待 IO 的时候，可以同时做其他任务。同步的核心就是 Selector，Selector 代替了线程本身轮询 IO 事件，避免了阻塞同时减少了不必要的线程消耗；非阻塞的核心就是通道和缓冲区，当 IO 事件就绪时，可以通过写道缓冲区，保证 IO 的成功，而无需线程阻塞式地等待。

**Channel**：是NIO中通道的接口类，只提供了两个抽象方法，isOpen()判断通道是否打开，close()关闭一个通道。

**AbstractInterruptibleChannel**：是一个抽象类，实现了Channel接口。任何一个通道如果想要实现在中断时实现异步关闭通道，那么必须继承这个类，这主要体现在两个方面：

- 当某个线程阻塞在channel上，而另一个线程调用了channel的close()方法，那么阻塞的线程会收到AsynchronousCloseException

- 如果某个线程阻塞在channel上，另一个线程调用了阻塞线程的interrupt()方法，那么阻塞线程会收到ClosedByInterruptException，并且通道会被关闭。

AbstractInterruptibleChannel抽象类中定义了一组协同方法begin()和end()方法来完成这两个功能，因此当线程执行一个可能阻塞的IO操作时，必须把这个IO操作放在begin()方法和end()方法之间，才能实现channel的异步关闭。

protected final void begin() {

if (interruptor == null) {

interruptor = new Interruptible() {

public void interrupt(Thread target) {

synchronized (closeLock) {

if (!open)

return;

open = false;

interrupted = target;

try {

//收到中断请求后会回调AbstractInterruptibleChannel类的close()方法关闭通道

AbstractInterruptibleChannel.this.implCloseChannel();

} catch (IOException x) { }

}

}};

}

blockedOn(interruptor);

Thread me = Thread.currentThread();

if (me.isInterrupted())

interruptor.interrupt(me);

}

protected final void end(boolean completed)

throws AsynchronousCloseException

{

blockedOn(null);

Thread interrupted = this.interrupted;

//中断触发器不为空，会抛出ClosedByInterruptException

if (interrupted != null && interrupted == Thread.currentThread()) {

interrupted = null;

throw new ClosedByInterruptException();

}

//触发器为空，没有中断，但是在阻塞的过程中channel被关闭了，抛出AsynchronousCloseException

if (!completed && !open)

throw new AsynchronousCloseException();

}

**SelectableChannel**：是一个抽象类，继承于AbstractInterruptibleChannel。需要注册到Selector上的通道必须继承这个类

- public abstract int validOps();获取这个通道支持的事件

- public abstract boolean isRegistered();通道是否注册到了某个Selector上

- public abstract SelectionKey keyFor(Selector sel);这个通道在指定Selector上注册的事件

- public abstract SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException;注册这个通道到指定的Selector上

- public abstract SelectableChannel configureBlocking(boolean block) throws IOException;修改这个通道为非阻塞或阻塞

- public abstract boolean isBlocking();如果这个通道是阻塞模式，返回true

**AbstractSelectableChannel**：是一个抽象类，是SelectableChannel的基础实现类。在AbstractSelectableChannel类中定义了一个SelectionKey数组，记录这个channel注册到了哪些Selector上，定义了一个keyCount记录这个channel注册的次数。并且channel的最初模式设置为阻塞模式。

- 判断这个channel是否注册了：keyCount不为0就表示注册了

public final boolean isRegistered() {

synchronized (keyLock) {

return keyCount != 0;

}

- 获取这个通道在指定Selector上的注册事件，给定Selector，在SelectionKey[]数组中查找Selector与指定Selector相等的SelectionKey

public final SelectionKey keyFor(Selector sel) {

return findKey(sel);

}

- 将这个通道注册到指定Selector上，如果给定的事件ops不是这个通道支持的事件validOps()将会抛出异常，而且阻塞的channel无法注册到Selector上，具体在注册的时候如果channel已经注册到了这个Selector上，那么更新ops和附加信息，如果这个channel还没有注册到这个Selector上，将会调用Selector 类的register(SelectableChannel ch, int ops, Object o)方法完成注册

public final SelectionKey register(Selector sel, int ops, Object att) throws ClosedChannelException{

synchronized (regLock) {

if (!isOpen())

throw new ClosedChannelException();

if ((ops & ~validOps()) != 0)

throw new IllegalArgumentException();

if (blocking)

throw new IllegalBlockingModeException();

SelectionKey k = findKey(sel);

if (k != null) {

k.interestOps(ops);

k.attach(att);

}

if (k == null) {

// New registration

synchronized (keyLock) {

if (!isOpen())

throw new ClosedChannelException();

k = ((AbstractSelector)sel).register(this, ops, att);

addKey(k);

}

return k;

}

- 关闭通道除了需要关闭通道之外，还需要把SelectionKey[]数组上的SelectionKey取消，而SelectionKey类中的cancel()方法又会调用AbstractSelector中的cancel()方法，AbstractSelector类中的cancel()方法将这个SelectionKey加入到cancelledKeys集合中。

protected final void implCloseChannel() throws IOException {

implCloseSelectableChannel();

synchronized (keyLock) {

int count = (keys == null) ? 0 : keys.length;

for (int i = 0; i < count; i++) {

SelectionKey k = keys[i];

if (k != null)

k.cancel();

}

- 判断channel是否是阻塞模式，只有非阻塞channel才能注册到Selector上

public final boolean isBlocking() {

synchronized (regLock) {

return blocking;

}

- 修改channel的阻塞模式

public final SelectableChannel configureBlocking(boolean block) throws IOException {

synchronized (regLock) {

if (!isOpen())

throw new ClosedChannelException();

if (blocking == block)

return this;

if (block && haveValidKeys())

throw new IllegalBlockingModeException();

implConfigureBlocking(block);

blocking = block;

}

return this;

}

**SelectionKey**：每次channel注册到一个Selector都会返回一个SelectionKey对象，因此SelectionKey描述的是一次注册事件中channel和Selector之间的映射关系。

- public abstract SelectableChannel channel();获取SelectionKey对应的channel

- public abstract Selector selector();获取对应SelectionKey对应的channel

- public abstract boolean isValid();这个SelectionKey是否有效

- public abstract void cancel();把这个SelectionKey置为无效

- public abstract int interestOps();获取这个SelectionKey关注的事件

- public abstract SelectionKey interestOps(int ops);设置这个SelectionKey关注的事件

- public abstract int readyOps();这个SelectionKey关注的事件中就绪了的事件

- public static final int OP_READ = 1 << 0;读事件

- public static final int OP_WRITE = 1 << 2;写事件

- public static final int OP_CONNECT = 1 << 3;channel连接到服务器的连接事件

- public static final int OP_ACCEPT = 1 << 4;服务器准备好了接受连接事件

- channel上有读事件就绪、写事件就绪（isWritable）、连接事件就绪(isConnectable)、接受连接事件就绪(isAcceptable)

public final boolean isReadable() {

return (readyOps() & OP_READ) != 0;

}

**AbstractSelectionKey**：是一个抽象类，是SelectionKey的基础实现类。一个channel注册到一个Selector上返回一个SelectionKey，这个SelectionKey初始就是有效的。

- 取消一个SelectionKey，先将SelectionKey置为无效，然后调用Selector的cancel(SelectionKey key)方法完成具体的取消操作

public final void cancel() {

synchronized (this) {

if (valid) {

valid = false;

((AbstractSelector)selector()).cancel(this);

}

**SelectionKeyImpl**：SelectionKey的最终实现类，继承于AbstractSelectionKey。在SelectionKeyImpl类中定义了int类型interestOps、int类型的readyOps，实现了SelectionKey抽象类中的interestOps()、readyOps()、interestOps(int ops)这三个方法。

## Selector ##

Selector是NIO得以实现的核心模块之一，NIO属于同步非阻塞IO，同步的核心就是 Selector，Selector 代替了线程本身轮询 IO 事件，避免了阻塞同时减少了不必要的线程消耗；

**Selector**：是一个抽象类。提供了静态方法open()创建一个Selector对象，open()方法是使用SelectorProvider类完成的。

- public abstract boolean isOpen();这个Selector是否打开了

- public abstract SelectorProvider provider();获取创建这个Selector的SelectorProvider

- public abstract Set keys();获取这个Selector上所有的注册的channel对应的SelectionKey

- public abstract Set selectedKeys();获取这个Selector上所有就绪了的channel对应的SelectionKey

- public abstract int selectNow() throws IOException;非阻塞的执行一次选择操作，没有就绪的channel就立即返回0，否则返回有多少个channel就绪了。

- public abstract int select(long timeout) throws IOException;执行一次选择操作，并且阻塞一段时间，在这段时间里，如果有channel就绪了，将会返回有多少个channel就绪了，如果到达了指定时间还没有channel就绪，就返回0.

- public abstract int select() throws IOException;阻塞执行选择操作，一直阻塞到有channel就绪，然后返回有多少个channel就绪了

- public abstract Selector wakeup();使阻塞在select()方法上的线程立即返回。

Selector selector = Selector.open();

channel.configureBlocking(false);

SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

while(true) {

int readyChannels = selector.select();

if(readyChannels == 0) continue;

Set selectedKeys = selector.selectedKeys();

Iterator keyIterator = selectedKeys.iterator();

while(keyIterator.hasNext()) {

SelectionKey key = keyIterator.next();

if(key.isAcceptable()) {

// a connection was accepted by a ServerSocketChannel.

} else if (key.isConnectable()) {

// a connection was established with a remote server.

} else if (key.isReadable()) {

// a channel is ready for reading

} else if (key.isWritable()) {

// a channel is ready for writing

}

keyIterator.remove();

}

**AbstractSelector**：是一个抽象类，是Selector的基础实现类.

- 定义了一个boolean类型的open变量，初始化为true，表示一个Selector创建时就为打开状态。

- 定义了一个cancelledKeys集合，表示已经取消的SelectionKey的集合。SelectionKey中的cancel()方法会调用AbstractSelector类的cancel()方法，cancel()方法将SelectionKey加入到cancelledKeys集合中。

void cancel(SelectionKey k) { // package-private

synchronized (cancelledKeys) {

cancelledKeys.add(k);

}

- 定义了一个SelectorProvider对象，用于实现Selector类中的provider()方法。

- 定义了一个反注册方法deregister(SelectionKey key)，调用的是channel的方法，将key从SelectionKey[] keys数组中移除

protected final void deregister(AbstractSelectionKey key) {

((AbstractSelectableChannel)key.channel()).removeKey(key);

}

- 定义了一组协同方法begin()和end()，与AbstractInterruptibleChannel类中协同方法类似，在一个可能阻塞的IO操作前使用begin()方法，在IO操作之后使用end()方法，但是这里的协同方法是为了在产生中断之后使select()方法立即返回。

protected final void begin() {

if (interruptor == null) {

interruptor = new Interruptible() {

public void interrupt(Thread ignore) {

//产生中断之后，调用wakeup()方法唤醒select()方法

AbstractSelector.this.wakeup();

}};

}

AbstractInterruptibleChannel.blockedOn(interruptor);

Thread me = Thread.currentThread();

if (me.isInterrupted())

interruptor.interrupt(me);

}

**SelectorImpl**：仍然是一个抽象类，继承于AbstractSelector类，进一步实现了Selector类。

- 定义了一个keys集合，用于实现Selector类的keys()方法，直接返回这个集合。

- 定义了一个selectedKeys集合，用于实现Selector类的selectedKeys()方法，直接返回这个集合。

- 将三个select()方法均委托给一个抽象方法，待子类进一步实现。

//委托给lockAndDoSelect()方法

public int select(long timeout) throws IOException{

if (timeout < 0)

throw new IllegalArgumentException("Negative timeout");

return lockAndDoSelect((timeout == 0) ? -1 : timeout);

}

//调用上一个方法

public int select() throws IOException {

return select(0);

}

//委托给lockAndDoSelect()方法

public int selectNow() throws IOException {

return lockAndDoSelect(0);

}

//同步锁，进一步委托给doSelect(long time)这个抽象方法。

private int lockAndDoSelect(long timeout) throws IOException {

synchronized (this) {

if (!isOpen())

throw new ClosedSelectorException();

synchronized (publicKeys) {

synchronized (publicSelectedKeys) {

return doSelect(timeout);

}

- 进一步实现了close()方法，但是没有完全实现，还是委托给一个抽象方法

//先调用wakeup()方法使select()方法立即返回，然后同步锁，调用抽象方法implClose()

public void implCloseSelector() throws IOException {

wakeup();

synchronized (this) {

synchronized (publicKeys) {

synchronized (publicSelectedKeys) {

implClose();

}

- 初步实现了register(channel, int ops, Object att)方法，委托给抽象方法implRegister(SelectionKey k)方法

protected final SelectionKey register(AbstractSelectableChannel ch, int ops, Object attachment){

if (!(ch instanceof SelChImpl))

throw new IllegalSelectorException();

//新建一个SelectionKey对象

SelectionKeyImpl k = new SelectionKeyImpl((SelChImpl)ch, this);

//设置附加信息

k.attach(attachment);

synchronized (publicKeys) {

implRegister(k);

}

//设置感兴趣事件

k.interestOps(ops);

return k;

}

- 定义了一个方法处理cancelledKeys集合，委托给抽象方法implDereg(SelectionKey k)方法

void processDeregisterQueue() throws IOException {

// Precondition: Synchronized on this, keys, and selectedKeys

//获取cancelledKeys集合

Set cks = cancelledKeys();

synchronized (cks) {

if (!cks.isEmpty()) {

Iterator i = cks.iterator();

while (i.hasNext()) {

SelectionKeyImpl ski = (SelectionKeyImpl)i.next();

try {

implDereg(ski);

} catch (SocketException se) {

IOException ioe = new IOException("Error deregistering key");

ioe.initCause(se);

throw ioe;

} finally {

i.remove();

}

## SelectorProvider ##

SelectorProvider为Selector、DatagramChannel、SocketChannel、ServerSocketChannel、Pipe这些Selector和channel提供打开方法。

**SelectorProvider**：是一个抽象类

- public abstract DatagramChannel openDatagramChannel() throws IOException;打开UDP通信channel

- public abstract Pipe openPipe() throws IOException;打开一个管道

- public abstract AbstractSelector openSelector() throws IOException;打开一个Selector

- public abstract ServerSocketChannel openServerSocketChannel() throws IOException;打开一个服务器socket channel

- public abstract SocketChannel openSocketChannel() throws IOException;打开一个TCP通信channel

**SelectorProviderImpl**：是一个抽象类，是SelectorProvider的基础实现类。

- 打开UDP通信channel，返回的是UDP channel的实现类对象。

public DatagramChannel openDatagramChannel() throws IOException {

return new DatagramChannelImpl(this);

}

- 打开TCP通信channel，返回的是Socket channel的实现类对象。

public SocketChannel openSocketChannel() throws IOException {

return new SocketChannelImpl(this);

}

- 打开TCP通信服务器端的channel，返回的是ServerSocket channel的实现类对象。

public ServerSocketChannel openServerSocketChannel() throws IOException {

return new ServerSocketChannelImpl(this);

}

- 打开一个管道，返回的是pipe实现类对象。

public Pipe openPipe() throws IOException {

return new PipeImpl(this);

}

- 打开Selector的方法没有实现，待子类实现。

**WindowsSelectorProvider **：SelectorProvider的最终实现类，继承于SelectorProviderImpl。实现了Selector的打开方法

- 打开Selector。调用的是Selector的最终实现类WindowsSelectorImpl，通过WindowsSelectorImpl的构造函数返回一个Selector

public AbstractSelector openSelector() throws IOException {

return new WindowsSelectorImpl(this);

}

## 再看 channel ##

**FileChannel**：是一个抽象类，继承于AbstractInterruptibleChannel类，没有继承SelectableChannel，即FileChannel无法注册到Selector上。FileChannel也无法以非阻塞模式读写。通过阻塞方式对文件读写。

- 打开一个FileChannel，一般通过传统IO流获取FileChannel。但是FileChannel类中也定义了open()函数打开一个FileChannel，getChannel()方法底层就是调用的FileChannel的open()方法

FileInputStream fis = new FileInputStream("C:\\mycode\\hello.txt");

FileChannel inChannel = fis.getChannel();

- 从FileChannel读取数据:public abstract int read(ByteBuffer dst) throws IOException;将通道中的数据读出来,并写道指定ByteBuffer中.FileChannel也支持分散写,即写到多个ByteBuffer中.

public abstract long read(ByteBuffer[] dsts, int offset, int length) throws IOException;

- 向FileChannel写数据: public abstract int write(ByteBuffer src) throws IOException;将ByteBuffer缓冲区中的数据写入到channel中.FileChannel也支持聚集写,即多个ByteBuffer中的数据写到channel中.

public abstract long write(ByteBuffer[] srcs, int offset, int length) throws IOException;

- 在FileChannel的某个特定位置进行数据的读/写操作,改变文件position的位置.

//获取文件position的位置

public abstract long position() throws IOException;

//设置文件position

public abstract FileChannel position(long newPosition) throws IOException;

- 将channel中的数据写入到另一个channel，或将另一个channel中的数据写入到这个channel中

//将本channel中的数据写入到另一个“可写入”的channel，从另一个channel的position处开始写入

public abstract long transferTo(long position, long count, WritableByteChannel target) throws IOException;

//将另一个channel中的数据写入到本channel中，从另一个channel的position处开始

public abstract long transferFrom(ReadableByteChannel src, long position, long count) throws IOException;

- 获取channel关联的文件的大小

public abstract long size() throws IOException;

- 截取一个指定大小的文件，截断channel关联的文件为size大小，size之后的数据会被丢弃

public abstract FileChannel truncate(long size) throws IOException;

**DatagramChannel**：是一个抽象类，真正的实现类是其子类DatagramChannelImpl，继承于AbstractSelectableChannel类。因此有阻塞和非阻塞两种模式，在非阻塞模式下可以注册到Selector上。UDP通信的channel。

- 创建一个DatagramChannel。调用DatagramChannel的静态方法open()，通过SelectorProvider去创建DatagramChannelImpl的实例

public static DatagramChannel open() throws IOException {

return SelectorProvider.provider().openDatagramChannel();

}

- DatagramChannel支持的事件：读和写

public final int validOps() {

return (SelectionKey.OP_READ | SelectionKey.OP_WRITE);

}

- 报文流本来是无连接的，在没有连接到一个指定地址时，channel可以同时发送数据报到多个远程地址、也可以同时从多个远程地址接收数据报。通过DatagramChannel的send(ByteBuffer src, SocketAddress add)方法发送数据报到指定远程地址，通过DatagramChannel的receive(ByteBuffer dst)方法从任意远程地址接收数据报，receive()方法会返回一个SocketAddress对象用以标识数据报来自哪个远程地址。在没有建立连接的时候，每一次调用send()方法发送数据报或者调用receive()方法接收数据报时都会接收安全检查。

//发送数据报到指定远程地址

public abstract int send(ByteBuffer src, SocketAddress target) throws IOException;

//从任意远程地址接收数据报，并返回数据来自哪个地址

public abstract SocketAddress receive(ByteBuffer dst) throws IOException;

- 报文流也可以建立连接。建立连接后，channel将只能从指定的远程地址接收数据报、同时也只能发送数据报到指定的远程地址。由于已经连接到了指定的远程地址，因此在发送或者接收数据报的时候可以调用write()方法已经read()方法。write(ByteBuffer src)方法将数据报发送到指定远程地址、read(ByteBuffer dst)方法从指定远程地址接收数据报。指定连接到指定远程地址的channel才能调用write()方法和read()方法，每次调用write()方法和read()方法时不需要接收安全检查。**将DatagramChannel置于已连接的状态可以使除了它所“连接”到的地址之外的任何其他源地址的数据报被忽略。这是很有帮助的，因为不想要的包都已经被网络层丢弃了，从而避免了使用代码来接收、检查然后丢弃包的麻烦。**

//将channel连接到指定远程地址

public abstract DatagramChannel connect(SocketAddress remote) throws IOException;

//断开channel与远程地址间的连接

public abstract DatagramChannel disconnect() throws IOException;

//channel是否连接到了某个远程地址

public abstract boolean isConnected();

//发送数据报到指定远程地址，支持聚集写

public abstract int write(ByteBuffer src) throws IOException;

public final long write(ByteBuffer[] srcs) throws IOException {

return write(srcs, 0, srcs.length);

}

public abstract long write(ByteBuffer[] srcs, int offset, int length) throws IOException;

//从指定远程地址接收数据报，支持分散读

public abstract int read(ByteBuffer dst) throws IOException;

public final long read(ByteBuffer[] dsts) throws IOException {

return read(dsts, 0, dsts.length);

}

public abstract long read(ByteBuffer[] dsts, int offset, int length) throws IOException;

- 如果channel处于阻塞模式：调用send()方法或者write()方法，调用线程可能会休眠直到数据报被加入传输队列。如果channel是非阻塞的：send()方法或者write()方法、read()返回值要么是字节缓冲区的字节数，要么是“0”，receive()方法的返回值要么是远程地址对象要么是null。

- 报文流可以绑定也可以不绑定。如果channel负责监听，那么必须绑定到一个指定端口，channel将会一直监听这个端口。当channel没有绑定的时候，仍然能够接收数据报，使用的是动态分配的端口号。已经绑定的channel将从指定端口接收或者发送数据报。

- 报文流是不可靠传输。1）假如receive()方法提供的ByteBuffer没有足够的剩余空间来存放您正在接收的数据包，没有被填充的字节都会被悄悄地丢弃。2）如果send()方法给定的ByteBuffer传输队列没有足够空间来承载整个数据报，那么什么内容都不会被发送。3）传输过程中的协议可能将数据报分解成碎片。例如，以太网不能传输超过1,500个字节左右的包。如果您的数据报比较大，那么就会存在被分解成碎片的风险，成倍地增加了传输过程中包丢失的几率。被分解的数据报在目的地会被重新组合起来，接收者将看不到碎片。但是，如果有一个碎片不能按时到达，那么整个数据报将被丢弃。

**SocketChannel**：是一个抽象类，真正的实现类是其子类SocketChannelImpl，继承于AbstractSelectableChannel类。因此有阻塞和非阻塞两种模式，在非阻塞模式下可以注册到Selector上。客户端的TCP通信的channel。

- 创建一个SocketChannel对象，通过SocketChannel的静态方法open委托给SelectorProvider类创建SocketChannelImpl类对象。

public static SocketChannel open() throws IOException {

return SelectorProvider.provider().openSocketChannel();

}

- SocketChannel支持的事件：读、写、发起连接

public final int validOps() {

return (SelectionKey.OP_READ | SelectionKey.OP_WRITE | SelectionKey.OP_CONNECT);

}

- channel必须在使用之前连接到远程地址，在阻塞模式下，连接操作会一直阻塞直到连接成功或者失败；在非阻塞模式下，连接操作即使没有连接成功也会立刻返回，因此需要通过finishConnect()方法判断是否连接成功并一直尝试连接。

//channel是否成连接到一个远程地址

public abstract boolean isConnected();

//channel是否正处于连接中

public abstract boolean isConnectionPending();

//channel连接一个远程地址

public abstract boolean connect(SocketAddress remote) throws IOException;

//channel是否完成了连接

public abstract boolean finishConnect() throws IOException;

while(!channel.finishConnect()){

//由于必须在连接成功之后才能进行IO操作，必须等待连接成功

doSomethingElse();

}

- 往channel中写数据，支持聚集写。写完之后，Selector的select()方法会检测到这个channel的WRITE事件就绪了

public abstract int write(ByteBuffer src) throws IOException;

public final long write(ByteBuffer[] srcs) throws IOException {

return write(srcs, 0, srcs.length);

}

public abstract long write(ByteBuffer[] srcs, int offset, int length) throws IOException;

- 从channel中读取数据，支持分散读。Selector的select()方法会检测到这个channel的READ事件就绪了

public abstract int read(ByteBuffer dst) throws IOException;

public final long read(ByteBuffer[] dsts) throws IOException {

return read(dsts, 0, dsts.length);

}

public abstract long read(ByteBuffer[] dsts, int offset, int length) throws IOException;

SocketChannel的一个实例：

public class MyClient {

private static Selector selector = null;

private volatile static boolean stop = false;

private static SocketChannel channel = null;

public static void main(String[] args) {

selector = Selector.open();

try {

channel = SocketChannel.open();

channel.configureBlocking(false);

channel.connect(new InetSocketAddress("127.0.0.1", 7777));

//注册一个连接请求事件

channel.register(selector, SelectionKey.OP_CONNECT);

try {

while (!stop) {

selector.select();

Set selectedKeys = selector.selectedKeys();

Iterator iterator = selectedKeys.iterator();

while (iterator.hasNext()) {

SelectionKey key = iterator.next();

//连接就绪

if (key.isConnectable()) {

//服务器接收了连接请求

SocketChannel sc = (SocketChannel) key.channel();

if (sc.finishConnect()) {

// 将关注的事件变成read

sc.register(selector, SelectionKey.OP_READ);

doWrite(sc, "dddddd");

}

// 读就绪

if (key.isReadable()) {

//服务器有数据过来了，处理数据，再发送数据

}

iterator.remove();

}

} catch (IOException e) {

e.printStackTrace();

}

} catch (ClosedChannelException e) {

System.out.println("client: 失去主机连接");

e.printStackTrace();

} catch (IOException e) {

e.printStackTrace();

}

**ServerSocketChannel**：是一个抽象类，真正的实现类是其子类ServerSocketChannelImpl，继承于AbstractSelectableChannel类。因此有阻塞和非阻塞两种模式，在非阻塞模式下可以注册到Selector上。服务器端的TCP通信的channel。

- 创建一个ServerSocketChannel，通过ServerSocketChannel的静态方法open委托给SelectorProvider类创建ServerSocketChannelImpl类对象。

public static ServerSocketChannel open() throws IOException {

return SelectorProvider.provider().openServerSocketChannel();

}

- ServerSocketChannel支持的事件：接收连接

public final int validOps() {

return SelectionKey.OP_ACCEPT;

}

- ServerSocketChannel必须先绑定到一个端口上，一直监听这个端口。

public abstract ServerSocketChannel bind(SocketAddress local, int backlog) throws IOException;

- 获取与这个ServerSocketChannel关联的SocketChannel，即发起连接请求的SocketChannel

public abstract SocketChannel accept() throws IOException;

ServerSocketChannel的一个实例：

public class MyService {

public static Selector selector = null;

public static void main(String[] args) {

selector = Selector.open();// 打开selector

ServerSocketChannel server = ServerSocketChannel.open();

server.socket().bind(new InetSocketAddress(7777), 1024);

server.configureBlocking(false);

//服务器开始监听等待连接，注册ACCEPT事件

server.register(selector, SelectionKey.OP_ACCEPT);

while (true) {

try {

selector.select(1000); // 阻塞selector

// ================如果有新连接

Set selectedKeys = selector.selectedKeys();// 获得事件集合;

// ================遍历selectedKeys

Iterator iterator = selectedKeys.iterator();

SelectionKey key = null;

while (iterator.hasNext()) {

key = iterator.next();// 获得到当前的事件

// ===============处理事件

// 连接就绪，有客户端请求连接，注册的事件发生

if (key.isAcceptable()) {

// 获得对应的ServerSocketChannel

ServerSocketChannel ssc = (ServerSocketChannel) key.channel();

// 得到对应的SocketChannel

SocketChannel channel = ssc.accept();

// 处理socketChannel

channel.configureBlocking(false);

channel.register(selector, SelectionKey.OP_READ);

}

// 读就绪

if (key.isReadable()) {

//客户端有数据过来了，之前注册的READ事件来了

}

// ===============

iterator.remove(); // 移除事件

}

} catch (IOException e) {

e.printStackTrace();

}

｝

Java NIO

推荐阅读更多精彩内容