ExecutorService 的未来任务没有真正取消

Posted 2023-02-19

【中文标题】ExecutorService 的未来任务没有真正取消

【英文标题】：Future task of ExecutorService not truly cancelling

【发布时间】：2012-06-24 21:15:23

【问题描述】：

我将我的 Futures 从 ExecutorService 推送到哈希映射中。稍后，我可能会从哈希映射中调用 Futures 的取消。虽然结果是真的，但我后来在 Callable 过程中打断点，好像 Future cancel() 没有任何效果。我认为这可能是两个不同引用的情况（即使在断点时引用 ID 被列为相同），但想知道是否有专家可以插话。代码如下所示：

ExecutorService taskExecutor = Executors.newCachedThreadPool();
Map <String, Future<Object>> results = new HashMap <String, Future<Object>>();      

Future<Object> future = taskExecutor.submit(new MyProcessor(uid));
results.put(uid, future);

我允许继续处理（这是一个在任务传入时提交任务的循环），稍后我可能会尝试通过调用此方法从外部源取消：

public static synchronized boolean cancelThread(String uid) 
    Future<Object> future = results.get(uid);
    boolean success = false;
    if (future != null) 
        success = (future.isDone() ? true : future.cancel(true));
        if (success)
            results.remove(uid);
    
    return success;     

但在调用 future.cancel() 之后，我仍然在 MyProcessor.call() 中遇到“未取消”路径 - 即它并没有真正被取消。

我在哪里做错了？有更好的办法吗？

【参考方案1】：

我后来在 Callable 过程中打断点，好像 Future cancel() 没有效果。

Future.cancel(true) 删除队列中尚未运行的作业，但如果该作业已在运行，则它在运行该作业的线程上执行与 Thread.interrupt() 等效的操作。这会在线程上设置中断位并导致任何sleep()、wait() 和其他一些方法抛出InterruptedException。

重要的是要意识到它确实不会停止线程。您需要主动检查线程循环中的中断标志或正确处理InterruptedException。

请在此处查看我的 SO 答案以获取更多详细信息：How to suspend thread using thread's id?

【讨论】：

还相关于：***.com/questions/6698977/…

我明白了，这是有道理的——此时我在 Callable 中没有处于 wait() 状态，所以它不会为我抛出 InterruptedException。不幸的是，我试图取消的是一个已经启动的数据库的单个语句调用，所以我充其量可以测试线程之后是否被中断。

澄清一下，这是我在请求取消时可能使用的语句，数据库中的 PreparedStatement：'stmt.execute();'所以我想我必须让数据库完成课程，然后检查中断。

好的，所以我将其更改为可运行线程而不是可调用线程，这样我就可以覆盖 interrupt() 方法并在其中使用 stmt.cancel() 取消我的数据库语句；但是，看起来 Future.cancel(true) 可能没有明确调用此中断方法，因为我从未参与其中。以为我有什么东西在那里。

问题是@Ryan 是您正在向线程池提交Runnable。它有自己的线程，所以它只使用你的 Thread.run() 方法。如果您需要覆盖interrupt()，那么您需要创建自己的线程而不是使用池。

【参考方案2】：

FutureTask :: boolean cancel(boolean mayInterruptIfRunning) 将在当前运行的线程上执行interrupt。

FutureTask.java
public boolean cancel(boolean mayInterruptIfRunning) 
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
    try     // in case call to interrupt throws exception
        if (mayInterruptIfRunning) 
            try 
                Thread t = runner;
                if (t != null)
                    t.interrupt();     ////////////HERE/////////////
             finally  // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            
        
     finally 
        finishCompletion();
    
    return true;

JavaDoc 说下面的 interrupt

公共无效中断（） 中断这个线程。除非当前 线程正在中断自己，这总是被允许的， 调用了该线程的 checkAccess 方法，这可能会导致 要抛出的 SecurityException。

如果该线程在调用 wait()、wait(long) 时被阻塞， 或 Object 类或 join() 的 wait(long, int) 方法， join(long)、join(long, int)、sleep(long) 或 sleep(long, int)、方法 这个类，那么它的中断状态将被清除，它会 收到一个 InterruptedException。

如果此线程在可中断的 I/O 操作中被阻塞 通道然后通道将被关闭，线程的中断状态 将被设置，并且线程将收到 ClosedByInterruptException。

如果该线程在 Selector 中被阻塞，则该线程的中断 状态将被设置，它将立即从选择中返回 操作，可能具有非零值，就像选择器的 唤醒方法被调用。

如果前面的条件都不成立，那么这个线程的中断 状态将被设置。

中断一个不活跃的线程不需要有任何效果。

抛出：SecurityException - 如果当前线程无法修改它 线程

总结； FutureTask 的取消仅在线程被阻塞时产生影响（在调用 wait() 时，...）否则开发人员有责任检查 Thread.currentThread().isInterrupted() 以退出；同时执行非阻塞操作。