IO與NIO –中斷，超時和緩沖區

假設有一個系統有時需要將文件復制到幾個位置，但是這種方式在響應速度至關重要的情況下。換句話說，如果由于某種原因文件系統過載，并且我們無法在不到一秒鐘的時間內寫入文件，則應該放棄。

ExecutorService是一項非常方便的工作工具。您可以輕松地將其用于并行執行多個任務（每個任務都寫入不同的文件系統）。 Yuo還可以告訴它在超時后放棄，它將為您打斷他們。完美，正是我們所需要的。

腳手架看起來像這樣：

void testCopy() throws Exception {ThreadPoolExecutor exec = (ThreadPoolExecutor) Executors.newCachedThreadPool();final long start = System.currentTimeMillis();Callable<Object> task = new Callable<Object>() {@Overridepublic Object call() throws Exception {try {copy("a.bin", "b.bin");} catch (Exception e) {e.printStackTrace();}System.out.println("Call really finished after: "+ (System.currentTimeMillis() - start));return null;}};Collection<Callable<Object>> taskWrapper = Arrays.asList(task);List<Future<Object>> futures = exec.invokeAll(taskWrapper, 50,TimeUnit.MILLISECONDS);System.out.println("invokeAll finished after: "+ (System.currentTimeMillis() - start));System.out.println("Future.isCancelled? "+ futures.get(0).isCancelled());Thread.sleep(20);System.out.println("Threads still active: " + exec.getActiveCount());
}

為了在低負載的運行狀況良好的系統上模擬對超時的響應，我使用了100 MB的文件并且超時非常短。任務總是超時，我的系統無法在50毫秒內復制100 MB。

我期望得到以下結果：

大約50毫秒后， invokeAll完成。
Future.isCancelled? 是真的。
活動線程計數為0。通過睡眠可以消除某些邊緣情況。長話短說，它給了復制功能一些時間來檢測中斷。
通話大約在50毫秒后真正結束。這非常重要，我絕對不希望取消任務后繼續執行IO操作。在較高的負載下，這會導致過多的線程卡在虛假的IO中。

以防萬一，這些測試是在64位Windows 7上的Oracle 1.6 JVM上運行的。

解決方案1：流復制

第一次嘗試可能很簡單-使用帶有緩沖區和經典IO的循環，如下所示：

private void copy(String in, String out) throws Exception {FileInputStream fin = new FileInputStream(in);FileOutputStream fout = new FileOutputStream(out);byte[] buf = new byte[4096];int read;while ((read = fin.read(buf)) > -1) {fout.write(buf, 0, read);}fin.close();fout.close();
}

這就是所有流行的流復制庫做的，包括IOUtils Apache的共享和ByteStreams番石榴。

它也不幸地失敗了：

invokeAll finished after: 53
Future.isCancelled? true
Threads still active: 1
Call really finished after: 338

原因很明顯：在循環中或任何地方都不檢查線程中斷狀態，因此線程可以正常繼續。

解決方案2：通過復制檢查流是否中斷

讓我們解決這個問題！一種方法是：

while ((read = fin.read(buf)) > -1) {fout.write(buf, 0, read);if (Thread.interrupted()) {throw new IOException("Thread interrupted, cancelling");}
}

現在可以正常工作了，打印：

invokeAll finished after: 52
java.io.IOException: Thread interrupted, cancellingat TransferTest.copyInterruptingStream(TransferTest.java:75)at TransferTest.access$0(TransferTest.java:66)at TransferTest$1.call(TransferTest.java:25)at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)at java.util.concurrent.FutureTask.run(FutureTask.java:138)at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)Future.isCancelled? trueat java.lang.Thread.run(Thread.java:662)Call really finished after: 53
Threads still active: 0

很好，但是我覺得不滿意。它看起來很臟，我對自己的IO庫中的這段代碼并不特別滿意。必須有更好的方法，這將我們帶到……

解決方案3：帶傳輸的NIO

NIO具有這個不錯的功能，它實際上尊重線程中斷。如果在線程中斷后嘗試讀取或寫入通道，則會收到ClosedByInterruptException 。

那正是我所需要的。由于某種原因，我還在StackOverflow上閱讀了以下答案：

“如果不需要，請不要使用緩沖區。如果目標是其他磁盤或NIC，為什么還要復制到內存中？對于較大的文件，確保的延遲是不平凡的。（…）使用FileChannel.transferTo()或FileChannel.transferFrom() 。此處的主要優勢在于，JVM使用操作系統對DMA（直接內存訪問）的訪問（如果存在）。（這取決于實現方式，但是可以在通用CPU上使用現代的Sun和IBM版本。）發生的情況是，數據直接通過/從磁盤，到總線，再到目的地……直接通過RAM傳遞任何電路或CPU。”

太好了，讓我們做吧！

private void copy(String in, String out) throws Exception {FileChannel fin = new FileInputStream(in).getChannel();FileChannel fout = new FileOutputStream(out).getChannel();fout.transferFrom(fin, 0, new File(in).length());fin.close();fout.close();
}

輸出：

invokeAll finished after: 52
Future.isCancelled? true
Threads still active: 1
java.nio.channels.ClosedByInterruptExceptionat java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184)at sun.nio.ch.FileChannelImpl.size(FileChannelImpl.java:304)at sun.nio.ch.FileChannelImpl.transferFrom(FileChannelImpl.java:587)at TransferTest.copyNioTransfer(TransferTest.java:91)at TransferTest.access$0(TransferTest.java:87)at TransferTest$1.call(TransferTest.java:27)at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)at java.util.concurrent.FutureTask.run(FutureTask.java:138)at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)at java.lang.Thread.run(Thread.java:662)
Call really finished after: 146

我所要做的只是簡單地調用transferFrom 。非常簡潔，并承諾會從硬件和操作系統中獲得如此多的支持……但是請稍等一下，為什么要花146毫秒？我的意思是，146毫秒比第一次測試中的338毫秒快得多，但是我希望它在50毫秒后終止。

讓我們在更大的文件（大約1.5 GB）上重復測試：

invokeAll finished after: 9012
Future.isCancelled? true
Threads still active: 1
java.nio.channels.ClosedByInterruptExceptionat java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184)(...)
Call really finished after: 9170

那有多可怕？這可能是可能發生的最糟糕的事情：

任務未及時中斷。 9秒太長了，我預計大約50毫秒。
在整個操作過程中（9秒）， invokeAll被阻止。我勒個去？

解決方案4 –帶緩沖的NIO

事實證明，我確實需要一些緩沖。讓我們嘗試一下：

private void copyNioBuffered(String in, String out) throws Exception {FileChannel fin = new FileInputStream(in).getChannel();FileChannel fout = new FileOutputStream(out).getChannel();ByteBuffer buff = ByteBuffer.allocate(4096);while (fin.read(buff) != -1 || buff.position() > 0) {buff.flip();fout.write(buff);buff.compact();}fin.close();fout.close();
}

輸出：

invokeAll finished after: 52
Future.isCancelled? true
java.nio.channels.ClosedByInterruptExceptionat java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:184)at sun.nio.ch.FileChannelImpl.write(FileChannelImpl.java:203)at TransferTest.copyNioBuffered(TransferTest.java:105)at TransferTest.access$0(TransferTest.java:98)at TransferTest$1.call(TransferTest.java:29)at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:303)at java.util.concurrent.FutureTask.run(FutureTask.java:138)at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)at java.lang.Thread.run(Thread.java:662)
Call really finished after: 55
Threads still active: 0

現在正是我所需要的。它本身就考慮到中斷，因此我不需要整個IO實用程序進行那些繁瑣的檢查。 怪癖：不同類型的頻道

如果我的IO實用程序僅用于復制按名稱獲取的文件，如下所示：

static public void copy(String source, String destination)

…然后很容易為NIO重寫方法。

但是，如果它是在流上運行的更通用的簽名，該怎么辦？

static public void copy(InputStream source, OutputStream destination)

NIO有一個Channels實用程序，它具有非常有用的方法，例如：

public static ReadableByteChannel newChannel(InputStream in)
public static WritableByteChannel newChannel(OutputStream out)

因此，似乎我們可以使用此幫助程序包裝流并從可中斷的NIO API中受益。在我們查看源代碼之前：

public static WritableByteChannel newChannel(final OutputStream out) {if (out == null) {throw new NullPointerException();}if (out instanceof FileOutputStream &&FileOutputStream.class.equals(out.getClass())) {return ((FileOutputStream)out).getChannel();}return new WritableByteChannelImpl(out);
}private static class WritableByteChannelImplextends AbstractInterruptibleChannel // Not really interruptibleimplements WritableByteChannel
{
// ... Ignores interrupts completely

小心！如果您的流是文件流，它們將是可中斷的。否則，您很不走運–它只是一個愚蠢的包裝器，更像是API兼容性的適配器。假設殺死，總是檢查源頭。

參考： IO與NIO – 松鼠博客上來自我們JCG合作伙伴 Konrad Garus的中斷，超時和緩沖區。