如何在 Java 中暂停/恢复 ExecutorService 中的所有线程？

How to pause/resume all threads in an ExecutorService in Java?

提问by pathikrit

I submitted bunch of jobs to an executorservice in Java and I somehow want to temporarily pause all these jobs. What's the best way to do this? How can I resume? Or am I doing this completely wrong? Should I follow some other pattern for what I want to achieve (i.e. ability to pause/resume execution services)?

我向 Java 中的 executorservice 提交了一堆作业，我想暂时暂停所有这些作业。做到这一点的最佳方法是什么？我怎样才能恢复？还是我这样做完全错误？我是否应该遵循一些其他模式来实现我想要实现的目标（即暂停/恢复执行服务的能力）？

采纳答案by pathikrit

To answer my own question, I found an example of a PausableThreadPoolExecutorin the javadocs of ThreadPoolExecutoritself. Here is my version using Guava's Monitors:

要回答我的问题，我发现的一个例子PausableThreadPoolExecutor中的javadocThreadPoolExecutor本身。这是我使用 Guava's Monitors 的版本：

import com.google.common.util.concurrent.Monitor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;

public class PausableExecutor extends ScheduledThreadPoolExecutor {

    private boolean isPaused;

    private final Monitor monitor = new Monitor();
    private final Monitor.Guard paused = new Monitor.Guard(monitor) {
        @Override
        public boolean isSatisfied() {
            return isPaused;
        }
    };

    private final Monitor.Guard notPaused = new Monitor.Guard(monitor) {
        @Override
        public boolean isSatisfied() {
            return !isPaused;
        }
    };

    public PausableExecutor(int corePoolSize, ThreadFactory threadFactory) {
        super(corePoolSize, threadFactory);
    }

    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        monitor.enterWhenUninterruptibly(notPaused);
        try {
            monitor.waitForUninterruptibly(notPaused);
        } finally {
            monitor.leave();
        }
    }

    public void pause() {
        monitor.enterIf(notPaused);
        try {
            isPaused = true;
        } finally {
            monitor.leave();
        }
    }

    public void resume() {
        monitor.enterIf(paused);
        try {
            isPaused = false;
        } finally {
            monitor.leave();
        }
    }
}

回答by Peter Lawrey

The problem is that the Runnable/Callable themselves need to check when to pause/resume. That being said there and many ways to do this, and it depends on your requirements on how best to do this. Whatever your solution you need to make the waiting interruptable, so the thread can be shutdown cleanly.

问题是 Runnable/Callable 本身需要检查何时暂停/恢复。话虽如此，但有很多方法可以做到这一点，这取决于您对如何最好地做到这一点的要求。无论您需要什么解决方案来使等待中断，因此线程可以干净地关闭。

回答by Corin

I made some criticisms on your accepted answer, but they weren't very constructive... So here's my solution. I would use a class like this one and then call checkInwherever/whenever I want pause functionality. Find it on GitHub!

我对你接受的答案提出了一些批评，但它们并不是很有建设性......所以这是我的解决方案。我会使用这样的类，然后checkIn随时随地调用暂停功能。在GitHub 上找到它！

import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Provides a mechanism to pause multiple threads.
 * If wish your thread to participate, then it must regularly check in with an instance of this object.
 * 
 * @author Corin Lawson <[email protected]>
 */
public class Continue {
    private boolean isPaused;
    private ReentrantLock pauseLock = new ReentrantLock();
    private Condition unpaused = pauseLock.newCondition();

    public void checkIn() throws InterruptedException {
        if (isPaused) {
            pauseLock.lock();
            try {
                while (isPaused)
                    unpaused.await();
            } finally {
                pauseLock.unlock();
            }
        }
    }

    public void checkInUntil(Date deadline) throws InterruptedException {
        if (isPaused) {
            pauseLock.lock();
            try {
                while (isPaused)
                    unpaused.awaitUntil(deadline);
            } finally {
                pauseLock.unlock();
            }
        }
    }

    public void checkIn(long nanosTimeout) throws InterruptedException {
        if (isPaused) {
            pauseLock.lock();
            try {
                while (isPaused)
                    unpaused.awaitNanos(nanosTimeout);
            } finally {
                pauseLock.unlock();
            }
        }
    }

    public void checkIn(long time, TimeUnit unit) throws InterruptedException {
        if (isPaused) {
            pauseLock.lock();
            try {
                while (isPaused)
                    unpaused.await(time, unit);
            } finally {
                pauseLock.unlock();
            }
        }
    }

    public void checkInUninterruptibly() {
        if (isPaused) {
            pauseLock.lock();
            try {
                while (isPaused)
                    unpaused.awaitUninterruptibly();
            } finally {
                pauseLock.unlock();
            }
        }
    }

    public boolean isPaused() {
        return isPaused;
    }

    public void pause() {
        pauseLock.lock();
        try {
            isPaused = true;
        } finally {
            pauseLock.unlock();
        }
    }

    public void resume() {
        pauseLock.lock();
        try {
            if (isPaused) {
                isPaused = false;
                unpaused.signalAll();
            }
        } finally {
            pauseLock.unlock();
        }
    }
}

For example:

例如：

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;

public class PausableExecutor extends ScheduledThreadPoolExecutor {
    private Continue cont;

    public PausableExecutor(int corePoolSize, ThreadFactory threadFactory, Continue c) {
        super(corePoolSize, threadFactory);
        cont = c;
    }

    protected void beforeExecute(Thread t, Runnable r) {
        cont.checkIn();
        super.beforeExecute(t, r);
    }
}

This has the added benefit that you can pause many threads with a single call to Continue's pause.

这有一个额外的好处，您可以通过一次调用Continue's来暂停多个线程pause。

回答by marcinj

I was looking for pause/resume functionality in executor, but with additional ability to await for any currently being processed tasks. Below is variant of other great implementations from this SO with addition of await functions. I was testing it on executor with single thread. So basic usage is:

我在 executor 中寻找暂停/恢复功能，但具有等待任何当前正在处理的任务的额外能力。以下是此 SO 中其他出色实现的变体，并添加了 await 函数。我正在用单线程在执行器上测试它。所以基本用法是：

executor.pause();
executor.await(10000); // blocks till current tasks processing ends

class code:

班级代码：

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class PausableScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {      
  public boolean isPaused;
  private ReentrantLock pauseLock = new ReentrantLock();
  private Condition unpaused = pauseLock.newCondition();
  private Latch activeTasksLatch = new Latch();

  private class Latch {
    private final Object synchObj = new Object();
    private int count;

    public boolean awaitZero(long waitMS) throws InterruptedException {
      long startTime = System.currentTimeMillis();
      synchronized (synchObj) {
        while (count > 0) {
          if ( waitMS != 0) {
            synchObj.wait(waitMS);
            long curTime = System.currentTimeMillis();
            if ( (curTime - startTime) > waitMS ) {                
              return count <= 0;
            }
          }
          else
            synchObj.wait();
        }
        return count <= 0; 
      }
    }
    public void countDown() {
      synchronized (synchObj) {
        if (--count <= 0) {
          // assert count >= 0;              
          synchObj.notifyAll();
        }
      }
    }
    public void countUp() {
      synchronized (synchObj) {
        count++;
      }
    }    
  }

  /**
   * Default constructor for a simple fixed threadpool
   */
  public PausableScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize);
  }

  /**
   * Executed before a task is assigned to a thread.
   */
  @Override
  protected void beforeExecute(Thread t, Runnable r) {
    pauseLock.lock();
    try {
      while (isPaused)
        unpaused.await();
    } catch (InterruptedException ie) {
      t.interrupt();
    } finally {
      pauseLock.unlock();
    }

    activeTasksLatch.countUp();
    super.beforeExecute(t, r);
  }

  @Override
  protected void afterExecute(Runnable r, Throwable t) {
    try {
      super.afterExecute(r, t);
    }
    finally {
      activeTasksLatch.countDown();
    }
  }

  /**
   * Pause the threadpool. Running tasks will continue running, but new tasks
   * will not start untill the threadpool is resumed.
   */
  public void pause() {
    pauseLock.lock();
    try {
      isPaused = true;
    } finally {
      pauseLock.unlock();
    }
  }

  /**
   * Wait for all active tasks to end.
   */ 
  public boolean await(long timeoutMS) {
    // assert isPaused;
    try {
      return activeTasksLatch.awaitZero(timeoutMS);
    } catch (InterruptedException e) {
      // log e, or rethrow maybe
    }
    return false;
  }

  /**
   * Resume the threadpool.
   */
  public void resume() {
    pauseLock.lock();
    try {
      isPaused = false;
      unpaused.signalAll();
    } finally {
      pauseLock.unlock();
    }
  }

}

回答by CallSign-Filter

I know this is old, but I tried all of these answers and none of them worked for what I was trying to do with a pausable timer; they all would throw out all of the data it would have been doing on a schedule as soon as it resumed (all at once).

我知道这很旧，但是我尝试了所有这些答案，但没有一个对我尝试使用可暂停计时器所做的工作有效；一旦它恢复（一次全部），它们都会丢弃它本应按计划执行的所有数据。

Instead I found this Timerclass on GitHub* here. This worked really well for me.

相反，我发现这个Timer在GitHub上类*这里。这对我来说非常有效。

*I did not write this code, just found it.

*这段代码不是我写的，只是找到了。