java.util.concurrent.ForkJoinPool#invokeAll ( )源码实例Demo

@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());
}
 

/**
 * 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());
    }
}
 

/**
 * 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<Callable<String>>();
        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());
    }
}
 

@Override
public void transactionMarker() throws Exception {
    ForkJoinPool pool = new ForkJoinPool();
    List<Callable<Void>> callables = Lists.newArrayList();
    callables.add(new SimpleCallable());
    callables.add(new SimpleCallable());
    callables.add(new SimpleCallable());
    for (Future<Void> future : pool.invokeAll(callables)) {
        future.get();
    }
}