Java NIO中InterruptibleChannel接口表示通道IO阻塞时可被异步的关闭和中断。
public interface InterruptibleChannel extends Channel{
/**
* Closes this channel.
*
* Any thread currently blocked in an I/O operation upon this channel
* will receive an AsynchronousCloseException.
* throws IOException If an I/O error occurs
*/
public void close() throws IOException;
}
AbstractInterruptibleChannel实现了InterruptibleChannel接口，并提供了实现可中断IO机制的重要的方法，比如begin()，end()。在调用IO操作时，需要如此调用才可实现中断：
try {
begin();
completed = ...; // Perform blocking I/O operation
return ...; // Return result
} finally {
end(completed);
}
completed参数说明I/O操作是不是真的完成。比如在read操作中，只有completed返回true，才真的表示读到了buffer中一些数据。
在具体的可中断channel中，如FileChannel，SocketChannel，DatagramChannel必须实现implCloseChannel方法，因为这是表示当前channel中断时希望做哪些操作的回调函数。
begin方法：
在io操作的开始之前负责添加Channel的中断处理器到当前线程
// -- Interruption machinery,持有中断对象 --
private Interruptible interruptor;
//持有实现中断机制的线程对象
private volatile Thread interrupted;
protected final void begin() {
//实例化中断处理对象，
//标记被中断的线程，并回调implCloseChannel，关闭channel
if (interruptor == null) {
interruptor = new Interruptible() {
public void interrupt(Thread target) {
synchronized (closeLock) {
//如果当前channel已被关闭，则直接返回
if (!open)
return;
//设置标志位，登记被中断的线程
open = false;
interrupted = target;
try {
//调用具体的实现关闭当前channel
AbstractInterruptibleChannel.this.implCloseChannel();
} catch (IOException x) { }
}
}};
}
//登记当前线程（当前channel）的中断处理对象
blockedOn(interruptor);
Thread me = Thread.currentThread();
//判断当前线程是不是已经被中断，如果被中断，则手动触发中断操作，关闭channel
if (me.isInterrupted())
interruptor.interrupt(me);
}
可见，nio的channel中断操作，是通过挂载 Interruptible自定义的中断处理对象，当当前线程被中断时，通过回调关闭channel的函数：implCloseChannel，从而实现对当前线程中断的响应。
当前线程如何绑定中断机制对象？看下blockedOn的实现：
// -- sun.misc.SharedSecrets --
static void blockedOn(Interruptible intr) { // package-private
sun.misc.SharedSecrets.getJavaLangAccess().blockedOn(Thread.currentThread(), intr);
}
通过JavaLangAccess类将当前线程的blocker设置为interruptor。
当Thread在中断时，如何调用nio的中断处理器？
public void interrupt() {
if (this != Thread.currentThread())
checkAccess();
synchronized (blockerLock) {
Interruptible b = blocker;
if (b != null) {
//nio中断处理逻辑
interrupt0(); // Just to set the interrupt flag
b.interrupt(this); //调用中断处理器的中断方法
return;
}
}
//普通流程，仅设置中断处理标志
interrupt0();
}
end 方法：
protected final void end(boolean completed)
throws AsynchronousCloseException {
//释放当前线程的处理器引用，避免线程一直存活无法回收掉中断处理器
blockedOn(null);
Thread interrupted = this.interrupted;
if (interrupted != null && interrupted == Thread.currentThread()) {
//如果当前线程被中断，则抛出ClosedByInterruptException异常，表示Channel因为线程中断而被关闭了，IO操作也随之中断了。
interrupted = null;
throw new ClosedByInterruptException();
}
if (!completed && !open){
//当前线程发现Channel被关闭了，并且是读取还未执行完毕的情况，则抛出AsynchronousCloseException异常，表示Channel被异步关闭了。
throw new AsynchronousCloseException();
}
如果手动实现一个可中断的channel时，手动处理中断应该是：
this.outstream = Channels.newChannel(outstream);
try {
outstream.write(message);
}
catch(AsynchronousCloseException e) {
System.out.println("Another thread closed the stream while this one was blocking on I/O!");
}
catch(ClosedByInterruptException e) {
System.out.println("This thread has been interrupted while blocking on I/O!");
}
参考资料：
http://www.importnew.com/29430.html
https://stackoverflow.com/questions/10087914/how-do-i-use-interruptiblechannel
网友评论