如何用通知唤醒RecursiveTask<V>？

Question

当这些任务被ForkJoinPool方法搁置时，我怎么才能唤醒由ForkJoinPool调用的。下面是MainRecursionClass.resume方法的简单示例，它不正确(没有唤醒RecursiveTasks)。

public class Program {
   public static void main(String[] args) {
      Program p = new Program();
      final MainRecursionClass mrc = p.new MainRecursionClass();

      //Thread outputs integers to simulate work
      new Thread() {
         public void run() {
            mrc.doJob();
         }
      }.start();

      //Thread performs wait and notify on MainRecursionClass object
      p.new PauseResume(mrc).start();
   }

/**
 * 
 * This class performs suspend and resume operations to the MainRecursionClass class object
 *
 */
   private class PauseResume extends Thread {
      private MainRecursionClass rv;

      public PauseResume(MainRecursionClass rv) {
         this.rv = rv;
      }

      @Override
      public void run() {
         while(!isInterrupted()) {
            try {
               sleep(4000);

               rv.suspend();
               sleep(8000);

               rv.resume();
            } catch (InterruptedException e) {
               e.printStackTrace();
            }
         }
      }
   }

   private class MainRecursionClass {
      private boolean pause = false;
      private MyRecursive rv;

      public void doJob() {
         rv = new MyRecursive(0, 100000);
         ForkJoinPool pool = new ForkJoinPool();      
         pool.invoke(rv);
      }

      public void suspend() {
         pause = true;

         System.out.println("Suspended");
      }

      /**
       * This method is incorrect. It should wake up all MyRecursive instances to continue their work.
       */
      public synchronized void resume() {
         pause = false;
         notifyAll();

         System.out.println("Resumed");
      }

      private class MyRecursive extends RecursiveTask<Object> {
         private static final long serialVersionUID = 1L;
         private int start;
         private int length;
         private int threshold = 15;

         public MyRecursive(int start, int length) {
            super();
            this.start = start;
            this.length = length;
         }

         protected void computeDirectly() throws Exception {
            for (int index = start; index < start + length; index++) {
               //PAUSE
               synchronized (this) {
                  try {
                     while(pause) {
                        wait();
                     }
                  } catch (InterruptedException e) {
                     e.printStackTrace();
                  }
               }
               //PAUSE

               //some output to simulate work...
               System.out.println(index);

               Thread.sleep(1000);
            }
         }

         /**
          * Recursion
          */
         @Override
         protected Object compute() {
            if (length <= threshold) {
               try {
                  computeDirectly();
               } catch (Exception e) {
                  return e;
               }

               return null;
            }

            int split = length / 2;

            invokeAll(new MyRecursive(start, split),
                      new MyRecursive(start + split, length - split));

            return null;
         }
      }
   }
}

Answer 1

最后，我找到了这个解决方案:我在MainRecursionClass中创建了MainRecursionClass对象，并在list中添加了递归创建的每个MyRecursive实例。类MyRecursive有新的方法：

public synchronized void resume() {
   notify();
}

当方法MainRecursionClass.resume()唤醒线程时，如下所示：

public void resume() {
   System.out.println("Resumed");

   pause = false;
   for(MyRecursive mr : list) {
      if(mr != null)
         mr.resume();
      }
   }
}

Answer 2

在线程池上运行的任务中，不应使用等待/通知。如果您的池是有界的，那么它可能导致线程饥饿(死锁的一种形式)。如果没有限制，就会创建过多的线程并耗尽主内存。

相反，您应该将任务拆分为2(或更多)，并根据其启动条件启动子任务。当您想要一个任务到wait()，然后重构它，使当前的子任务退出，下一个子任务准备运行。