创建一个不退出的线程

Question

我想知道创建不终止的 Java 线程的最佳方法是什么。 目前，我基本上有一个“跑步者”，基本上看起来像：

ExecutorService pool = Executors.newFixedThreadPool(3);
for (int i = 0; i < numThreads; ++i) {
    pool.submit(new Task());
}
pool.shutdown();

Task 看起来像这样

public class Task {
    ...
    public void run() {
        while(true) { }
    }
}

我的方法有两个顾虑：

1. 我是否应该创建一个在完成工作后才返回的任务并继续生成执行最少工作量的线程？我担心开销，但不知道如何衡量它。

2. 如果我有一个无限循环的线程，当我强制退出可执行文件时，这些线程会被关闭并清理吗？经过一些测试，当包含ExecutorService的代码被强制关闭时，并没有出现InterruptException。

编辑： 详细地说，任务看起来像

public void run() {
    while(true) {
        // Let queue be a synchronized, global queue
        if (queue has an element) {
            // Pop from queue and do a very minimal amount of work on it
            // Involves a small amount of network IO (maybe 10-100 ms)
        } else {
            sleep(2000);
        }
    }
}

Answer 1

我同意@D Levant，阻塞队列是这里使用的关键。使用阻塞队列，您无需处理队列空或队列满的情况。

在你的任务类中，

while(true) {
        // Let queue be a synchronized, global queue
        if (queue has an element) {
            // Pop from queue and do a very minimal amount of work on it
            // Involves a small amount of network IO (maybe 10-100 ms)
        } else {
            sleep(2000);
        }
    }

这确实不是一个好方法，效率低下，因为您的 while 循环不断轮询，即使您已将线程 sleep()，但它仍然是每次线程唤醒和睡眠时不必要的上下文切换的开销.

在我看来，您使用 Executors 的方法看起来很适合您的情况。线程创建显然是一个代价高昂的过程，而 Executor 为我们提供了为不同任务重用同一个线程的灵活性。 您可以通过execute(Runnable) 或submit(Runnable/Callable) 传递您的任务，然后休息将由内部执行者负责。 Executors 内部只使用阻塞队列的概念。

您甚至可以通过使用ThreadPoolExecutor 类并在其构造函数中传递所需的参数来创建自己的线程池，在这里您可以传递自己的阻塞队列来保存任务。其余线程管理将根据构造函数中的配置传递由它负责，因此，如果您对配置参数非常有信心，那么您可以去做。

最后一点，如果您不想使用 Java 的内置 Executors 框架，那么您可以通过使用 BlockingQueue 来保存任务并启动一个线程来设计您的解决方案，该线程将从这个阻塞队列中获取任务到执行，下面是高级实现：

class TaskRunner {
   private int noOfThreads;  //The no of threads which you want to run always
   private boolean started;
   private int taskQueueSize; //No. of tasks that can be in queue at a time, when try to add more tasks, then you have to wait.
   private BlockingQueue<Runnable> taskQueue;
   private List<Worker> workerThreads;

   public TaskRunner(int noOfThreads, int taskQueueSize) {
      this.noOfThreads = noOfThreads;
      this.taskQueueSize = taskQueueSize;
   }

  //You can pass any type of task(provided they are implementing Runnable)
   public void submitTask(Runnable task) {
      if(!started) {
         init();
      }
      try {
         taskQueue.put(task);
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
   }

   public void shutdown() {
      for(Worker worker : workerThreads){
         worker.stopped = true;
      }
   }

   private void init() {
      this.taskQueue = new LinkedBlockingDeque<>(taskQueueSize);
      this.workerThreads = new ArrayList<>(noOfThreads);
      for(int i=0; i< noOfThreads; i++) {
         Worker worker = new Worker();
         workerThreads.add(worker);
         worker.start();
      }
   }

   private class Worker extends Thread {
      private volatile boolean stopped;
      public void run() {
         if(!stopped) {
            try {
               taskQueue.take().run();
            } catch (InterruptedException e) {
               e.printStackTrace();
            }
         }
      }
   }
}

class Task1 implements Runnable {
   @Override
   public void run() {
      //Your implementation for the task of type 1
   }
}

class Task2 implements Runnable {
   @Override
   public void run() {
      //Your implementation for the task of type 2
   }
}

class Main {

   public static void main(String[] args) {
      TaskRunner runner = new TaskRunner(3,5);
      runner.submitTask(new Task1());
      runner.submitTask(new Task2());
      runner.shutdown();
   }
}