我正在尝试使用JAVA线程池,并且想知道是否仍然可以在没有更多线程提交时关闭线程池,并在一个线程完成执行时提交线程。
更明确地说,我有一个THREAD_LIMIT
类型的Integer
变量,该变量是最大线程数不能执行并行和递归算法,该算法假定在调用自身之前检查活动线程数。如果活动的线程数少于线程限制,它将向线程池提交新线程,否则在同一线程上调用递归。
我面临的问题是跟踪活动线程并在没有新线程提交时关闭线程池。我想使用多线程从代码中获得最大性能。
我按照this教程创建自己的线程池并使用
public int getTaskQueueSize() {
return taskQueue.size();
}
在
ThreadPool
类中激活线程数。在上课时我正在使用
void shutdownExecutor() {
while (tp.getTaskQueueSize() != 0) {
try {
Thread.sleep(100);
} catch (InterruptedException ex) {
Logger.getLogger(HamiltonPaths.class.getName()).log(Level.SEVERE, null, ex);
}
// System.out.println("Not STopping tp");
}
tp.shutdown();
// System.out.println("Stopped tp");
}
在主类中关闭线程池。但是过了一段时间,它将停止产生新线程。但是一个线程继续递归地进行工作。
更新1:添加代码
所以我发现将任务提交到线程池工作正常。但是我不小心添加了一个代码最近更改的错误,这阻止了我向线程池提交更多任务,并由于
shutdownExecutor()
返回查询的初始大小或未删除任务而从tp.getTaskQueueSize()
函数关闭线程池从que。我正在使用以下逻辑来决定是提交新任务还是继续递归工作。
if ((this.tp.getTaskQueueSize() < threadLimit) && (!this.tp.isPoolShutDownInitiated())) {
spawnNewThread(new PTask(cp, get));//Class that implements the Runnable and do the same thing as the function called in below statement.
} else {
PPath(cp, get);// call to the function
}
BlockingQueueCustom.java
package threadpool;
abstract interface BlockingQueueCustom<E>
{
public abstract void put(E paramE)
throws InterruptedException;
public abstract E take()
throws InterruptedException;
public abstract int size();
}
LinkedBlockingQueueCustom.java
package threadpool;
import java.util.LinkedList;
import java.util.List;
class LinkedBlockingQueueCustom<E>
implements BlockingQueueCustom<E> {
private List<E> queue;
private int maxSize;
public LinkedBlockingQueueCustom(int maxSize) {
this.maxSize = maxSize;
this.queue = new LinkedList();
}
public synchronized void put(E item)
throws InterruptedException {
if (this.queue.size() == this.maxSize) {
wait();
}
this.queue.add(item);
notifyAll();
}
public synchronized E take()
throws InterruptedException {
if (this.queue.isEmpty()) {
wait();
}
notifyAll();
if (this.queue.isEmpty()) {
return null;
}
return (E) this.queue.remove(0);
}
public synchronized int size() {
return this.queue.size();
}
}
线程池
package threadpool;
import java.util.logging.Level;
import java.util.logging.Logger;
public class ThreadPool {
private BlockingQueueCustom<Runnable> taskQueue;
int size = 0;
int taskExecuted = 0;
ThreadPoolsThread[] threadPoolsThread;
public int getTaskExecuted() {
return this.taskExecuted;
}
public synchronized void dectaskExec() {
this.taskExecuted -= 1;
}
public int getSize() {
return this.size;
}
public BlockingQueueCustom<Runnable> getTaskQueue() {
return this.taskQueue;
}
public int getTaskQueueSize() {
return this.taskQueue.size();
}
private boolean poolShutDownInitiated = false;
public ThreadPool(int nThreads) {
this.taskQueue = new LinkedBlockingQueueCustom(nThreads);
this.size = nThreads;
this.threadPoolsThread = new ThreadPoolsThread[nThreads + 1];
for (int i = 1; i <= nThreads; i++) {
this.threadPoolsThread[i] = new ThreadPoolsThread(this.taskQueue, this);
this.threadPoolsThread[i].setName("" + i);
this.threadPoolsThread[i].start();
}
}
public synchronized void execute(Runnable task) {
if (this.poolShutDownInitiated) {
try {
throw new Exception("ThreadPool has been shutDown, no further tasks can be added");
} catch (Exception ex) {
Logger.getLogger(ThreadPool.class.getName()).log(Level.SEVERE, null, ex);
}
}
this.taskExecuted += 1;
try {
this.taskQueue.put(task);
} catch (InterruptedException ex) {
Logger.getLogger(ThreadPool.class.getName()).log(Level.SEVERE, null, ex);
}
}
public boolean isPoolShutDownInitiated() {
return this.poolShutDownInitiated;
}
public synchronized void shutdown() {
this.poolShutDownInitiated = true;
}
}
ThreadPoolsThread.java
package threadpool;
import java.util.logging.Level;
import java.util.logging.Logger;
class ThreadPoolsThread
extends Thread {
private BlockingQueueCustom<Runnable> taskQueue;
private ThreadPool threadPool;
public ThreadPoolsThread(BlockingQueueCustom<Runnable> queue, ThreadPool threadPool) {
this.taskQueue = queue;
this.threadPool = threadPool;
}
public void run() {
for (;;) {
Runnable runnable = null;
while ((!this.threadPool.isPoolShutDownInitiated()) && (this.taskQueue.size() == 0)) {
}
if ((this.threadPool.isPoolShutDownInitiated()) && (this.taskQueue.size() == 0)) {
break;
}
try {
runnable = (Runnable) this.taskQueue.take();
} catch (InterruptedException ex) {
Logger.getLogger(ThreadPoolsThread.class.getName()).log(Level.SEVERE, null, ex);
break;
}
if (runnable == null) {
break;
}
runnable.run();
if ((this.threadPool.isPoolShutDownInitiated())
&& (this.taskQueue.size() == 0)) {
interrupt();
try {
Thread.sleep(1L);
} catch (InterruptedException ex) {
Logger.getLogger(ThreadPoolsThread.class.getName()).log(Level.SEVERE, null, ex);
break;
}
}
}
}
}
最佳答案
似乎存在根本的误解。您不会“将新线程提交到线程池”,而是会将任务提交到执行程序服务(可能作为线程池实现)。执行程序服务管理线程并决定是启动,停止还是重用它们。如果将执行程序配置为具有最大线程数,它将使用不超过指定数量的线程。如果提交的任务多于可用线程,则这些任务将排队。因此,队列的大小不是线程数。
通常,您不应该实现自己的线程池,尤其是当您不完全了解它们的作用时。 Java内置的实现已有十多年了。您可以使用Executors.newFixedThreadPool(threadCount)
使用指定数量的线程来获取线程池执行程序。
如果您要强制执行任务不在队列中,而是在所有线程繁忙时始终由调用者执行,则实现起来并非难事。使用the constructor allowing to specify a custom queue and ThreadPoolExecutor
以及RejectedExecutionHandler
和SynchronousQueue
显式创建CallerRunsPolicy
:
ThreadPoolExecutor e=new ThreadPoolExecutor(threadCount, threadCount,
1, TimeUnit.MINUTES, new SynchronousQueue<>(), Executors.defaultThreadFactory(),
new ThreadPoolExecutor.CallerRunsPolicy());
在Java 8之前,您必须明确指定队列类型:
new SynchronousQueue<Runnable>()
这会将已提交的任务传递给空闲线程(如果有),否则将在提交线程中执行它。请注意,如上例所示,当核心大小和最大大小相同时,超时值无关紧要。
如果已提交所有任务,则可以简单地调用
shutDown()
而不进行其他检查。在停止所有线程之前,它仍将执行挂起的任务。