停止执行java线程和join()的作用

Question

我的程序分析了大量文档，偶尔会得到一个导致无限循环或很长循环的页面。这是无法提前分析的。我想杀死特定页面并继续下一个（丢弃违规页面的任何结果）。我已阅读 SO 答案 such as this How to stop execution after a certain time in 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

但是当超过时间限制时，进程会挂在t.join()。

    =========== 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.concurrent 包中的 executor 和 futures 来处理线程管理和超时策略：

• 执行器将使用线程池处理线程管理，必要时重复使用它们
• 可以通过超时查询任务提交时返回的未来 - 如果任务未在超时内完成，未来将抛出 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-方法对正在设置的线程的中断状态没有反应。后者join-调用interrupt之后的调用没有超时，会无限期地等待线程t死掉。您应该检查 runActions 方法中的中断状态，如果设置了中断状态（Thread.interrupted() 返回 true），则通过终止操作来做出反应。

Answer 3

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

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();
        }        
    }

}