我是并发编码新手，并尝试自己实现简单的
ThreadPool但是，我认为
threadisStoppedvolatilePoolThreadRunnableAtomicisStopped=truethis.thread.interrupt()this.thread=Thread.currentThread()while(!isStopped())我的理解正确还是我遗漏了什么？
public class ThreadPool {
private BlockingQueue taskQueue = null;
private List<PoolThreadRunnable> runnables = new ArrayList<>();
private boolean isStopped = false;
public ThreadPool(int noOfThreads, int maxNoOfTasks){
taskQueue = new ArrayBlockingQueue(maxNoOfTasks);
for(int i=0; i<noOfThreads; i++){
PoolThreadRunnable poolThreadRunnable =
new PoolThreadRunnable(taskQueue);
runnables.add(poolThreadRunnable);
}
for(PoolThreadRunnable runnable : runnables){
new Thread(runnable).start();
}
}
public synchronized void execute(Runnable task) throws Exception{
if(this.isStopped) throw
new IllegalStateException("ThreadPool is stopped");
this.taskQueue.offer(task);
}
public synchronized void stop(){
this.isStopped = true;
for(PoolThreadRunnable runnable : runnables){
runnable.doStop();
}
}
public synchronized void waitUntilAllTasksFinished() {
while(this.taskQueue.size() > 0) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class PoolThreadRunnable implements Runnable {
private Thread thread = null;
private BlockingQueue taskQueue = null;
private boolean isStopped = false;
public PoolThreadRunnable(BlockingQueue queue){
taskQueue = queue;
}
public void run(){
this.thread = Thread.currentThread();
while(!isStopped()){
try{
Runnable runnable = (Runnable) taskQueue.take();
runnable.run();
} catch(Exception e){
//log or otherwise report exception,
//but keep pool thread alive.
}
}
}
public synchronized void doStop(){
isStopped = true;
//break pool thread out of dequeue() call.
this.thread.interrupt();
}
public synchronized boolean isStopped(){
return isStopped;
}
}
这两个
isStoppedwhilerun()isStopped()threaddoStop()根据类的其余部分的实现方式，以下内容将是一个很好的解决方案：
public void run(){
synchronized (this) {
if (isStopped) {
return;
}
this.thread = Thread.currentThread();
}
while(!isStopped()){
try{
Runnable runnable = (Runnable) taskQueue.take();
runnable.run();
} catch(Exception e){
//log or otherwise report exception,
//but keep pool thread alive.
}
}
}
为什么要在
thisthisNullPointerExceptiondoStop()public synchronized void doStop(){
isStopped = true;
if (thread != null) {
//break pool thread out of dequeue() call.
thread.interrupt();
}
}
我还有另一个问题（尽管无关紧要），即执行 Callable 会发生哪些变化。如果您知道任何参考资料，请给我指点，我将不胜感激。
我不知道有什么入门教程。但
CallableRunnableFutureThreadPoolExecutor和
FutureTask的实现，了解它们是如何实现的。
因此，使用 Java Executor Framework 获得灵感，首先您应该创建一个类似于
Futureimport java.util.concurrent.ExecutionException;
public interface Result<V> {
/**
* Gets the value of this {@code Result}, blocking until the background task
* is done if necessary.
*
* @return the value of the background task
* @throws ExecutionException if the background task throws an exception; the
* thrown exception will be the cause of the {@code ExecutionException}
* @throws InterruptedException if the calling thread is interrupted
*/
V get() throws ExecutionException, InterruptedException;
/**
* Tests if the background task is complete.
*
* @return {@code true} if the background task is complete, {@code false}
* otherwise
*/
boolean isDone();
}
然后您应该创建一个实现 also 实现
RunnableCallableimport java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
public class ResultTask<V> implements Result<V>, Runnable {
private final AtomicBoolean executed = new AtomicBoolean();
private final Lock lock = new ReentrantLock();
private final Condition isDone = lock.newCondition();
private final Callable<V> callable;
// these three fields are guarded by 'lock'
private boolean done;
private V result;
private Throwable error;
public ResultTask(Callable<V> callable) {
this.callable = Objects.requireNonNull(callable);
}
@Override
public V get() throws ExecutionException, InterruptedException {
lock.lockInterruptibly();
try {
while (!done) {
// Called in a loop to handle a so-called "spurious wakeup" of the thread
isDone.await();
}
if (error != null) {
throw new ExecutionException(error);
}
return result;
} finally {
lock.unlock();
}
}
@Override
public boolean isDone() {
lock.lock();
try {
return done;
} finally {
lock.unlock();
}
}
@Override
public void run() {
if (!executed.compareAndSet(false, true)) {
throw new IllegalStateException("task already executed");
}
try {
// don't hold the lock while invoking the Callable
V result = callable.call();
complete(result, null);
} catch (Throwable error) {
complete(null, error);
}
}
private void complete(V result, Throwable error) {
lock.lock();
try {
this.result = result;
this.error = error;
done = true;
// Signal all threads blocked on 'isDone.await()' to wake them up
isDone.signalAll();
} finally {
lock.unlock();
}
}
}
然后你的
ThreadPoolCallableResultTaskResultTaskRunnableResultTaskResultpublic <V> Result<V> execute(Callable<V> callable) throws Exception {
Objects.requireNonNull(callable);
synchronized (this) {
if (isStopped)
throw new IllegalStateException("ThreadPool is stopped");
ResultTask<V> task = new ResultTask<>(callable);
taskQueue.offer(task);
return task;
}
}
警告： 可能应该检查
offer您甚至可以将
execute(Runnable)public Result<Void> execute(Runnable runnable) throws Exception {
Objects.requireNonNull(runnable);
return execute(() -> {
runnable.run();
return null;
});
}
这会让调用者知道
Runnable顺便说一句，如果
BlockingQueuejava.util.concurrent.BlockingQueue该界面是通用的。您尚未对其进行参数化，这意味着您正在使用原始类型。不要使用原始类型；使用
BlockingQueue<Runnable>new ArrayBlockingQueue<>(...)<>RunnablePoolThreadRunnable#run()offer(E)方法将返回
falseexecuteofferfalseJava Executor Framework中，它将是一个
RejectedExecutionException。