java - 中断时间超过阈值的线程的最佳实践

Question

我正在使用 Java ExecutorService 框架来提交可调用任务以供执行。这些任务与 Web 服务通信，并应用 5 分钟的 Web 服务超时。但是我已经看到，在某些情况下，超时被忽略并且线程在 API 调用时“挂起”——因此，我想取消所有需要超过 5 分钟的任务。

目前，我有一个 future 列表，我遍历它们并调用 future.get 直到所有任务完成。现在，我已经看到 future.get 重载方法需要超时并在任务未在该窗口中完成时抛出超时。所以我想到了一种方法，我在超时时执行 future.get() 并且在 TimeoutException 的情况下我执行 future.cancel(true) 以确保此任务被中断。

我的主要问题
1. get with a timeout 是解决这个问题的最好方法吗？
2. 是否有可能我正在等待尚未放置在线程池中的任务的 get 调用(不是活跃的工作人员)。在那种情况下，我可能会终止一个线程，当它启动时实际上可能会在要求的时间限制内完成？

如有任何建议，我们将不胜感激。

Answer 1

1. 超时获取是解决此问题的最佳方法吗？

   • 这还不够。例如，如果您的任务并非旨在响应中断，它将继续运行或只是被阻止

2. 是否有可能我正在等待尚未放置在线程池中的任务的 get 调用(不是 Activity 的工作程序)。在那种情况下，我可能会终止一个线程，当它启动时实际上可能会在要求的时间限制内完成？

   • 是的，如果您的线程池配置不当，您可能会因为从未计划运行的任务而最终取消

当您的任务包含不可中断的阻塞时，以下代码片段可能是使您的任务响应中断的方法之一。它也不会取消未计划运行的任务。这里的想法是通过关闭套接字、数据库连接等来覆盖中断方法并关闭正在运行的任务。此代码并不完美，您需要根据要求进行更改、处理异常等。

class LongRunningTask extends Thread {
private  Socket socket; 
private volatile AtomicBoolean atomicBoolean;


public LongRunningTask() {
    atomicBoolean = new AtomicBoolean(false);
}

@Override
public void interrupt() {
    try {
        //clean up any resources, close connections etc.
        socket.close();
    } catch(Throwable e) {
    } finally {
        atomicBoolean.compareAndSet(true, false);
        //set the interupt status of executing thread.
        super.interrupt();
    }
}

public boolean isRunning() {
    return atomicBoolean.get();
}

@Override
public void run() {
    atomicBoolean.compareAndSet(false, true);
    //any long running task that might hang..for instance
    try {
        socket  = new Socket("0.0.0.0", 5000);
        socket.getInputStream().read();
    } catch (UnknownHostException e) {
    } catch (IOException e) {
    } finally {

    }
}
}
//your task caller thread
//map of futures and tasks 
    Map<Future, LongRunningTask> map = new HashMap<Future, LongRunningTask>();
    ArrayList<Future> list = new ArrayList<Future>();
    int noOfSubmittedTasks = 0;

    for(int i = 0; i < 6; i++) {
        LongRunningTask task = new LongRunningTask();
        Future f = execService.submit(task);
        map.put(f, task);
        list.add(f);
        noOfSubmittedTasks++;
    }

    while(noOfSubmittedTasks > 0) {
        for(int i=0;i < list.size();i++) {
            Future f = list.get(i);
            LongRunningTask task = map.get(f);
            if (task.isRunning()) {
                /*
                 * This ensures that you process only those tasks which are run once
                 */
                try {
                    f.get(5, TimeUnit.MINUTES);
                    noOfSubmittedTasks--;
                } catch (InterruptedException e) {
                } catch (ExecutionException e) {
                } catch (TimeoutException e) {
                                            //this will call the overridden interrupt method
                    f.cancel(true);
                    noOfSubmittedTasks--;
                }
            }

        }
    }
    execService.shutdown();