停止执行java线程和join()的角色

Question

我的程序分析了大量文档，偶尔会得到一个导致无限或非常长循环的页面。这是无法预先分析的。我想杀死特定的页面，然后继续下一个页面（丢弃违规页面的任何结果）。我读过这样的SO答案：如何在Java中一段时间后停止执行？并编写了以下代码：

// main program 
    private void runThread() throws InterruptedException {
        long timeout = 15000L;
        RunPageAnalyzer runPageAnalyzer = new RunPageAnalyzer(this);
        Thread t = new Thread(runPageAnalyzer); 
        long startTime = System.currentTimeMillis();
        t.start();
        while (t.isAlive()) {
            t.join(1000);
            long delta = System.currentTimeMillis() - startTime;
            LOG.debug("delta: "+delta);
            if (delta > timeout && t.isAlive()) {
                t.interrupt();
                t.join;
                break;
            }           
        }
    }

线程调用的同一类中的方法

    void runActions() {
        // variable length calculation which should be abandoned if too long
    }

和Runnable：

    class RunPageAnalyzer implements Runnable {
    private PageAnalyzerAction pageAnalyzerAction;
        public RunPageAnalyzer(PageAnalyzerAction pageAnalyzerAction) {
            this.pageAnalyzerAction = pageAnalyzerAction;
        }
        public void run() {
        try {
            pageAnalyzerAction.runActions();
        } catch (Exception e) {
            LOG.debug("Exception running thread ", e);
        }
    }

正常终止runActions（）的输出似乎还可以：

    =========== page 1 =============
13863 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - pageActions: 24 on page 0
14863 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 1000
15864 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 2001
16864 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 3001
16975 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 3112
16975 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - finished page

但是当超过时间限制时，进程挂起在CCD_ 1中。

    =========== page 2 =============
16975 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - pageActions: 24 on page 0
17976 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 1001
18976 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 2001
// ...
30976 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 14001
31976 [main] DEBUG org.xmlcml.graphics.control.page.PageAnalyzerAction  - delta: 15001

如果我省略了t.join()，那么进程的行为与我预期的一样，但我担心这可能只是建立了大量的线程，这将在以后成为一个问题。

更新：到目前为止，答案表明这不是一件小事（我觉得标准的Java示例/教程没有太大帮助）。关键是runActions()必须明确地知道它可能被中断。join()不是主要问题，因为线程只是继续运行。

进一步的问题：我是否必须在runActions()中的所有位置插入Thread.currentThread().isInterrupted()，我预计这些位置将处于不可预测的长循环中？

Answer 1

这里我假设pageAnalyzerAction.runActions();可以被中断（即它通过相当快地退出来处理中断）。

如果您对低级线程API不满意，您可以使用java.current包中的执行器和期货来处理线程管理和超时策略：

• 执行器将使用线程池处理线程管理，必要时可重用它们
• 任务提交时返回的future可以用超时进行查询-如果任务没有在超时内完成，则future将抛出TimeOutException，然后您可以取消任务

一个人为的例子是：

//declare an executor  somewhere in your code, at a high level to recycle threads
ExecutorService executor = Executors.newFixedThreadPool(10); //number of threads: to be adjusted
private void runThread() throws InterruptedException {
    long timeout = 15000L;
    RunPageAnalyzer runPageAnalyzer = new RunPageAnalyzer(this);
    Future future = executor.submit(runPageAnalyzer);
    try {
        future.get(timeout, TimeUnit.MILLISECONDS);
    } catch (ExecutionException e) {
        //the runnable threw an exception: handle it
    } catch (TimeoutException e) {
        //the task could not complete before the timeout
        future.cancel(true); //interrrupt it
    }
}

Answer 2

听起来您的runActions-方法不会对正在设置的线程的中断状态做出反应。调用t.join()0之后的后一个join-调用没有超时，并且将无限期地等待线程t死亡。您应该在runActions-方法中检查中断状态，并在设置了中断状态（Thread.interrupted()返回true）的情况下终止操作。

Answer 3

这里的答案中没有提到其他内容。如果你想取消从线程完成的I/O，你不能"取消"它并期望实际的I/O被取消。基本上，您必须在"任务"中处理中断异常，并相应地处理它，甚至可能关闭套接字连接。我有一个小片段专门用于"停止"使用线程运行的任务，你可能会觉得这很有帮助（如果它有拼写错误，很抱歉，它是很久以前写的）。

public class ThreadStopTest {
    public static void main(String[] args) {
        testSqlThreadStop();
    }

    private static void testSocketReadStop() {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        SocketTask task = new SocketTask("http://www.yahoo.com", 80);
        Future<Integer> future = executor.submit(task);
        try {
            Integer result = future.get(1, TimeUnit.SECONDS);
            System.out.println("Computation complete; result: " + result);
        } catch(TimeoutException te) {
            future.cancel(true);
            task.cleanupAfterCancel();
            System.out.println("Computation cancelled");
        } catch(Exception e) {
            e.printStackTrace();
        }
        executor.shutdown();
    }
}

class SocketTask implements CleanableTask<Integer> {
    private final String host;
    private final int port;
    private Socket socket;
    public SocketTask(final String host, final int port) {
        this.host = host;
        this.port = port;
    }
    @Override
    public Integer call() throws Exception {
        InputStream in = null;
        // TODO: Actually update the count and cleanly handle exceptions
        int bytesRead = 0;
        try {
            this.socket = new Socket(this.host, this.port);
            in = this.socket.getInputStream();
            byte[] bytes = new byte[1000000];
            System.out.println("Started reading bytes");
            // The below behavior of waiting for a forceful close can be avoided
            // if we modify the FutureTask class (the default Future impl)
            // by passing in a CleanupHandler whose cleanup() method would be
            // invoked after invoking the `cancel` method or by making all 
            // your tasks implement a CancelledTask interface which has a 
            // `cleanupAfterCancel` method to do the same. :)
            try {
                in.read(bytes);
            } catch(SocketException se) {
                if(Thread.currentThread().isInterrupted()) {
                    System.out.println("All OK; this socket was forcefully closed");
                } else {
                    se.printStackTrace();   // something was seriously wrong
                }
            }
        } catch(Exception e) {
            e.printStackTrace();
        } finally {
            if(in != null)  in.close();
        }
        return Integer.valueOf(bytesRead);
    }
    @Override
    public void cleanupAfterCancel() {
        try {
            this.socket.close();
        } catch (IOException e) {
            e.printStackTrace();
        }        
    }
}