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,
0,
size,
size,
off))
: (CharBuffer)(new ByteBufferAsCharBufferL(this,
-1,
0,
size,
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();
}
}
}
｝
