下面列出了怎么用java.util.concurrent.ForkJoinTask的API类实例代码及写法,或者点击链接到github查看源代码。
/**
* timed get of a forked task throws exception when task completes abnormally
*/
public void testAbnormalForkTimedGetSingleton() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FailingCCF f = new LFCCF(8);
assertSame(f, f.fork());
try {
f.get(LONG_DELAY_MS, MILLISECONDS);
shouldThrow();
} catch (ExecutionException success) {
Throwable cause = success.getCause();
assertTrue(cause instanceof FJException);
checkCompletedAbnormally(f, cause);
}
}};
testInvokeOnPool(singletonPool(), a);
}
/**
* invokeAll(tasks) with > 2 argument invokes tasks
*/
public void testInvokeAll3() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(8);
CCF g = new LCCF(9);
CCF h = new LCCF(7);
invokeAll(f, g, h);
assertEquals(21, f.number);
assertEquals(34, g.number);
assertEquals(13, h.number);
checkCompletedNormally(f);
checkCompletedNormally(g);
checkCompletedNormally(h);
}};
testInvokeOnPool(mainPool(), a);
}
/**
* invokeAll(collection) throws exception if any task does
*/
public void testAbnormalInvokeAllCollectionSingleton() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
FailingAsyncFib f = new FailingAsyncFib(8);
AsyncFib g = new AsyncFib(9);
AsyncFib h = new AsyncFib(7);
ForkJoinTask[] tasks = { f, g, h };
shuffle(tasks);
try {
invokeAll(Arrays.asList(tasks));
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(f, success);
}
}};
testInvokeOnPool(singletonPool(), a);
}
/**
* peekNextLocalTask returns most recent unexecuted task.
*/
public void testPeekNextLocalTask() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF g = new LCCF(9);
assertSame(g, g.fork());
CCF f = new LCCF(8);
assertSame(f, f.fork());
assertSame(f, peekNextLocalTask());
assertNull(f.join());
checkCompletedNormally(f);
helpQuiesce();
checkCompletedNormally(g);
}};
testInvokeOnPool(singletonPool(), a);
}
/**
* invokeAll(tasks) with > 2 argument throws exception if any task does
*/
public void testAbnormalInvokeAll3CC() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(null, 8);
FailingCCF g = new LFCCF(null, 9);
CCF h = new LCCF(null, 7);
try {
invokeAll(f, g, h);
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(g, success);
}
}};
checkInvoke(a);
}
private void testInvokeOnPool(ForkJoinPool pool, ForkJoinTask 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());
}
}
/**
* mainly use for {@link Stream#parallel()} with specific thread pool
* see https://stackoverflow.com/questions/21163108/custom-thread-pool-in-java-8-parallel-stream
*/
public static <R, X extends Throwable> R supplyParallel(ForkJoinPool pool,
ThrowableSupplier<R, X> func) throws X {
checkNotNull(pool);
Throwable[] throwable = { null };
ForkJoinTask<R> task = pool.submit(() -> {
try {
return func.get();
} catch (Throwable e) {
throwable[0] = e;
return null;
}
});
R r;
try {
r = task.get();
} catch (ExecutionException | InterruptedException impossible) {
throw new AssertionError(impossible);
}
if (throwable[0] != null) {
//noinspection unchecked
throw (X) throwable[0];
}
return r;
}
/**
* invokeAll(tasks) with > 2 argument throws exception if any task does
*/
public void testAbnormalInvokeAll3() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(8);
FailingCCF g = new LFCCF(9);
CCF h = new LCCF(7);
try {
invokeAll(f, g, h);
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(g, success);
}
}};
testInvokeOnPool(mainPool(), a);
}
protected void compute()
{
if(endValue - startValue > MAX_ELEMENTS_TO_PROCESS) // If there are too many elements to process in one operation...
{
if(ForkJoinTask.inForkJoinPool()) // ... and if we are in a ForkJoinPool ...
{
int halfWay = (endValue + startValue) / 2;
invokeAll(new ArrayDoubler(array, startValue, halfWay), new ArrayDoubler(array, halfWay, endValue));
return;
}
}
for(int i = startValue; i < endValue; i++) // If we aren't in a ForkJoinPool or if there are not a large number of elements to be processed
{
array[i] = array[i] * 2;
}
}
/**
* timed get of a forked task throws exception when task cancelled
*/
public void testCancelledForkTimedGet() throws Exception {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
CCF f = new LCCF(8);
assertTrue(f.cancel(true));
assertSame(f, f.fork());
try {
f.get(LONG_DELAY_MS, MILLISECONDS);
shouldThrow();
} catch (CancellationException success) {
checkCancelled(f);
}
}};
testInvokeOnPool(mainPool(), a);
}
/**
* invokeAll(collection) throws exception if any task does
*/
public void testAbnormalInvokeAllCollection() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
FailingCCF f = new LFCCF(8);
CCF g = new LCCF(9);
CCF h = new LCCF(7);
HashSet set = new HashSet();
set.add(f);
set.add(g);
set.add(h);
try {
invokeAll(set);
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(f, success);
}
}};
testInvokeOnPool(mainPool(), a);
}
void checkNotDone(ForkJoinTask a) {
assertFalse(a.isDone());
assertFalse(a.isCompletedNormally());
assertFalse(a.isCompletedAbnormally());
assertFalse(a.isCancelled());
assertNull(a.getException());
assertNull(a.getRawResult());
if (a instanceof BinaryAsyncAction)
assertTrue(((BinaryAsyncAction)a).getForkJoinTaskTag() == INITIAL_STATE);
try {
a.get(0L, SECONDS);
shouldThrow();
} catch (TimeoutException success) {
} catch (Throwable fail) { threadUnexpectedException(fail); }
}
/**
* get of a forked task throws exception when task cancelled
*/
public void testCancelledForkGet() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
CCF f = new LCCF(8);
assertTrue(f.cancel(true));
assertSame(f, f.fork());
try {
f.get();
shouldThrow();
} catch (CancellationException success) {
checkCancelled(f);
}
}};
testInvokeOnPool(mainPool(), a);
}
public void testAbnormalInvokeAll3(ForkJoinPool pool) {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
AsyncFib f = new AsyncFib(8);
FailingAsyncFib g = new FailingAsyncFib(9);
AsyncFib h = new AsyncFib(7);
ForkJoinTask[] tasks = { f, g, h };
shuffle(tasks);
try {
invokeAll(tasks[0], tasks[1], tasks[2]);
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(g, success);
}
}};
testInvokeOnPool(pool, a);
}
/**
* invokeAll(tasks) with > 2 argument invokes tasks
*/
public void testInvokeAll3() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(8);
CCF g = new LCCF(9);
CCF h = new LCCF(7);
invokeAll(f, g, h);
assertEquals(21, f.number);
assertEquals(34, g.number);
assertEquals(13, h.number);
checkCompletedNormally(f);
checkCompletedNormally(g);
checkCompletedNormally(h);
}};
testInvokeOnPool(mainPool(), a);
}
/**
* timed get of a forked task throws exception when task completes abnormally
*/
public void testAbnormalForkTimedGetCC() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
FailingCCF f = new LFCCF(null, 8);
assertSame(f, f.fork());
try {
f.get(LONG_DELAY_MS, MILLISECONDS);
shouldThrow();
} catch (ExecutionException success) {
Throwable cause = success.getCause();
assertTrue(cause instanceof FJException);
checkCompletedAbnormally(f, cause);
}
}};
checkInvoke(a);
}
/**
* invokeAll(t1, t2) throw exception if any task does
*/
public void testAbnormalInvokeAll2() {
RecursiveAction a = new CheckedRecursiveAction() {
protected void realCompute() {
AsyncFib f = new AsyncFib(8);
FailingAsyncFib g = new FailingAsyncFib(9);
ForkJoinTask[] tasks = { f, g };
shuffle(tasks);
try {
invokeAll(tasks);
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(g, success);
}
}};
testInvokeOnPool(mainPool(), a);
}
@Test
public void test() {
ForkJoinPool pool = new ForkJoinPool(2);
String homePath = System.getProperty("user.home");
FileCountTask task = new FileCountTask(homePath);
ForkJoinTask<Integer> result = pool.submit(task);
try {
Integer count = result.get();
System.out.println("file count = " + count);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
pool.shutdown();
while (!pool.isTerminated()) {
}
System.out.println("All thread finish...");
}
/**
* timed get of a forked task throws exception when task cancelled
*/
public void testCancelledForkTimedGetCC() throws Exception {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
CCF f = new LCCF(null, 8);
assertTrue(f.cancel(true));
assertSame(f, f.fork());
try {
f.get(LONG_DELAY_MS, MILLISECONDS);
shouldThrow();
} catch (CancellationException success) {
checkCancelled(f);
}
}};
checkInvoke(a);
}
/**
* get of a forked task throws exception when task cancelled
*/
public void testCancelledForkGet() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() throws Exception {
CCF f = new LCCF(8);
assertTrue(f.cancel(true));
assertSame(f, f.fork());
try {
f.get();
shouldThrow();
} catch (CancellationException success) {
checkCancelled(f);
}
}};
testInvokeOnPool(mainPool(), a);
}
/**
* join of a forked task throws exception when task cancelled
*/
public void testCancelledForkJoin() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(8);
assertTrue(f.cancel(true));
assertSame(f, f.fork());
try {
f.join();
shouldThrow();
} catch (CancellationException success) {
checkCancelled(f);
}
}};
testInvokeOnPool(mainPool(), a);
}
/**
* inForkJoinPool of non-FJ task returns false
*/
public void testInForkJoinPool2() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
assertFalse(inForkJoinPool());
}};
assertNull(a.invoke());
}
/**
* A task submitted after shutdown is rejected
*/
public void testSubmitAfterShutdown() {
ForkJoinPool p = new ForkJoinPool(1);
try (PoolCleaner cleaner = cleaner(p)) {
p.shutdown();
assertTrue(p.isShutdown());
try {
ForkJoinTask<Integer> f = p.submit(new FibTask(8));
shouldThrow();
} catch (RejectedExecutionException success) {}
}
}
/**
* @return
*/
public Collection<SimulationResult> computeWithoutMove() {
Collection<SimulationFeeder> feeders = SimulationFeeder.getFeedersFor(0, getStage(), possibleBlocks,
preferredCount);
Collection<SimulationTask> toRun = new SimulationCreationTask(this, null, feeders).invoke();
ForkJoinTask<SimulationResult> assembler = new SimulationResultsAssembler(null, processUUID, toRun, startTime)
.fork();
SimulationResult settleResult = assembler.join();
if (settleResult.getBoard().equals(board)) {
return null;
} else {
return Arrays.asList(settleResult);
}
}
/**
* invokeAll(tasks) with any null task throws NPE
*/
public void testInvokeAllNPE() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(8);
CCF g = new LCCF(9);
CCF h = null;
try {
invokeAll(f, g, h);
shouldThrow();
} catch (NullPointerException success) {}
}};
testInvokeOnPool(mainPool(), a);
}
/**
* invoke task throws exception when task completes abnormally
*/
public void testAbnormalInvoke() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
FailingCCF f = new LFCCF(8);
try {
f.invoke();
shouldThrow();
} catch (FJException success) {
checkCompletedAbnormally(f, success);
}
}};
testInvokeOnPool(mainPool(), a);
}
public static void main(String[] args) throws Exception {
if(args.length == 0) {
logger.error("please set config path");
System.exit(-1);
}
if(args.length > 1) {
TestSize.setClientSize(Integer.valueOf(args[1]));
}
TestConfig properties = new TestConfig(args[0]);
ForkJoinPool forkJoinPool = new ForkJoinPool(TestSize.getClientSize());
ForkJoinTask<?>[] list = new ForkJoinTask[TestSize.getClientSize()];
for (int i = 0; i < TestSize.getClientSize(); i++) {
int index = i;
ForkJoinTask<?> fork = forkJoinPool.submit(new Thread(new Runnable() {
@Override
public void run() {
MqttMutiClientSubTest handler = new MqttMutiClientSubTest();
String clientId = "client"+index;
String clientName = clientId+"Sub";
Topic[] topics = new Topic[] { new Topic(clientId+topic0, QoS.AT_MOST_ONCE),
new Topic(clientId+topic1, QoS.AT_LEAST_ONCE),
new Topic(clientId+topic2, QoS.EXACTLY_ONCE) };
handler.init(properties, topics, clientName, false);
logger.info(clientName+" testConn inited");
}
}));
list[i] = fork;
}
for (int i = 0; i < TestSize.getClientSize(); i++) {
list[i].join();
}
Thread.sleep(Integer.MAX_VALUE);
}
public static void createClientTest(TestConfig properties,String clientId) throws Exception {
MqttMutiClientSendTest handler = new MqttMutiClientSendTest();
String clientName = clientId+"Send";
handler.init(properties, null, clientName, false);
logger.info(clientName+" testConn inited");
ForkJoinPool forkJoinPool = new ForkJoinPool(TestSize.getThreadSize());
ForkJoinTask<?>[] list = new ForkJoinTask[TestSize.getThreadSize()];
for (int i = 0; i < TestSize.getThreadSize(); i++) {
ForkJoinTask<?> fork = forkJoinPool.submit(new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < TestSize.getMsgNums(); i++) {
try {
Thread.sleep(TestSize.getSleepTimes());
} catch (InterruptedException e) {
e.printStackTrace();
}
handler.send(clientId+"QOS" + (i % 3), (clientName + "::" + i).getBytes(),
QoS.values()[(i % 3)], false);
}
}
}));
list[i] = fork;
}
long start = System.currentTimeMillis();
for (int i = 0; i < TestSize.getThreadSize(); i++) {
list[i].join();
}
logger.info(clientName + " total time "+(System.currentTimeMillis() - start));
}
/**
* invokeAll(t1, t2) invokes all task arguments
*/
public void testInvokeAll2() {
ForkJoinTask a = new CheckedRecursiveAction() {
protected void realCompute() {
CCF f = new LCCF(8);
CCF g = new LCCF(9);
invokeAll(f, g);
assertEquals(21, f.number);
assertEquals(34, g.number);
checkCompletedNormally(f);
checkCompletedNormally(g);
}};
testInvokeOnPool(mainPool(), a);
}
@Advice.OnMethodEnter(suppress = Throwable.class)
public static void onExecute(@Advice.This Executor thiz,
@Advice.Argument(value = 0, readOnly = false) @Nullable ForkJoinTask<?> task) {
if (ExecutorInstrumentation.isExcluded(thiz)) {
return;
}
task = JavaConcurrent.withContext(task, tracer);
}