CountedCompleter

public abstract class CountedCompleter
extends ForkJoinTask<T>

一个ForkJoinTask ，触发时执行完成操作，并且没有剩余的待处理操作。 与其他形式的ForkJoinTasks相比，CountedCompleters在存在子任务停滞和阻塞方面通常更加健壮，但编程起来不太直观。 CountedCompleter的使用与其他基于完成的组件相似，只是可能需要多个待执行完成来触发完成操作onCompletion(CountedCompleter) ，而不仅仅是一个。 除非另有初始化， pending count开始于零，但也可以是（原子），使用方法改变setPendingCount(int) ， addToPendingCount(int) ，和compareAndSetPendingCount(int, int) 。 在调用tryComplete() ，如果挂起的操作计数为非零，则递减; 否则，完成动作被执行，并且如果这个完成者本身具有完成者，则该过程继续与其完成者。 与相关同步组件（如Phaser和Semaphore ，这些方法仅影响内部计数; 他们不建立任何进一步的内部簿记。 具体而言，待定任务的身份不会被维护。 如下图所示，您可以创建子类，以便在需要时记录部分或全部未决任务或其结果。 如下所示，还提供了支持自定义完成遍历的实用程序方法。 但是，因为CountedCompleters只提供基本的同步机制，所以创建更多的抽象子类可能会很有用，这些子类可以维护适用于一组相关用法的链接，字段和其他支持方法。

具体的CountedCompleter类必须定义方法compute() ，应该在大多数情况下（如下图所示）在返回之前调用tryComplete()一次。 该类也可以可选地覆盖方法onCompletion(CountedCompleter)以在正常完成时执行动作，并且方法onExceptionalCompletion(Throwable, CountedCompleter)在任何异常时执行动作。

CountedCompleters通常不会产生结果，在这种情况下，它们通常被声明为CountedCompleter<Void> ，并且总是会返回null作为结果值。 在其他情况下，您应该重写方法getRawResult()以提供join(), invoke()和相关方法的结果。 一般而言，此方法应返回完成时保存结果的CountedCompleter对象的字段值（或一个或多个字段的函数）。 方法setRawResult(T)默认在CountedCompleters中不起作用。 可能但很少适用覆盖此方法以维护其他对象或保存结果数据的字段。

一个没有完成的getCompleter() （即getCompleter()返回null ）可以用作一个常规的ForkJoinTask，具有这个附加功能。 然而，任何完成者又有另一个完成者只能作为其他计算的内部帮助者，所以它自己的任务状态（如isDone()等方法中所isDone() ）是任意的; 这种状况只有在明确调用改变complete(T) ， cancel(boolean) ， completeExceptionally(Throwable)或方法的特殊结束后compute 。 在任何异常完成时，例外情况可能会传递给任务完成者（及其完成者等）（如果存在并且尚未完成）。 同样，取消一个内部CountedCompleter对该完成者只有一个本地影响，所以通常不会有用。

样本使用。

并行递归分解。 CountedCompleters可以安排在与RecursiveAction经常使用的树相似的树上，尽管设置它们的构造通常会有所不同。 这里，每个任务的完成者是计算树中的父代。 即使它们需要更多簿记，当将可能耗时的操作（不能进一步细分）应用到数组或集合的每个元素时，CountedCompleters可能会是更好的选择; 尤其是当操作需要比其他操作花费大量时间完成某些元素时，无论是因为内在变化（例如I / O）还是辅助效果（如垃圾收集）。 因为CountedCompleters提供了自己的延续，所以其他线程不需要阻止等待执行它们。

例如，下面是一个类的初始版本，它使用二分递归分解将工作分成单个部分（叶子任务）。 即使将工作分解为单个调用，基于树的技术通常也更适合直接分派叶任务，因为它们减少了线程间通信并提高了负载平衡。 在递归情况下，每对子任务中的第二onCompletion任务完成触发其父项完成（因为没有执行结果组合，方法onCompletion的默认无操作实现未被覆盖）。 静态实用程序方法设置基本任务并调用它（此处隐式使用commonPool() ）。

 class MyOperation<E> { void apply(E e) { ... }  }

 class ForEach<E> extends CountedCompleter<Void> {

   public static <E> void forEach(E[] array, MyOperation<E> op) {
     new ForEach<E>(null, array, op, 0, array.length).invoke();
   }

   final E[] array; final MyOperation<E> op; final int lo, hi;
   ForEach(CountedCompleter<?> p, E[] array, MyOperation<E> op, int lo, int hi) {
     super(p);
     this.array = array; this.op = op; this.lo = lo; this.hi = hi;
   }

   public void compute() { // version 1
     if (hi - lo >= 2) {
       int mid = (lo + hi) >>> 1;
       setPendingCount(2); // must set pending count before fork
       new ForEach(this, array, op, mid, hi).fork(); // right child
       new ForEach(this, array, op, lo, mid).fork(); // left child
     }
     else if (hi > lo)
       op.apply(array[lo]);
     tryComplete();
   }
 }

 class ForEach<E> ... {
   ...
   public void compute() { // version 2
     if (hi - lo >= 2) {
       int mid = (lo + hi) >>> 1;
       setPendingCount(1); // only one pending
       new ForEach(this, array, op, mid, hi).fork(); // right child
       new ForEach(this, array, op, lo, mid).compute(); // direct invoke
     }
     else {
       if (hi > lo)
         op.apply(array[lo]);
       tryComplete();
     }
   }
 }

 class ForEach<E> ... {
   ...
   public void compute() { // version 3
     int l = lo, h = hi;
     while (h - l >= 2) {
       int mid = (l + h) >>> 1;
       addToPendingCount(1);
       new ForEach(this, array, op, mid, h).fork(); // right child
       h = mid;
     }
     if (h > l)
       op.apply(array[l]);
     propagateCompletion();
   }
 }

搜索。 CountedCompleters树可以在数据结构的不同部分中搜索值或属性，并在找到一个结果后立即报告结果AtomicReference 。 其他人可以查询结果以避免不必要的工作。 （您还可以执行其他任务，但通常让他们注意到结果已设置，并且如果跳过了进一步的处理，通常会更简单，更高效。）使用完全分区再次说明数组（在实践中，叶子任务几乎总是会处理多个元素）：

 class Searcher<E> extends CountedCompleter<E> {
   final E[] array; final AtomicReference<E> result; final int lo, hi;
   Searcher(CountedCompleter<?> p, E[] array, AtomicReference<E> result, int lo, int hi) {
     super(p);
     this.array = array; this.result = result; this.lo = lo; this.hi = hi;
   }
   public E getRawResult() { return result.get(); }
   public void compute() { // similar to ForEach version 3
     int l = lo, h = hi;
     while (result.get() == null && h >= l) {
       if (h - l >= 2) {
         int mid = (l + h) >>> 1;
         addToPendingCount(1);
         new Searcher(this, array, result, mid, h).fork();
         h = mid;
       }
       else {
         E x = array[l];
         if (matches(x) && result.compareAndSet(null, x))
           quietlyCompleteRoot(); // root task is now joinable
         break;
       }
     }
     tryComplete(); // normally complete whether or not found
   }
   boolean matches(E e) { ... } // return true if found

   public static <E> E search(E[] array) {
       return new Searcher<E>(null, array, new AtomicReference<E>(), 0, array.length).invoke();
   }
 }

记录子任务。 将多个子任务的结果组合在一起的CountedCompleter任务通常需要通过方法onCompletion(CountedCompleter)访问这些结果。 如下面的类所示（执行映射和缩减的简化形式，映射和缩减都是类型E ），在分而治之设计中做到这一点的一种方式是让每个子任务记录其兄弟，以便它可以可通过方法onCompletion访问。 该技术适用于左右结果组合顺序无关紧要的减少; 有序的减少需要明确的左/右指定。 上述例子中看到的其他流线型的变体也可能适用。

 class MyMapper<E> { E apply(E v) {  ...  } }
 class MyReducer<E> { E apply(E x, E y) {  ...  } }
 class MapReducer<E> extends CountedCompleter<E> {
   final E[] array; final MyMapper<E> mapper;
   final MyReducer<E> reducer; final int lo, hi;
   MapReducer<E> sibling;
   E result;
   MapReducer(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
              MyReducer<E> reducer, int lo, int hi) {
     super(p);
     this.array = array; this.mapper = mapper;
     this.reducer = reducer; this.lo = lo; this.hi = hi;
   }
   public void compute() {
     if (hi - lo >= 2) {
       int mid = (lo + hi) >>> 1;
       MapReducer<E> left = new MapReducer(this, array, mapper, reducer, lo, mid);
       MapReducer<E> right = new MapReducer(this, array, mapper, reducer, mid, hi);
       left.sibling = right;
       right.sibling = left;
       setPendingCount(1); // only right is pending
       right.fork();
       left.compute();     // directly execute left
     }
     else {
       if (hi > lo)
           result = mapper.apply(array[lo]);
       tryComplete();
     }
   }
   public void onCompletion(CountedCompleter<?> caller) {
     if (caller != this) {
       MapReducer<E> child = (MapReducer<E>)caller;
       MapReducer<E> sib = child.sibling;
       if (sib == null || sib.result == null)
         result = child.result;
       else
         result = reducer.apply(child.result, sib.result);
     }
   }
   public E getRawResult() { return result; }

   public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
     return new MapReducer<E>(null, array, mapper, reducer,
                              0, array.length).invoke();
   }
 }

完成遍历 。 如果使用onCompletion处理完成不适用或不方便，则可以使用方法firstComplete()和nextComplete()创建自定义遍历。 例如，要定义仅以第三个ForEach示例的形式分割右侧任务的MapReducer，完成必须合作减少未占用的子任务链接，这可以按如下方式完成：

 class MapReducer<E> extends CountedCompleter<E> { // version 2
   final E[] array; final MyMapper<E> mapper;
   final MyReducer<E> reducer; final int lo, hi;
   MapReducer<E> forks, next; // record subtask forks in list
   E result;
   MapReducer(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
              MyReducer<E> reducer, int lo, int hi, MapReducer<E> next) {
     super(p);
     this.array = array; this.mapper = mapper;
     this.reducer = reducer; this.lo = lo; this.hi = hi;
     this.next = next;
   }
   public void compute() {
     int l = lo, h = hi;
     while (h - l >= 2) {
       int mid = (l + h) >>> 1;
       addToPendingCount(1);
       (forks = new MapReducer(this, array, mapper, reducer, mid, h, forks)).fork();
       h = mid;
     }
     if (h > l)
       result = mapper.apply(array[l]);
     // process completions by reducing along and advancing subtask links
     for (CountedCompleter<?> c = firstComplete(); c != null; c = c.nextComplete()) {
       for (MapReducer t = (MapReducer)c, s = t.forks; s != null; s = t.forks = s.next)
         t.result = reducer.apply(t.result, s.result);
     }
   }
   public E getRawResult() { return result; }

   public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
     return new MapReducer<E>(null, array, mapper, reducer,
                              0, array.length, null).invoke();
   }
 }

触发。 一些CountedCompleters本身从不分叉，而是在其他设计中用作管道的一部分; 包括那些完成一个或多个异步任务的人触发另一个异步任务。 例如：

 class HeaderBuilder extends CountedCompleter<...> { ... }
 class BodyBuilder extends CountedCompleter<...> { ... }
 class PacketSender extends CountedCompleter<...> {
   PacketSender(...) { super(null, 1); ... } // trigger on second completion
   public void compute() { } // never called
   public void onCompletion(CountedCompleter<?> caller) { sendPacket(); }
 }
 // sample use:
 PacketSender p = new PacketSender();
 new HeaderBuilder(p, ...).fork();
 new BodyBuilder(p, ...).fork();

Summary

Inherited methods
From class java.util.concurrent.ForkJoinTask static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) 返回一个新的 ForkJoinTask ，它执行给定 Runnable的 run方法作为其操作，并返回给定结果 join() 。 static ForkJoinTask<?> adapt(Runnable runnable) 返回一个新的 ForkJoinTask ，它执行给定 Runnable的 run方法作为其操作，并在 join()返回空结果。 static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) 返回一个新的 ForkJoinTask ，它执行给定 Callable的 call方法作为其操作，并返回 join()结果，将所有遇到的检查异常转换为 RuntimeException 。 boolean cancel(boolean mayInterruptIfRunning) 试图取消执行此任务。 final boolean compareAndSetForkJoinTaskTag(short expect, short update) 原子上有条件地设置此任务的标记值。 void complete(T value) 完成此任务，如果尚未中止或取消，则返回给定值，作为后续调用 join及相关操作的结果。 void completeExceptionally(Throwable ex) 完成此任务异常，并且如果尚未中止或取消，导致它在 join和相关操作时抛出给定的异常。 abstract boolean exec() 立即执行此任务的基本操作，并返回true，如果从此方法返回后，此任务保证已正常完成。 final ForkJoinTask<T> fork() 安排在池异步执行这一任务，当前任务运行在（如果适用），或使用 commonPool()如果不是 inForkJoinPool() 。 final T get(long timeout, TimeUnit unit) 如果需要，最多等待计算完成的给定时间，然后检索其结果（如果可用）。 final T get() 如果需要，等待计算完成，然后检索其结果。 final Throwable getException() 返回基本计算抛出的异常，或者 CancellationException如果取消）或 null如果没有取消）或方法尚未完成。 final short getForkJoinTaskTag() 返回此任务的标签。 static ForkJoinPool getPool() 返回宿主当前线程的池，如果当前线程在任何ForkJoinPool外执行，则返回 null 。 static int getQueuedTaskCount() 返回当前工作线程已分叉但尚未执行的任务数的估计值。 abstract T getRawResult() 返回 join()返回的结果，即使此任务异常完成，或者 null如果此任务未知完成。 static int getSurplusQueuedTaskCount() 返回当前工作线程持有多少本地排队任务的估计值，而不是其他可能窃取它们的工作线程的值，如果此线程未在ForkJoinPool中运行，则返回零。 static void helpQuiesce() 可能执行任务，直到承载当前任务的池 is quiescent 。 static boolean inForkJoinPool() 返回 true如果当前线程是 ForkJoinWorkerThread执行作为ForkJoinPool计算。 final T invoke() 开始执行此任务，如果需要，等待其完成，并返回其结果，或者如果底层计算执行此操作，则抛出（未选中） RuntimeException或 Error 。 static void invokeAll(ForkJoinTask...<?> tasks) isDone给定的任务，当 isDone对每个任务保持 isDone或遇到（未经检查）的异常时返回，在这种情况下，将重新抛出异常。 static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) isDone给定的任务，当 isDone对每个任务保持 isDone或遇到（未经检查）的异常时返回，在这种情况下重新排除异常。 static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) isDone指定集合中的所有任务，当每个任务保留 isDone或遇到（未检查）异常时返回，在这种情况下，将重新抛出异常。 final boolean isCancelled() 如果此任务在正常完成之前取消，则返回 true 。 final boolean isCompletedAbnormally() 如果此任务引发异常或被取消，则返回 true 。 final boolean isCompletedNormally() 如果此任务完成而没有抛出异常并且未取消，则返回 true 。 final boolean isDone() 如果此任务完成，则返回 true 。 final T join() 返回 is done时的计算结果。 static ForkJoinTask<?> peekNextLocalTask() 返回但不排除或执行由当前线程排队但尚未执行的任务（如果立即可用）。 static ForkJoinTask<?> pollNextLocalTask() 如果当前线程正在ForkJoinPool中运行，那么取消调度并返回（不执行）下一个由当前线程排队但尚未执行的任务。 static ForkJoinTask<?> pollTask() 如果当前线程正在ForkJoinPool中运行，则取消调度并返回（不执行）由当前线程排队但尚未执行的下一个任务（如果有）或者不可用时由其他线程分叉的任务，如果可供使用的话。 final void quietlyComplete() 通常无需设置值即可完成此任务。 final void quietlyInvoke() 开始执行此任务并在必要时等待完成，而不返回结果或抛出异常。 final void quietlyJoin() 加入此任务，但不返回结果或抛出异常。 void reinitialize() 重置此任务的内部簿记状态，允许随后的 fork 。 final short setForkJoinTaskTag(short newValue) 以原子方式设置此任务的标记值并返回旧值。 abstract void setRawResult(T value) 因此强制返回给定的值。 boolean tryUnfork() 试图排除这个任务的执行。
From class java.lang.Object Object clone() 创建并返回此对象的副本。 boolean equals(Object obj) 指示其他某个对象是否“等于”这一个。 void finalize() 当垃圾收集确定没有更多对该对象的引用时，由对象上的垃圾回收器调用。 final Class<?> getClass() 返回此 Object的运行时类。 int hashCode() 返回对象的哈希码值。 final void notify() 唤醒正在等待该对象监视器的单个线程。 final void notifyAll() 唤醒在该对象监视器上等待的所有线程。 String toString() 返回对象的字符串表示形式。 final void wait(long millis, int nanos) 导致当前线程等待，直到另一个线程为此对象调用 notify()方法或 notifyAll()方法，或者某个其他线程中断当前线程，或者经过一定的实时时间。 final void wait(long millis) 导致当前线程等待，直到另一个线程调用此对象的 notify()方法或 notifyAll()方法或经过了指定的时间量。 final void wait() 导致当前线程等待，直到另一个线程调用此对象的 notify()方法或 notifyAll()方法。
From interface java.util.concurrent.Future abstract boolean cancel(boolean mayInterruptIfRunning) 试图取消执行此任务。 abstract T get(long timeout, TimeUnit unit) 如果需要，最多等待计算完成的给定时间，然后检索其结果（如果可用）。 abstract T get() 如果需要，等待计算完成，然后检索其结果。 abstract boolean isCancelled() 如果此任务在正常完成之前取消，则返回 true 。 abstract boolean isDone() 如果此任务完成，则返回 true 。

Protected constructors

CountedCompleter (CountedCompleter<?> completer, 
                int initialPendingCount)

CountedCompleter (CountedCompleter<?> completer)

CountedCompleter ()

Public methods

void addToPendingCount (int delta)

boolean compareAndSetPendingCount (int expected, 
                int count)

void complete (T rawResult)

void compute ()

int decrementPendingCountUnlessZero ()

CountedCompleter<?> firstComplete ()

CountedCompleter<?> getCompleter ()

int getPendingCount ()

T getRawResult ()

CountedCompleter<?> getRoot ()

void helpComplete (int maxTasks)

CountedCompleter<?> nextComplete ()

 for (CountedCompleter<?> c = firstComplete();
      c != null;
      c = c.nextComplete()) {
   // ... process c ...
 }

void onCompletion (CountedCompleter<?> caller)

boolean onExceptionalCompletion (Throwable ex, 
                CountedCompleter<?> caller)

void propagateCompletion ()

void quietlyCompleteRoot ()

void setPendingCount (int count)

void tryComplete ()

Protected methods

boolean exec ()

void setRawResult (T t)