在保存状态时使用 ThreadPool 进行多线程答案

【问题标题】：Multi Threading with ThreadPool while saving state在保存状态时使用 ThreadPool 进行多线程
【发布时间】：2017-10-24 16:56:35
【问题描述】：

假设队列中有 100 个可运行对象。如果我使用某个固定大小的 ExecutorServicePool 比如说 10，那么至少需要终止 10 个线程中的一个，以便从队列中执行下一个 Runnable 对象。有没有其他方法可以在特定时间间隔后从队列中安排下一个 Runnable 并将当前的 Runnable 状态保存在队列末尾，有点像循环方式？

【问题讨论】：

你真正的目标是什么？可能有比您描述的情况更好的方法。
你的意思是如果池中的线程还在运行，你想在一定时间后让出线程吗？
线程不需要被终止（也不是）。其中一个runnables需要完成。你想完成什么？
我的主要目标是运行队列中的所有这 100 个可运行对象，每个对象都运行固定时间，即使它们直到完成才运行。但是我需要再次将可运行对象的当前状态添加到队列中，以便从之前停止的位置开始执行。
那么你需要拆分runnables并重新提交新的，同时让前一个完成。

标签： java multithreading executorservice threadpoolexecutor scheduledexecutorservice

【解决方案1】：

有没有其他方法可以在特定时间间隔后从队列中安排下一个 Runnable 并保存当前的 Runnable 状态以保存在队列末尾，有点像循环方式？

我希望我能理解这里的问题。如果这不是您所说的，请编辑您的问题并提供更多详细信息。

如果您在谈论保存状态，简单的解决方案是使用ThreadLocal，以便执行器服务中运行的每个线程都可以保存自己的状态。

private final ThreadLocal<State> stateThreadLocal = new ThreadLocal<>() {
   // state initialization if needed
   protected State initialValue() { return new State(); }
};
...
public void run() {
   State state = stateThreadLocal.get();
   // now you can process the job with the state
}

但是，如果您需要关闭 http-client 或其他东西，此解决方案不会让您有机会控制何时释放状态。

更好的解决方案可能是使用 executor-service 来启动一个静态工作人员列表，然后使用您自己的 BlockingQueue 将工作注入工作人员。

例如，您可以执行以下操作：

private final static int NUM_WORKERS = 10;
private final ExecutorService threadPool = Executors.newFixedThreadPool(NUM_WORKERS);
...

final BlockingQueue<Job> queue = new LinkedBlockingQueue<>();
for (int i = 0; i < NUM_WORKERS; i++) {
    threadPool.submit(new WorkerWithState(queue));
}
// shutdown the queue once the last worker is submitted
threadPool.shutdown();
 ...
// then you submit a number of jobs to your own queue for the workers with state to run
queue.add(new Job(...));
...

您的工作人员将持有他们将在运行之间持有的本地状态字段。

private class WorkerWithState implements Runnable {
   // some state held by the worker
   private SomeState state = new SomeState();
   private final BlockingQueue queue;
   public WorkerWithState(BlockingQueue queue) {
      this.queue = queue;
   }
   public void run() {
      while (!Thread.currentThread().isInterrupted()) {
         try {
            // wait for a job to process
            Job job = queue.take();
            // process the job here using the state
            processJob(job, state);
         } catch (InterruptedException ie) {
            // always a good pattern
            Thread.currentThread().interrupt();
            return;
         }
      }
   }
}

要终止这些工作线程，您可以在线程池上调用threadPool.shutdownAll(true) 来中断它们，或者注入一个常量QUIT_JOB 让它们在作业队列耗尽后自行停止。

【讨论】：