public SortTask(long[] array, int lo, int hi) { this.array = array; this.lo = lo; this.hi = hi; }
protected void compute() { if (hi - lo < THRESHOLD) sequentiallySort(array, lo, hi); else { int pivot = partition(array, lo, hi); coInvoke(new SortTask(array, lo, pivot - 1), new SortTask(array, pivot + 1, hi)); } }
private int partition(long[] array, int lo, int hi) { long x = array[hi]; int i = lo - 1; for (int j = lo; j < hi; j++) { if (array[j] <= x) { i++; swap(array, i, j); } } swap(array, i + 1, hi); return i + 1; }
private void swap(long[] array, int i, int j) { if (i != j) { long temp = array[i]; array[i] = array[j]; array[j] = temp; } }
private void sequentiallySort(long[] array, int lo, int hi) { Arrays.sort(array, lo, hi + 1); } }
public SortTask(long[] array, int lo, int hi) { this.array = array; this.lo = lo; this.hi = hi; }
protected void compute() { if (hi - lo < THRESHOLD) sequentiallySort(array, lo, hi); else { int pivot = partition(array, lo, hi); System.out.println("/npivot = " + pivot + ", low = " + lo + ", high = " + hi); System.out.println("array" + Arrays.toString(array)); coInvoke(new SortTask(array, lo, pivot - 1), new SortTask(array, pivot + 1, hi)); } }
private int partition(long[] array, int lo, int hi) { long x = array[hi]; int i = lo - 1; for (int j = lo; j < hi; j++) { if (array[j] <= x) { i++; swap(array, i, j); } } swap(array, i + 1, hi); return i + 1; }
private void swap(long[] array, int i, int j) { if (i != j) { long temp = array[i]; array[i] = array[j]; array[j] = temp; } }
private void sequentiallySort(long[] array, int lo, int hi) { Arrays.sort(array, lo, hi + 1); } }
public class TestForkJoinSimple { private static final int NARRAY = 16; //For demo only long[] array = new long[NARRAY]; Random rand = new Random();
@Before public void setUp() { for (int i = 0; i < array.length; i++) { array[i] = rand.nextLong()%100; //For demo only } System.out.println("Initial Array: " + Arrays.toString(array)); }
@Test public void testSort() throws Exception { ForkJoinTask sort = new SortTask(array); ForkJoinPool fjpool = new ForkJoinPool(); fjpool.submit(sort); fjpool.shutdown();
fjpool.awaitTermination(30, TimeUnit.SECONDS);
assertTrue(checkSorted(array)); }
boolean checkSorted(long[] a) { for (int i = 0; i < a.length - 1; i++) { if (a[i] > (a[i + 1])) { return false; } } return true; } }
public Integer compute() { if (n <= 10) { return compute(n); } Fibonacci f1 = new Fibonacci(n - 1); Fibonacci f2 = new Fibonacci(n - 2); System.out.println("fork new thread for " + (n - 1)); f1.fork(); System.out.println("fork new thread for " + (n - 2)); f2.fork(); return f1.join() + f2.join(); } }
class ConcurrentPrint extends RecursiveAction { protected void compute() { TaskBarrier b = new TaskBarrier() { protected boolean terminate(int cycle, int registeredParties) { System.out.println("Cycle is " + cycle + ";" + registeredParties + " parties"); return cycle >= 10; } }; int n = 3; CyclicAction[] actions = new CyclicAction[n]; for (int i = 0; i < n; ++i) { final int index = i; actions[i] = new CyclicAction(b) { protected void compute() { System.out.println("I'm working " + getCycle() + " " + index); try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } }; } for (int i = 0; i < n; ++i) actions[i].fork(); for (int i = 0; i < n; ++i) actions[i].join(); } }
public class TestForkJoin { @Test public void testBarrier () throws InterruptedException, ExecutionException { System.out.println("/ntesting Task Barrier ..."); ForkJoinTask fjt = new ConcurrentPrint(); ForkJoinPool fjpool = new ForkJoinPool(4); fjpool.submit(fjt); fjpool.shutdown(); }
@Test public void testFibonacci () throws InterruptedException, ExecutionException { System.out.println("/ntesting Fibonacci ..."); final int num = 14; //For demo only ForkJoinTask<Integer> fjt = new Fibonacci(num); ForkJoinPool fjpool = new ForkJoinPool(); Future<Integer> result = fjpool.submit(fjt);
// do something System.out.println("Fibonacci(" + num + ") = " + result.get()); } }
运行以上代码,我们可以得到以下结果:
testing Task Barrier ... I'm working 0 2 I'm working 0 0 I'm working 0 1 Cycle is 0; 3 parties I'm working 1 2 I'm working 1 0 I'm working 1 1 Cycle is 1; 3 parties I'm working 2 0 I'm working 2 1 I'm working 2 2 Cycle is 2; 3 parties I'm working 3 0 I'm working 3 2 I'm working 3 1 Cycle is 3; 3 parties I'm working 4 2 I'm working 4 0 I'm working 4 1 Cycle is 4; 3 parties I'm working 5 1 I'm working 5 0 I'm working 5 2 Cycle is 5; 3 parties I'm working 6 0 I'm working 6 2 I'm working 6 1 Cycle is 6; 3 parties I'm working 7 2 I'm working 7 0 I'm working 7 1 Cycle is 7; 3 parties I'm working 8 1 I'm working 8 0 I'm working 8 2 Cycle is 8; 3 parties I'm working 9 0 I'm working 9 2
testing Fibonacci ... fork new thread for 13 fork new thread for 12 fork new thread for 11 fork new thread for 10 fork new thread for 12 fork new thread for 11 fork new thread for 10 fork new thread for 9 fork new thread for 10 fork new thread for 9 fork new thread for 11 fork new thread for 10 fork new thread for 10 fork new thread for 9 Fibonacci(14) = 610