强制停止在外部线程上运行的Java Files.copy()

Question

在Java 8之前，这里的答案似乎是一个有效的解决方案：如何在Java中取消Files.copy()操作？

但现在它不起作用了，因为ExtendedCopyOption.INTERRUPTIBLE是私有的。

---

基本上，我需要从给定的URL下载文件，并使用Files.copy()将其保存到我的本地文件系统中。 目前，我正在使用JavaFX服务，因为我需要在ProgressBar中显示进度。

然而，如果操作时间过长，我不知道如何阻止运行Files.copy()的线程。 至少不想使用Thread.stop()。即使是Thread.interrupt()也无法成功。

我还希望如果互联网连接不可用，操作能够正常终止。

为了测试没有互联网连接时的情况，我拔掉以太网电缆，并在3秒后重新插入它。 不幸的是，Files.copy()只有在我重新插入以太网电缆时才返回，而我希望它立即失败。

据我所见，内部Files.copy()正在运行一个循环，这会防止线程退出。

---

测试人员（下载OBS Studio exe）：

/**
 * @author GOXR3PLUS
 *
 */
public class TestDownloader extends Application {

    /**
     * @param args
     */
    public static void main(String[] args) {
    launch(args);
    }

    @Override
    public void start(Stage primaryStage) throws Exception {
    // Block From exiting
    Platform.setImplicitExit(false);

    // Try to download the File from URL
    new DownloadService().startDownload(
        "https://github.com/jp9000/obs-studio/releases/download/17.0.2/OBS-Studio-17.0.2-Small-Installer.exe",
        System.getProperty("user.home") + File.separator + "Desktop" + File.separator + "OBS-Studio-17.0.2-Small-Installer.exe");

    }

}

DownloadService:

@sillyfly的评论使用FileChannel并删除File.copy似乎只在调用Thread.interrupt()时起作用，但当网络不可用时它并未退出。

import java.io.File;
import java.net.URL;
import java.net.URLConnection;
import java.nio.channels.Channels;
import java.nio.channels.FileChannel;
import java.nio.file.StandardOpenOption;
import java.util.logging.Level;
import java.util.logging.Logger;

import javafx.concurrent.Service;
import javafx.concurrent.Task;

/**
 * JavaFX Service which is Capable of Downloading Files from the Internet to the
 * LocalHost
 * 
 * @author GOXR3PLUS
 *
 */
public class DownloadService extends Service<Boolean> {

    // -----
    private long totalBytes;
    private boolean succeeded = false;
    private volatile boolean stopThread;

    // CopyThread
    private Thread copyThread = null;

    // ----
    private String urlString;
    private String destination;

    /**
     * The logger of the class
     */
    private static final Logger LOGGER = Logger.getLogger(DownloadService.class.getName());

    /**
     * Constructor
     */
    public DownloadService() {
    setOnFailed(f -> System.out.println("Failed with value: " + super.getValue()+" , Copy Thread is Alive? "+copyThread.isAlive()));
    setOnSucceeded(s -> System.out.println("Succeeded with value: " + super.getValue()+" , Copy Thread is Alive? "+copyThread.isAlive()));
    setOnCancelled(c -> System.out.println("Succeeded with value: " + super.getValue()+" , Copy Thread is Alive? "+copyThread.isAlive()));
    }

    /**
     * Start the Download Service
     * 
     * @param urlString
     *            The source File URL
     * @param destination
     *            The destination File
     */
    public void startDownload(String urlString, String destination) {
    if (!super.isRunning()) {
        this.urlString = urlString;
        this.destination = destination;
        totalBytes = 0;
        restart();
    }
    }

    @Override
    protected Task<Boolean> createTask() {
    return new Task<Boolean>() {
        @Override
        protected Boolean call() throws Exception {

        // Succeeded boolean
        succeeded = true;

        // URL and LocalFile
        URL urlFile = new URL(java.net.URLDecoder.decode(urlString, "UTF-8"));
        File destinationFile = new File(destination);

        try {
            // Open the connection and get totalBytes
            URLConnection connection = urlFile.openConnection();
            totalBytes = Long.parseLong(connection.getHeaderField("Content-Length"));





            // --------------------- Copy the File to External Thread-----------
            copyThread = new Thread(() -> {

            // Start File Copy
            try (FileChannel zip = FileChannel.open(destinationFile.toPath(), StandardOpenOption.CREATE,
                StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.WRITE)) {

                zip.transferFrom(Channels.newChannel(connection.getInputStream()), 0, Long.MAX_VALUE);


                // Files.copy(dl.openStream(), fl.toPath(),StandardCopyOption.REPLACE_EXISTING)

            } catch (Exception ex) {
                stopThread = true;
                LOGGER.log(Level.WARNING, "DownloadService failed", ex);
            }

            System.out.println("Copy Thread exited...");
            });
            // Set to Daemon
            copyThread.setDaemon(true);
            // Start the Thread
            copyThread.start();
            // -------------------- End of Copy the File to External Thread-------






            // ---------------------------Check the %100 Progress--------------------
            long outPutFileLength;
            long previousLength = 0;
            int failCounter = 0;
            // While Loop
            while ((outPutFileLength = destinationFile.length()) < totalBytes && !stopThread) {

            // Check the previous length
            if (previousLength != outPutFileLength) {
                previousLength = outPutFileLength;
                failCounter = 0;
            } else
                ++failCounter;

            // 2 Seconds passed without response
            if (failCounter == 40 || stopThread)
                break;

            // Update Progress
            super.updateProgress((outPutFileLength * 100) / totalBytes, 100);
            System.out.println("Current Bytes:" + outPutFileLength + " ,|, TotalBytes:" + totalBytes
                + " ,|, Current Progress: " + (outPutFileLength * 100) / totalBytes + " %");

            // Sleep
            try {
                Thread.sleep(50);
            } catch (InterruptedException ex) {
                LOGGER.log(Level.WARNING, "", ex);
            }
            }

            // 2 Seconds passed without response
            if (failCounter == 40)
            succeeded = false;
           // --------------------------End of Check the %100 Progress--------------------

        } catch (Exception ex) {
            succeeded = false;
            // Stop the External Thread which is updating the %100
            // progress
            stopThread = true;
            LOGGER.log(Level.WARNING, "DownloadService failed", ex);
        }







        //----------------------Finally------------------------------

        System.out.println("Trying to interrupt[shoot with an assault rifle] the copy Thread");

        // ---FORCE STOP COPY FILES
        if (copyThread != null && copyThread.isAlive()) {
            copyThread.interrupt();
            System.out.println("Done an interrupt to the copy Thread");

            // Run a Looping checking if the copyThread has stopped...
            while (copyThread.isAlive()) {
            System.out.println("Copy Thread is still Alive,refusing to die.");
            Thread.sleep(50);
            }
        }

        System.out.println("Download Service exited:[Value=" + succeeded + "] Copy Thread is Alive? "
            + (copyThread == null ? "" : copyThread.isAlive()));

        //---------------------- End of Finally------------------------------




        return succeeded;
        }

    };
    }

}

有趣的问题：

1-> java.lang.Thread.interrupt()做什么？

- GOXR3PLUS

---

如果线程没有陷入循环，它会自动停止。现在，如果线程继续运行，这意味着它陷入了循环，您似乎想使用Thread.interrupt()来停止它。但是，只有在线程内部发生任何中断（例如Thread.pause(x)）时，此操作才有效。否则JVM不知道何时中断线程，在文件写入操作中间进行中断并不好。 - Voltboyy

我不确定，但看起来你需要使用类似于 FileChannel 的东西，就像 这个问题 中描述的那样。虽然这似乎有点过头了，也许有更简单的方法。 - Itai

@sillyfly 我认为你的选项完美无缺，对于连接丢失测试，可以使用超时，就像这个问题中所示。 - Voltboyy

我注意到您还没有接受任何答案，请考虑在某个时候这样做。 - GhostCat

@GhostCat 我的朋友，没有一个答案是完全正确的。 - GOXR3PLUS

Answer 1

我强烈建议您使用FileChannel。它拥有transferFrom()方法，当运行它的线程被中断时会立即返回。（这里的Javadoc说它应该引发ClosedByInterruptException，但它并没有。）

try (FileChannel channel = FileChannel.open(Paths.get(...), StandardOpenOption.CREATE,
                                            StandardOpenOption.WRITE)) {
    channel.transferFrom(Channels.newChannel(new URL(...).openStream()), 0, Long.MAX_VALUE);
}

它还有比其 java.io 替代品更好的性能潜力。（但是，实现 Files.copy() 的方法可能会选择委托此方法而不是自己执行复制。）

---

这是一个可重用的 JavaFX Service 示例，允许你从互联网获取资源并将其保存到本地文件系统，如果操作时间过长则会自动优雅地终止。

• 服务任务（由 createTask() 生成）是文件通道 API 的使用者。
• 单独使用ScheduledExecutorService 来处理时间限制。
• 始终遵循 好的做法 来扩展 Service。
• 如果选择使用这种高级方法，则无法跟踪任务的进度。
• 如果连接不可用，则 transferFrom() 应该最终返回而不抛出异常。

要启动服务（可以从任何线程执行）：

DownloadService downloadService = new DownloadService();
downloadService.setRemoteResourceLocation(new URL("http://speedtest.ftp.otenet.gr/files/test1Gb.db"));
downloadService.setPathToLocalResource(Paths.get("C:", "test1Gb.db"));
downloadService.start();

然后取消它（否则它将在时间到期后自动取消）：

downloadService.cancel();

请注意，同一项服务可以被重复使用，在重新开始之前，请确保重置它：

downloadService.reset();

这里是DownloadService类：

public class DownloadService extends Service<Void> {

    private static final long TIME_BUDGET = 2; // In seconds

    private final ScheduledExecutorService watchdogService =
            Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
                private final ThreadFactory delegate = Executors.defaultThreadFactory();

                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = delegate.newThread(r);
                    thread.setDaemon(true);
                    return thread;
                }
            });
    private Future<?> watchdogThread;

    private final ObjectProperty<URL> remoteResourceLocation = new SimpleObjectProperty<>();
    private final ObjectProperty<Path> pathToLocalResource = new SimpleObjectProperty<>();

    public final URL getRemoteResourceLocation() {
        return remoteResourceLocation.get();
    }

    public final void setRemoteResourceLocation(URL remoteResourceLocation) {
        this.remoteResourceLocation.set(remoteResourceLocation);
    }

    public ObjectProperty<URL> remoteResourceLocationProperty() {
        return remoteResourceLocation;
    }

    public final Path getPathToLocalResource() {
        return pathToLocalResource.get();
    }

    public final void setPathToLocalResource(Path pathToLocalResource) {
        this.pathToLocalResource.set(pathToLocalResource);
    }

    public ObjectProperty<Path> pathToLocalResourceProperty() {
        return pathToLocalResource;
    }

    @Override
    protected Task<Void> createTask() {
        final Path pathToLocalResource = getPathToLocalResource();
        final URL remoteResourceLocation = getRemoteResourceLocation();
        if (pathToLocalResource == null) {
            throw new IllegalStateException("pathToLocalResource property value is null");
        }
        if (remoteResourceLocation == null) {
            throw new IllegalStateException("remoteResourceLocation property value is null");
        }

        return new Task<Void>() {
            @Override
            protected Void call() throws IOException {
                try (FileChannel channel = FileChannel.open(pathToLocalResource, StandardOpenOption.CREATE,
                                                            StandardOpenOption.WRITE)) {
                    channel.transferFrom(Channels.newChannel(remoteResourceLocation.openStream()), 0, Long.MAX_VALUE);
                }
                return null;
            }
        };
    }

    @Override
    protected void running() {
        watchdogThread = watchdogService.schedule(() -> {
            Platform.runLater(() -> cancel());
        }, TIME_BUDGET, TimeUnit.SECONDS);
    }

    @Override
    protected void succeeded() {
        watchdogThread.cancel(false);
    }

    @Override
    protected void cancelled() {
        watchdogThread.cancel(false);
    }

    @Override
    protected void failed() {
        watchdogThread.cancel(false);
    }

}

- spongebob

Answer 2

其他答案/评论没有涵盖一个重要的方面，那就是你的错误假设：

我想要的是在没有网络连接时立即失败。

这并不容易。TCP协议栈/状态机实际上是一件相当复杂的事情；而且根据您的上下文（操作系统类型、TCP协议栈实现、内核参数等），可能会出现网络分区的情况，发送方在15分钟或更长时间内都没有察觉到。请点击这里了解更多详情。

换句话说："拔掉插头"并不等同于"立即断开"您现有的TCP连接。只是为了记录：您无需手动插拔电缆来模拟网络故障。在合理的测试环境中，像iptables这样的工具可以为您完成这项工作。

- GhostCat

Answer 3

你似乎需要一个异步/可取消的HTTP GET，这可能很困难。

问题在于，如果读取等待更多数据（电缆被拔出），它不会退出，直到套接字死亡或新数据进入为止。

有几条路径可以选择，可以通过调整套接字工厂来设置良好的超时时间，使用带有超时的http客户端等。

我建议看一下Apache Http Components，它基于java NIO Sockets提供了非阻塞HTTP。

- minus