Java 类java.util.concurrent.ForkJoinPool 实例源码

public static void main(String[] args) throws InterruptedException {
    long startTime = System.currentTimeMillis();
    int count = 0;
    for (int i = 1; i < 10; i++) {
        count = count + i;
        Thread.sleep(1000);
    }
    System.out.println(count);
    long endTime = System.currentTimeMillis(); // 获取结束时间
    System.out.println("程序运行时间： " + (startTime - endTime) + "ms");

    long startTime1 = System.currentTimeMillis();
    CountTask countTask = new CountTask(1, 10);
    ForkJoinPool forkJoinPool = new ForkJoinPool();
    Future<Integer> futureTask = forkJoinPool.submit(countTask);
    try {
        System.out.println(futureTask.get());
    } catch (ExecutionException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
    }
    long endTime1 = System.currentTimeMillis(); // 获取结束时间
    System.out.println("程序运行时间： " + (startTime1 - endTime1) + "ms");

}

@Test
public void testCreateKeyMultithreaded() {
    final int count = 100000;

    final Collection<Callable<String>> tasks = IntStream.range(0, count).boxed()
        .map(i -> (Callable<String>) () -> KeyGenerator.createKey()).collect(Collectors.toList());

    final ForkJoinPool pool = ForkJoinPool.commonPool();

    final List<Future<String>> results = pool.invokeAll(tasks);

    final Set<String> keys = results.stream().map(t -> {
        try {
            return t.get();
        } catch (InterruptedException | ExecutionException e) {
            throw new IllegalStateException(e);
        }
    }).collect(Collectors.toSet());

    Assert.assertEquals("If " + count + " key generations are performed in parallel, it should yield " + count
        + " of distinct keys", count, keys.size());
}

/**
 * execute(runnable) runs it to completion
 */
public void testExecuteRunnable() throws Throwable {
    ExecutorService e = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(e)) {
        final AtomicBoolean done = new AtomicBoolean(false);
        Future<?> future = e.submit(new CheckedRunnable() {
            public void realRun() {
                done.set(true);
            }});
        assertNull(future.get());
        assertNull(future.get(randomExpiredTimeout(), randomTimeUnit()));
        assertTrue(done.get());
        assertTrue(future.isDone());
        assertFalse(future.isCancelled());
    }
}

public GeneratorIterator(int maxWorkQueueDepth, Generator<U> theGenerator, Completion completion,
    Consumer<Throwable> exceptionHandler)
{
  this.maxWorkQueueDepth = maxWorkQueueDepth;
  this.completion = completion;
  this.exceptionHandler = exceptionHandler;
  this.workQueue = new ArrayBlockingQueue<>(maxWorkQueueDepth);
  this.drainedItems = new ArrayList<>(maxWorkQueueDepth + 1);
  this.isForkJoinTaskComplete = new AtomicBoolean(false);
  this.future = ForkJoinPool.commonPool().submit(() -> {
    try {
      theGenerator.call(this.workQueue::put);
    } catch(InterruptedException ex) {
    } finally {
      this.workQueue.done();
    }
    return null; // Void future requires a return value of null
  });
  this.drainedItemsCount = this.position = maxWorkQueueDepth;
}

private void testInvokeOnPool(ForkJoinPool pool, RecursiveAction a) {
    try (PoolCleaner cleaner = cleaner(pool)) {
        assertFalse(a.isDone());
        assertFalse(a.isCompletedNormally());
        assertFalse(a.isCompletedAbnormally());
        assertFalse(a.isCancelled());
        assertNull(a.getException());
        assertNull(a.getRawResult());

        assertNull(pool.invoke(a));

        assertTrue(a.isDone());
        assertTrue(a.isCompletedNormally());
        assertFalse(a.isCompletedAbnormally());
        assertFalse(a.isCancelled());
        assertNull(a.getException());
        assertNull(a.getRawResult());
    }
}

/**
 * setUncaughtExceptionHandler changes handler for uncaught exceptions.
 *
 * Additionally tests: Overriding ForkJoinWorkerThread.onStart
 * performs its defined action
 */
public void testSetUncaughtExceptionHandler() throws InterruptedException {
    final CountDownLatch uehInvoked = new CountDownLatch(1);
    final Thread.UncaughtExceptionHandler ueh =
        new Thread.UncaughtExceptionHandler() {
            public void uncaughtException(Thread t, Throwable e) {
                threadAssertTrue(e instanceof MyError);
                threadAssertTrue(t instanceof FailingFJWSubclass);
                uehInvoked.countDown();
            }};
    ForkJoinPool p = new ForkJoinPool(1, new FailingThreadFactory(),
                                      ueh, false);
    try (PoolCleaner cleaner = cleaner(p)) {
        assertSame(ueh, p.getUncaughtExceptionHandler());
        try {
            p.execute(new FibTask(8));
            await(uehInvoked);
        } finally {
            p.shutdownNow(); // failure might have prevented processing task
        }
    }
}

final <T> void post(@NotNull Consumer<T> callback, T state, boolean mainThreadAffinitized) {
    Requires.notNull(callback, "callback");

    if (mainThreadAffinitized) {
        JoinableFuture<?> transientFuture = this.runAsync(() -> {
            this.getContext().getAmbientFuture().post(callback, state, true);
            return Futures.completedNull();
        });

        if (transientFuture.getFuture().isCompletedExceptionally()) {
            // rethrow the exception.
            transientFuture.getFuture().join();
        }
    } else {
        ForkJoinPool.commonPool().execute(ExecutionContext.wrap(() -> callback.accept(state)));
    }
}

AbstractJFBSorting(int maximumPoints, int maximumDimension, int allowedThreads) {
    super(maximumPoints, maximumDimension);

    if (allowedThreads == 1) {
        pool = null; // current thread only execution
    } else {
        pool = allowedThreads > 1 ? new ForkJoinPool(allowedThreads) : new ForkJoinPool();
    }
    this.allowedThreads = allowedThreads > 0 ? allowedThreads : -1;

    sorter = new DoubleArraySorter(maximumPoints);
    medianSwap = new double[maximumPoints];
    indices = new int[maximumPoints];
    ranks = new int[maximumPoints];
    points = new double[maximumPoints][];
    transposedPoints = new double[maximumDimension][maximumPoints];
    rankQuery = createStructure(maximumPoints);

    internalIndices = new int[maximumPoints];
    lastFrontOrdinates = new double[maximumPoints];
    splitMerge = new SplitMergeHelper(maximumPoints);
}

/**
 * get of returned element of invokeAll(c) throws
 * ExecutionException on failed task
 */
public void testInvokeAll4() throws Throwable {
    ExecutorService e = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new NPETask());
        List<Future<String>> futures = e.invokeAll(l);
        assertEquals(1, futures.size());
        try {
            futures.get(0).get();
            shouldThrow();
        } catch (ExecutionException success) {
            assertTrue(success.getCause() instanceof NullPointerException);
        }
    }
}

private static void realMain(String[] args) throws Throwable {
    if (debug) {
        String pp = System.getProperty(
                "java.util.concurrent.ForkJoinPool.common.parallelism");
        System.out.println(
                "java.util.concurrent.ForkJoinPool.common.parallelism:" + pp);
        String tf = System.getProperty(
                "java.util.concurrent.ForkJoinPool.common.threadFactory");
        System.out.println(
                "java.util.concurrent.ForkJoinPool.common.threadFactory:" + tf);
    }

    long from = 0, to = 50000;
    RecursiveTask<Long> task = new SumTask(from, to, Thread.currentThread());
    long sum = task.invoke();
    System.out.printf("%nSum: from [%d] to [%d] = [%d]%n", from, to, sum);

    task.fork();
    sum = task.join();
    System.out.printf("%nSum: from [%d] to [%d] = [%d]%n", from, to, sum);

    sum = ForkJoinPool.commonPool().invoke(task.fork());
    System.out.printf("%nSum: from [%d] to [%d] = [%d]%n", from, to, sum);
}

private static void demo1_class_level_integration() {
    String result = IntStream.rangeClosed(1, speedLimitByLane.length).mapToDouble(i -> {
        AverageSpeed averageSpeed = new AverageSpeed(trafficUnitsNumber, timeSec, dateLocation, speedLimitByLane, i,100);
        ForkJoinPool commonPool = ForkJoinPool.commonPool();
        return commonPool.invoke(averageSpeed);
    }).mapToObj(Double::toString).collect(Collectors.joining(", "));
    System.out.println("Average speed = " + result);

    TrafficDensity trafficDensity = new TrafficDensity();
    Integer[] trafficByLane = trafficDensity.trafficByLane(trafficUnitsNumber, timeSec, dateLocation, speedLimitByLane );
    System.out.println("Traffic density = " + Arrays.stream(trafficByLane).map(Object::toString).collect(Collectors.joining(", ")));
}

public static BufferedImage blur(BufferedImage srcImage) {
    int w = srcImage.getWidth();
    int h = srcImage.getHeight();

    int[] src = srcImage.getRGB(0, 0, w, h, null, 0, w);
    int[] dst = new int[src.length];

    System.out.println("Array size is " + src.length);
    System.out.println("Threshold is " + sThreshold);

    int processors = Runtime.getRuntime().availableProcessors();
    System.out.println(Integer.toString(processors) + " processor"
            + (processors != 1 ? "s are " : " is ")
            + "available");

    ForkBlur fb = new ForkBlur(src, 0, src.length, dst);

    ForkJoinPool pool = new ForkJoinPool();

    long startTime = System.currentTimeMillis();
    pool.invoke(fb);
    long endTime = System.currentTimeMillis();

    System.out.println("Image blur took " + (endTime - startTime) +
            " milliseconds.");

    BufferedImage dstImage =
            new BufferedImage(w, h, BufferedImage.TYPE_INT_ARGB);
    dstImage.setRGB(0, 0, w, h, dst, 0, w);

    return dstImage;
}

public double getAverageMoreProcessors() throws InterruptedException, ExecutionException{
    ToIntFunction<Employee> sizeEmpArr = (e) -> {
        System.out.println("Thread: " + Thread.currentThread().getName());
            return e.getAge();
       };
    Callable<Double> task = () -> employeeDaoImpl.getEmployees().stream().mapToInt(sizeEmpArr).average().getAsDouble();
    ForkJoinPool forkJoinPool = new ForkJoinPool(4);

    double avgAge = forkJoinPool.submit(task).get();
    return avgAge;

}

public void testAbnormalInvoke(ForkJoinPool pool) {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            FailingAsyncFib f = new FailingAsyncFib(8);
            try {
                f.invoke();
                shouldThrow();
            } catch (FJException success) {
                checkCompletedAbnormally(f, success);
            }
        }};
    testInvokeOnPool(pool, a);
}

/**
 * timed invokeAll(null) throws NullPointerException
 */
public void testTimedInvokeAll1() throws Throwable {
    ExecutorService e = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(e)) {
        try {
            e.invokeAll(null, randomTimeout(), randomTimeUnit());
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

/**
 * timed invokeAll(c) returns results of all completed tasks in c
 */
public void testTimedInvokeAll5() throws Throwable {
    ForkJoinPool e = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(e)) {
        List<Callable<String>> l = new ArrayList<>();
        l.add(new StringTask());
        l.add(new StringTask());
        List<Future<String>> futures
            = e.invokeAll(l, LONG_DELAY_MS, MILLISECONDS);
        assertEquals(2, futures.size());
        for (Future<String> future : futures)
            assertSame(TEST_STRING, future.get());
    }
}

/**
 * invokeAny(null) throws NullPointerException
 */
public void testInvokeAny1() throws Throwable {
    ExecutorService e = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(e)) {
        try {
            e.invokeAny(null);
            shouldThrow();
        } catch (NullPointerException success) {}
    }
}

protected ForkJoinPool createService() {
    // TODO Need spring config for this.
    Integer processors = Integer.getInteger("org.eclipse.scanning.level.runner.pool.count");
    if (processors==null || processors<1) processors = Runtime.getRuntime().availableProcessors();
    return new ForkJoinPool(processors);
       // Slightly faster than thread pool executor @see ScanAlgorithmBenchMarkTest
}

/**
 * Returns an array of the elements in this range
 * @param parallel  true to assemble the array using fork & join
 * @return          the array of elements in range
 */
@SuppressWarnings("unchecked")
default Array<T> toArray(boolean parallel) {
    if (!parallel) {
        final int length = (int)estimateSize();
        final Iterable<Object> iterable = (Iterable<Object>)this;
        return (Array<T>)ArrayBuilder.of(length).addAll(iterable).toArray();
    } else {
        final ToArrayTask<T> task = new ToArrayTask<>(this, 1000);
        return ForkJoinPool.commonPool().invoke(task);
    }
}

/** Root task constructor */
public IntCumulateTask(IntCumulateTask parent,
                       IntBinaryOperator function,
                       int[] array, int lo, int hi) {
    super(parent);
    this.function = function; this.array = array;
    this.lo = this.origin = lo; this.hi = this.fence = hi;
    int p;
    this.threshold =
        (p = (hi - lo) / (ForkJoinPool.getCommonPoolParallelism() << 3))
        <= MIN_PARTITION ? MIN_PARTITION : p;
}

private void waitForBackgroundProcessToComplete() throws InterruptedException
{
    while (((ForkJoinPool) executor).getActiveThreadCount() > 0)
    {
        Thread.sleep(1);
    }
}

/**
 * getPool of executing task returns its pool
 */
public void testGetPool() {
    final ForkJoinPool mainPool = mainPool();
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            assertSame(mainPool, getPool());
        }};
    testInvokeOnPool(mainPool, a);
}

/** Root task constructor */
public CumulateTask(CumulateTask<T> parent,
                    BinaryOperator<T> function,
                    T[] array, int lo, int hi) {
    super(parent);
    this.function = function; this.array = array;
    this.lo = this.origin = lo; this.hi = this.fence = hi;
    int p;
    this.threshold =
            (p = (hi - lo) / (ForkJoinPool.getCommonPoolParallelism() << 3))
            <= MIN_PARTITION ? MIN_PARTITION : p;
}

/** Root task constructor */
public LongCumulateTask(LongCumulateTask parent,
                        LongBinaryOperator function,
                        long[] array, int lo, int hi) {
    super(parent);
    this.function = function; this.array = array;
    this.lo = this.origin = lo; this.hi = this.fence = hi;
    int p;
    this.threshold =
            (p = (hi - lo) / (ForkJoinPool.getCommonPoolParallelism() << 3))
            <= MIN_PARTITION ? MIN_PARTITION : p;
}

/** Root task constructor */
public DoubleCumulateTask(DoubleCumulateTask parent,
                          DoubleBinaryOperator function,
                          double[] array, int lo, int hi) {
    super(parent);
    this.function = function; this.array = array;
    this.lo = this.origin = lo; this.hi = this.fence = hi;
    int p;
    this.threshold =
            (p = (hi - lo) / (ForkJoinPool.getCommonPoolParallelism() << 3))
            <= MIN_PARTITION ? MIN_PARTITION : p;
}

/**
 * get of submit(callable) throws ExecutionException if callable
 * throws exception
 */
public void testSubmitEE() throws Throwable {
    ForkJoinPool p = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(p)) {
        try {
            p.submit(new Callable() {
                    public Object call() { throw new ArithmeticException(); }})
                .get();
            shouldThrow();
        } catch (ExecutionException success) {
            assertTrue(success.getCause() instanceof ArithmeticException);
        }
    }
}

private void testInvokeOnPool(ForkJoinPool pool, RecursiveAction a) {
    try (PoolCleaner cleaner = cleaner(pool)) {
        checkNotDone(a);

        assertNull(pool.invoke(a));

        checkCompletedNormally(a);
    }
}

/**
 * Main method of the class
 */
public static void main(String[] args) {
    // Array of 100 integers
    int array[]=new int[100];
    // Task to process the array
    Task task=new Task(array,0,100);
    // ForkJoinPool to execute the Task
    ForkJoinPool pool=new ForkJoinPool();

    // Execute the task
    pool.execute(task);

    // Shutdown the ForkJoinPool
    pool.shutdown();

    // Wait for the finalization of the task
    try {
        pool.awaitTermination(1, TimeUnit.DAYS);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    // Check if the task has thrown an Exception. If it's the case, write it
    // to the console

    if (task.isCompletedAbnormally()) {
        System.out.printf("Main: An exception has ocurred\n");
        System.out.printf("Main: %s\n",task.getException());
    }

    System.out.printf("Main: Result: %d",task.join());
}

@Override
public int count(Predicate<ArrayValue<T>> predicate) {
    if (isParallel() && length() > 0) {
        final int processors = Runtime.getRuntime().availableProcessors();
        final int splitThreshold = Math.max(length() / processors, 10000);
        return ForkJoinPool.commonPool().invoke(new CountTask<>(this, 0, length()-1, splitThreshold, predicate));
    } else {
        final CountTask task = new CountTask<>(this, 0, length()-1, Integer.MAX_VALUE, predicate);
        return task.compute();
    }
}

@Override
public final Array<Boolean> mapToBooleans(ToBooleanFunction<ArrayValue<T>> mapper) {
    final Array<Boolean> result = Array.of(Boolean.class, length());
    final MapValues<Boolean> action = new MapValues<>(0, length() - 1, mapper, result);
    if (isParallel()) {
        ForkJoinPool.commonPool().invoke(action);
        return result;
    } else {
        action.compute();
        return result;
    }
}

/**
 * A submitted privileged action runs to completion
 */
public void testSubmitPrivilegedAction() throws Exception {
    final Callable callable = Executors.callable(new PrivilegedAction() {
            public Object run() { return TEST_STRING; }});
    Runnable r = new CheckedRunnable() {
    public void realRun() throws Exception {
        ExecutorService e = new ForkJoinPool(1);
        try (PoolCleaner cleaner = cleaner(e)) {
            Future future = e.submit(callable);
            assertSame(TEST_STRING, future.get());
        }
    }};

    runWithPermissions(r, new RuntimePermission("modifyThread"));
}

/**
 * Completed submit(ForkJoinTask) returns result
 */
public void testSubmitForkJoinTask() throws Throwable {
    ForkJoinPool p = new ForkJoinPool(1);
    try (PoolCleaner cleaner = cleaner(p)) {
        ForkJoinTask<Integer> f = p.submit(new FibTask(8));
        assertEquals(21, (int) f.get());
    }
}

public void testForkQuietlyJoin(ForkJoinPool pool) {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib f = new AsyncFib(8);
            assertSame(f, f.fork());
            f.quietlyJoin();
            f.checkCompletedNormally();
        }};
    testInvokeOnPool(pool, a);
}

public static void main(String[] args) throws Throwable {
    final ForkJoinPool e = new ForkJoinPool(1);
    final AtomicBoolean b = new AtomicBoolean();
    final Runnable setFalse = () -> b.set(false);
    for (int i = 0; i < 100000; i++) {
        b.set(true);
        e.execute(setFalse);
        long st = System.nanoTime();
        while (b.get()) {
            if (System.nanoTime() - st >= TimeUnit.SECONDS.toNanos(10)) {
                throw new RuntimeException("Submitted task failed to execute");
            }
        }
    }
}

/**
 * Pool maintains parallelism when using ManagedBlocker
 */
public void testBlockingForkJoinTask() throws Throwable {
    ForkJoinPool p = new ForkJoinPool(4);
    try {
        ReentrantLock lock = new ReentrantLock();
        ManagedLocker locker = new ManagedLocker(lock);
        ForkJoinTask<Integer> f = new LockingFibTask(20, locker, lock);
        p.execute(f);
        assertEquals(6765, (int) f.get());
    } finally {
        p.shutdownNow(); // don't wait out shutdown
    }
}

public void testForkTimedGet(ForkJoinPool pool) {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() throws Exception {
            AsyncFib f = new AsyncFib(8);
            assertSame(f, f.fork());
            assertNull(f.get(LONG_DELAY_MS, MILLISECONDS));
            f.checkCompletedNormally();
        }};
    testInvokeOnPool(pool, a);
}

public void testInvokeAllNullTask(ForkJoinPool pool) {
    RecursiveAction a = new CheckedRecursiveAction() {
        protected void realCompute() {
            AsyncFib nul = null;
            Runnable[] throwingActions = {
                () -> invokeAll(nul),
                () -> invokeAll(nul, nul),
                () -> invokeAll(new AsyncFib(8), new AsyncFib(9), nul),
                () -> invokeAll(new AsyncFib(8), nul, new AsyncFib(9)),
                () -> invokeAll(nul, new AsyncFib(8), new AsyncFib(9)),
            };
            assertThrows(NullPointerException.class, throwingActions);
        }};
    testInvokeOnPool(pool, a);
}

private void startConfiguredLead(LeadStateAssistant assistant, GapDetectionStrategy gapDetectionStrategy) {
    RuleTimeouts ruleTimeouts = new RuleTimeouts();
    Heartbeats heartbeats = new Heartbeats(ruleTimeouts.getHearbeatExpirationThreshold());

    proxyReconciler = new ProxyReconciler();
    lead = new LeadImpl(assistant, new ReadTransactions(), CommittedTransactions.createNotReady(), heartbeats,
            gapDetectionStrategy, proxyReconciler, ruleTimeouts);
    ForkJoinPool.commonPool().submit(() -> lead.execute());
}

static void parallel() {
//        System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "3");
        System.out.println(String.format("parallel() on %d threads. Available CPU: %d\n",
                ForkJoinPool.commonPool().getParallelism(), Runtime.getRuntime().availableProcessors()));

        int result = IntStream.range(0, 5)
                .parallel()
                .peek(it -> System.out.printf("Thread [%s] peek: %d\n", Thread.currentThread().getName(), it))
                .sum();
        System.out.println("sum: " + result);
    }

private void onCompletedImpl(@NotNull Runnable continuation, boolean useExecutionContext) {
    Requires.notNull(continuation, "continuation");

    Executor executor = ForkJoinPool.commonPool();
    SynchronizationContext synchronizationContext = SynchronizationContext.getCurrent();
    if (synchronizationContext != null && synchronizationContext.getClass() != SynchronizationContext.class) {
        executor = synchronizationContext;
    }

    Runnable wrappedContinuation = useExecutionContext ? ExecutionContext.wrap(continuation) : continuation;
    executor.execute(wrappedContinuation);
}