线程池实现代码中的疑惑

Question

在花费大量时间研究线程池概念并阅读大量博客上的不同代码并在 Stackoverflow.com 上发布问题之后，现在我对这个概念有了清晰的认识。但与此同时，我在代码中发现了一些疑点。

1. 当 pool.assign(new TestWorkerThread()); 在 TestThreadPool 类中执行时，它调用 done.workerBegin(); 方法，在 Done 类 中，它增加 _activeThreads 变量。但我认为，LOGICALLY 这是不正确的，因为如果线程数（在本例中为 2）少于任务数（在 TestThreadPool 类 中给出）（在在这种情况下 5），它会不必要地增加 _activeThreads（即 _activeThreads = 5）计数。

2. _started 变量在 Done 类 中有什么作用？

3. waitDone() 和 waitBegin()（在 Done Class 中）如何执行它们的功能？ （这两种方法一步一步解释就好）

代码如下。我正在按照流程安排代码。

TestThreadPool 类：-

package hitesh;

/**
 *
 * @author jhamb
 */

public class TestThreadPool {

 public static void main(String args[]) throws InterruptedException
 {
  ThreadPool pool = new ThreadPool(2);

  for (int i = 1;i <= 5;i++) {
   pool.assign(new TestWorkerThread());
  }
  System.out.println("All tasks are assigned");

  pool.complete();

  System.out.println("All tasks are done.");
 }
}

TestWorkerThread 类：-

package hitesh;

/**
 *
 * @author jhamb
 */
/**
 * This class shows an example worker thread that can
 * be used with the thread pool. It demonstrates the main
 * points that should be included in any worker thread. Use
 * this as a starting point for your own threads.
 */

public class TestWorkerThread implements Runnable {
 static private int count = 0;
 private int taskNumber;
 protected Done done;

 /**
  * 
  * @param done
  */
 TestWorkerThread()
 {
  count++;
  taskNumber = count;
  //System.out.println("tasknumber  --->  " + taskNumber);
 }

 public void run()
 {
  System.out.println("TWT run starts   -->  "  + this.toString());
  for (int i=0;i <= 100;i += 25) {
   System.out.println("Task number: " + taskNumber + 
             ",percent complete = " + i );
   try {
    Thread.sleep((int)(Math.random()*500));
   } catch (InterruptedException e) {
   }
  }
  System.out.println("task for thread --> " + this.toString() + "   completed");
 }
}

线程池类：-

package hitesh;

/**
 *
 * @author jhamb
 */
import java.util.*;


/* 
 * This is the main class for the thread pool. You should
 * create an instance of this class and assign tasks to it.
 */

public class ThreadPool {

 protected Thread threads[] = null;

 Collection assignments = new ArrayList(3);

 protected Done done = new Done();


 public ThreadPool(int size) throws InterruptedException
 {
   threads = new WorkerThread[size];
   for (int i=0;i<threads.length;i++) {
    threads[i] = new WorkerThread(this);
    threads[i].start();
    System.out.println ("thread " + i + " started");
    threads[i].sleep(1000);
  }

 }

 public synchronized void assign(Runnable r)
 {
  done.workerBegin();
  assignments.add(r);
  System.out.println("Collection size --->   " + assignments.size() +   "  Thread can work on this");
  notify();
 }

 public synchronized Runnable getAssignment()
 {
  try {
   while ( !assignments.iterator().hasNext() )
    wait();

   Runnable r = (Runnable)assignments.iterator().next();
   assignments.remove(r);
   return r;
  } catch (InterruptedException e) {
   done.workerEnd();
   return null;
  }
 }

 public void complete()
 {   
  done.waitBegin();
  done.waitDone();
 }

}

WorkerThread 类：-

package hitesh;
import java.util.*;
/**
 *
 * @author jhamb
 */

/**
 * The worker threads that make up the thread pool.
 */
class WorkerThread extends Thread {
 /**
  * True if this thread is currently processing.
  */
 public boolean busy;
 /**
  * The thread pool that this object belongs to.
  */
 public ThreadPool owner;

 /**
  * The constructor.
  * 
  * @param o the thread pool 
  */
 WorkerThread(ThreadPool o)
 {
  owner = o;
 }

 /**
  * Scan for and execute tasks.
  */
    //@Override
 public void run()
 {
  System.out.println("Threads name : "+ this.getName() + "  working.....");
  Runnable target = null;

  do {
   System.out.println("enter in do while " + this.getName() );
   target = owner.getAssignment();
   System.out.println("GetAssignment k aage aa gya mai "  +  target);
   if (target!=null) {
    target.run();
    //target.
    owner.done.workerEnd();
   }
  } while (target!=null);
  System.out.println("do while finishes for "+ this.getName());
 }
}

完成课程：-

package hitesh;

/**
 *
 * @author jhamb
 */
/**
 * 
 * This is a thread pool for Java, it is
 * simple to use and gets the job done. This program and
 * all supporting files are distributed under the Limited
 * GNU Public License (LGPL, http://www.gnu.org).
 * 
 * This is a very simple object that
 * allows the TheadPool to determine when 
 * it is done. This object implements
 * a simple lock that the ThreadPool class
 * can wait on to determine completion.
 * Done is defined as the ThreadPool having
 * no more work to complete.
 * 
 * Copyright 2001 by Jeff Heaton
 *
 * @author Jeff Heaton (http://www.jeffheaton.com)
 * @version 1.0
 */
public class Done {

 /**
  * The number of Worker object
  * threads that are currently working
  * on something.
  */
 private int _activeThreads = 0;

 /**
  * This boolean keeps track of if
  * the very first thread has started
  * or not. This prevents this object
  * from falsely reporting that the ThreadPool 
  * is done, just because the first thread
  * has not yet started.
  */
 private boolean _started = false;
 /**
  * This method can be called to block
  * the current thread until the ThreadPool
  * is done.
  */

 synchronized public void waitDone()
 {
  try {
   while ( _activeThreads>0 ) {
    wait();
   }
  } catch ( InterruptedException e ) {
  }
 }
 /**
  * Called to wait for the first thread to 
  * start. Once this method returns the
  * process has begun.
  */

 synchronized public void waitBegin()
 {
  try {
   while ( !_started ) {
    wait();
   }
  } catch ( InterruptedException e ) {
  }
 }


 /**
  * Called by a Worker object
  * to indicate that it has begun 
  * working on a workload.
  */
 synchronized public void workerBegin()
 {
  _activeThreads++;
  _started = true;
  notify();
 }

 /**
  * Called by a Worker object to 
  * indicate that it has completed a 
  * workload.
  */
 synchronized public void workerEnd()
 {
  _activeThreads--;
  notify();
 }

 /**
  * Called to reset this object to
  * its initial state.
  */
 synchronized public void reset()
 {
  _activeThreads = 0;
 }

}

请帮忙。提前致谢。期待您的善意回应。

Answer 1

现在我非常完美地理解了整个代码。如果您对这段代码有任何疑问，可以提问。

在阅读了很多关于此的问题后，我的问题的答案如下。

1. 是的，你是对的，这在逻辑上是错误的。最好是 _activeTasks 。它用于杀死所有线程，当线程池没有更多工作时，因为 waitDone() 函数只有在 _activeTasks 时才能成功执行。

2. 此变量用于 waitBegin() 方法。每当任何任务启动时，它都会以 TRUE 更新_started，表示用户分配的任务现在正在由线程处理，表示线程开始处理这些任务。如果用户没有给出任务，那么所有线程仍然处于活动状态，并等待任务。这里就是这个变量的使用。

3. waitBegin() 方法在线程开始处理任务时成功执行，因为在这种情况下只有 _started 变为 true。否则，线程会继续等待某些任务。 waitDone() 仅在_activeTasks 变为0 时执行成功，因为这是线程池没有任何工作要执行的唯一情况，意味着线程池完成了它的工作。否则，它会一直等到所有任务完成，这意味着它会等到 _activeTasks 变为 ZERO