Learn java util concurrent part5

This is the last series of learning into java util concurrent package. If you have not read the series before, you can find part1, part2, part3 and part4 at the respectively links. In this series, we will study remaining 19 classes in java.util.concurrent package.

ForkJoinTask<V>

• Abstract base class for tasks that run within a ForkJoinPool.

1:        ForkJoinTask<Integer> fjt = ForkJoinTask.adapt(new Summer(44,55));  
2:        fjt.invoke();  
3:        Integer sum = fjt.get();  
4:        System.out.println(sum);  
5:        System.out.println(fjt.isDone());  
6:        fjt.join();  

Noticed that a new callable class was adapted into ForkJoinTask. The execution is commenced with invokeking the task. You can check if the task is complete using isDone method.

ForkJoinWorkerThread

• A thread managed by a ForkJoinPool, which executes ForkJoinTasks.

1:        ForkJoinWorkerThreadFactory customFactory = new ForkJoinWorkerThreadFactory() {  
2:           @Override  
3:           public ForkJoinWorkerThread newThread(ForkJoinPool pool) {  
4:              return null;  
5:           }  
6:        };  

As explained by ForkJoinWorkerThread javadoc, ForkJoinWorkerThreadFactory return a thread from the pool.

FutureTask<V>

• A cancellable asynchronous computation.
• A FutureTask can be used to wrap a Callable or Runnable object. Because FutureTask implements Runnable, a FutureTask can be submitted to an Executor for execution.

1:        FutureTask<Integer> ft = new FutureTask<Integer>(new Summer(66,77));  
2:        ft.run();  
3:        System.out.println(ft.get());  

If you have a long running task, you can use FutureTask so the task can be cancel. Next, we will go into another three queues.

LinkedBlockingDeque<E>

• An optionally-bounded blocking deque based on linked nodes.
• The capacity, if unspecified, is equal to Integer.MAX_VALUE.
• Most operations run in constant time (ignoring time spent blocking). Exceptions include remove, removeFirstOccurrence, removeLastOccurrence, contains, iterator.remove(), and the bulk operations, all of which run in linear time.

LinkedBlockingQueue<E>

• An optionally-bounded blocking queue based on linked nodes.
• This queue orders elements FIFO (first-in-first-out).
• The head of the queue is that element that has been on the queue the longest time.
• The tail of the queue is that element that has been on the queue the shortest time.
• New elements are inserted at the tail of the queue, and the queue retrieval operations obtain elements at the head of the queue.
• The capacity, if unspecified, is equal to Integer.MAX_VALUE.

LinkedTransferQueue<E>

• An unbounded TransferQueue based on linked nodes.
• This queue orders elements FIFO (first-in-first-out) with respect to any given producer.
• The head of the queue is that element that has been on the queue the longest time for some producer.
• the size method is NOT a constant-time operation.

1:        LinkedBlockingDeque<Integer> lbd = new LinkedBlockingDeque<Integer>();  
2:        lbd.add(1);  
3:        lbd.add(2);  
4:        lbd.add(3);  
5:    
6:        LinkedBlockingQueue<Integer> lbq = new LinkedBlockingQueue<Integer>();  
7:        lbq.add(4);  
8:        lbq.add(5);  
9:        lbq.add(6);  
10:          
11:        LinkedTransferQueue<Integer> ltq = new LinkedTransferQueue<Integer>();  
12:        ltq.add(7);  
13:        ltq.add(8);  
14:        ltq.add(9);  

Like the queues we talked about in part4, these queues will shown its benefits under multithreaded codes.

Phaser

• A reusable synchronization barrier, similar in functionality to CyclicBarrier and CountDownLatch but supporting more flexible usage.
• This implementation restricts the maximum number of parties to 65535.

1:        Phaser phaser = new Phaser();  
2:        phaser.register();  
3:        System.out.println("current phase number : " + phaser.getPhase());  
4:        testPhaser(phaser, 2000);  
5:        testPhaser(phaser, 4000);  
6:        testPhaser(phaser, 6000);  
7:          
8:        phaser.arriveAndDeregister();  
9:        Thread.sleep(10000);  
10:        System.out.println("current phase number : " + phaser.getPhase());  

PriorityBlockingQueue<E>

• An unbounded blocking queue that uses the same ordering rules as class PriorityQueue and supplies blocking retrieval operations.
• This class does not permit null elements.
• The Iterator provided in method iterator() is not guaranteed to traverse the elements of the PriorityBlockingQueue in any particular order

1:        PriorityBlockingQueue<Integer> pbq = new PriorityBlockingQueue<Integer>();  
2:        pbq.add(10);  
3:        pbq.add(11);  
4:        pbq.add(12);  

RecursiveAction

• A recursive resultless ForkJoinTask.

1:        long[] array = {1,3,2,5,4,9,5,7,8};  
2:        RecursiveAction ar = new SortTask(array);  
3:        ar.invoke();  
4:        System.out.println("array " + array[0]);  
5:        System.out.println("array " + array[1]);  
6:        System.out.println("array " + array[2]);  
7:        System.out.println("array " + array[3]);  
8:        System.out.println("array " + array[4]);  
9:        System.out.println("array " + array[5]);  
10:        System.out.println("array " + array[6]);  
11:        System.out.println("array " + array[7]);  
12:        System.out.println("array " + array[8]);  
13:    
14:     static class SortTask extends RecursiveAction {  
15:          
16:        final long[] array;  
17:        final int lo, hi;  
18:          
19:        SortTask(long[] array, int lo, int hi) {  
20:           this.array = array;  
21:           this.lo = lo;  
22:           this.hi = hi;  
23:        }  
24:          
25:        SortTask(long[] array) {  
26:           this(array, 0, array.length);  
27:        }  
28:    
29:        @Override  
30:        protected void compute() {  
31:           if (hi - lo < THRESHOLD)  
32:              sortSequentially(lo,hi);  
33:           else {  
34:              int mid = (lo + hi) >>> 1;  
35:              invokeAll(new SortTask(array, lo, mid), new SortTask(array, mid, hi));  
36:              merge(lo, mid, hi);  
37:           }  
38:        }  
39:          
40:        // implementation details follow:  
41:        static final int THRESHOLD = 1000;  
42:          
43:        void sortSequentially(int lo, int hi) {  
44:           Arrays.sort(array, lo, hi);  
45:        }  
46:          
47:        void merge(int lo, int mid, int hi) {  
48:           long[] buf = Arrays.copyOfRange(array, lo, mid);  
49:           for (int i = 0, j = lo, k = mid; i < buf.length; j++)  
50:              array[j] = (k == hi || buf[i] < array[k]) ? buf[i++] : array[k++];  
51:        }  
52:           
53:     }  

recursive sorting to the array by invoke commence to the object ar.

RecursiveTask<V>

• A recursive result-bearing ForkJoinTask.

1:        RecursiveTask<Integer> fibo = new Fibonacci(10);  
2:        fibo.invoke();  
3:        System.out.println(fibo.get());  
4:          
5:     static class Fibonacci extends RecursiveTask<Integer> {  
6:          
7:        final int n;  
8:    
9:        Fibonacci(int n) {  
10:           this.n = n;  
11:        }  
12:    
13:        protected Integer compute() {  
14:           if (n <= 1)  
15:              return n;  
16:           Fibonacci f1 = new Fibonacci(n - 1);  
17:           f1.fork();  
18:           Fibonacci f2 = new Fibonacci(n - 2);  
19:           return f2.compute() + f1.join();  
20:        }  
21:     }  

ScheduledThreadPoolExecutor

• A ThreadPoolExecutor that can additionally schedule commands to run after a given delay, or to execute periodically.

1:        ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(10);  
2:        Future<Integer> total = stpe.submit(new Summer(88,99));  
3:        System.out.println(total.get());  
4:        stpe.shutdown();  

Semaphore

• A counting semaphore.
• Semaphores are often used to restrict the number of threads than can access some (physical or logical) resource.

1:        ConnectionLimiter cl = new ConnectionLimiter(3);  
2:        URLConnection conn = cl.acquire(new URL("http://www.google.com"));  
3:        conn = cl.acquire(new URL("http://www.yahoo.com"));  
4:        conn = cl.acquire(new URL("http://www.youtube.com"));  
5:        cl.release(conn);  
6:          
7:     static class ConnectionLimiter {  
8:        private final Semaphore semaphore;  
9:          
10:        private ConnectionLimiter(int max) {  
11:           semaphore = new Semaphore(max);  
12:        }  
13:          
14:        public URLConnection acquire(URL url) throws IOException, InterruptedException {  
15:           semaphore.acquire();  
16:           return url.openConnection();  
17:        }  
18:          
19:        public void release(URLConnection conn) {  
20:           try {  
21:              // blahblah  
22:           } finally {  
23:              semaphore.release();  
24:           }  
25:        }  
26:     }  

SynchronousQueue<E>

• A blocking queue in which each insert operation must wait for a corresponding remove operation by another thread, and vice versa.
• This queue does not permit null elements.

1:        final SynchronousQueue<String> queue = new SynchronousQueue<String>();  
2:        Thread a = new Thread(new QueueProducer(queue));  
3:        a.start();  
4:        Thread b = new Thread(new QueueConsumer(queue));  
5:        b.start();  
6:          
7:        Thread.sleep(1000);  
8:          
9:        a.interrupt();  
10:        b.interrupt();  
11:          
12:          
13:     static class QueueProducer implements Runnable {  
14:          
15:        private SynchronousQueue<String> queue;  
16:    
17:        public QueueProducer(SynchronousQueue<String> queue) {  
18:           this.queue = queue;  
19:        }  
20:    
21:        @Override  
22:        public void run() {  
23:           String event = "SYNCHRONOUS_EVENT";  
24:           String another_event = "ANOTHER_EVENT";  
25:             
26:           try {  
27:              queue.put(event);  
28:              System.out.printf("[%s] published event : %s %n", Thread.currentThread().getName(), event);  
29:                
30:              queue.put(another_event);  
31:              System.out.printf("[%s] published event : %s %n", Thread.currentThread().getName(), another_event);  
32:           } catch (InterruptedException e) {  
33:           }  
34:             
35:        }  
36:          
37:     }  
38:       
39:     static class QueueConsumer implements Runnable {  
40:          
41:        private SynchronousQueue<String> queue;  
42:    
43:        public QueueConsumer(SynchronousQueue<String> queue) {  
44:           this.queue = queue;  
45:        }  
46:    
47:        @Override  
48:        public void run() {  
49:           try {  
50:              String event = queue.take();  
51:              // thread will block here  
52:              System.out.printf("[%s] consumed event : %s %n", Thread.currentThread().getName(), event);  
53:           } catch (InterruptedException e) {  
54:           }  
55:             
56:        }  
57:          
58:     }  

ThreadLocalRandom

• A random number generator isolated to the current thread.

1:        ThreadLocalRandom tlr = ThreadLocalRandom.current();  
2:        System.out.println(tlr.nextInt());  

ThreadPoolExecutor

• An ExecutorService that executes each submitted task using one of possibly several pooled threads, normally configured using Executors factory methods.

1:        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(4);  
2:        ThreadPoolExecutor tpe = new ThreadPoolExecutor(1, 1, 1, TimeUnit.SECONDS, blockingQueue);  

The last four classes are policies of the previous ThreadPoolExecutor which execute under specific condition.

ThreadPoolExecutor.AbortPolicy

• A handler for rejected tasks that throws a RejectedExecutionException.

ThreadPoolExecutor.CallerRunsPolicy

• A handler for rejected tasks that runs the rejected task directly in the calling thread of the execute method, unless the executor has been shut down, in which case the task is discarded.

ThreadPoolExecutor.DiscardOldestPolicy

• A handler for rejected tasks that discards the oldest unhandled request and then retries execute, unless the executor is shut down, in which case the task is discarded.

ThreadPoolExecutor.DiscardPolicy

• A handler for rejected tasks that silently discards the rejected task.

1:        ThreadPoolExecutor.AbortPolicy ap = new ThreadPoolExecutor.AbortPolicy();  
2:        try {  
3:        ap.rejectedExecution(() -> System.out.println("abort"), tpe);  
4:        } catch (Exception e) {  
5:             
6:        }  
7:          
8:        ThreadPoolExecutor.CallerRunsPolicy crp = new ThreadPoolExecutor.CallerRunsPolicy();  
9:        try {  
10:        crp.rejectedExecution(() -> System.out.println("run"), tpe);  
11:        } catch (Exception e) {  
12:             
13:        }  
14:          
15:        ThreadPoolExecutor.DiscardOldestPolicy dop = new ThreadPoolExecutor.DiscardOldestPolicy();  
16:        try {  
17:        dop.rejectedExecution(() -> System.out.println("abort"), tpe);  
18:        } catch (Exception e) {  
19:             
20:        }  
21:          
22:        ThreadPoolExecutor.DiscardPolicy dp = new ThreadPoolExecutor.DiscardPolicy();  
23:        try {  
24:        dp.rejectedExecution(() -> System.out.println("discard"), tpe);  
25:        } catch (Exception e) {  
26:             
27:        }  
28:    

That's it for these long learning series of java util concurrent.

Learn java util concurrent part4

This is yet another series of learning into java util concurrent package. If you have not read the series before, you can find part1, part2 and part3 at the respectively links. In this series, we will study the classes in java.util.concurrent package.

AbstractExecutorService
Provides default implementations of ExecutorService execution methods. This class implements the submit, invokeAny and invokeAll methods using a RunnableFuture returned by newTaskFor, which defaults to the FutureTask class provided in this package.

1:        AbstractExecutorService aes = null;  
2:          
3:        aes = new ForkJoinPool();  
4:        System.out.println(aes.isShutdown());  
5:        Future<Integer> total = aes.submit(new Summer(33, 44));  
6:        System.out.println(total.get());  
7:        aes.shutdown();  
8:          
9:          
10:        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(4);  
11:        aes = new ThreadPoolExecutor(1, 1, 1, TimeUnit.SECONDS, blockingQueue);  
12:        System.out.println(aes.isShutdown());  
13:        total = aes.submit(new Summer(33, 44));  
14:        System.out.println(total.get());  
15:        aes.shutdown();  

In the example above, we see that two concrete implementation of AbstractExecutorService, ForkJoinPool and ThreadPoolExecutor both invoking method submit from abstract class  AbstractExecutorService.

ArrayBlockingQueue
A bounded blocking queue backed by an array. This queue orders elements FIFO (first-in-first-out). The head of the queue is that element that has been on the queue the longest time. The tail of the queue is that element that has been on the queue the shortest time. New elements are inserted at the tail of the queue, and the queue retrieval operations obtain elements at the head of the queue.

1:        ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<Integer>(5);  
2:        abq.add(1);  
3:        abq.offer(2);  
4:        System.out.println(abq.size());  

A simple queue implementation, just like any other collections in java collection framwork, you can add, remove, or drain the collection.

CompletableFuture
A Future that may be explicitly completed (setting its value and status), and may be used as a CompletionStage, supporting dependent functions and actions that trigger upon its completion.

1:        CompletableFuture<Integer> cf = new CompletableFuture<Integer>();  
2:        System.out.println(cf.isCancelled());  

ConcurrentHashMap
A hash table supporting full concurrency of retrievals and high expected concurrency for updates.
However, even though all operations are thread-safe, retrieval operations do not entail locking, and there is not any support for locking the entire table in a way that prevents all access.
this class does not allow null to be used as a key or value.

1:        ConcurrentHashMap<String,Integer> chm = new ConcurrentHashMap<String,Integer>();  
2:        chm.put("one", 1);  
3:        chm.put("two", 2);  
4:        chm.put("six", 6);  

ConcurrentHashMap.KeySetView<K,V>
A view of a ConcurrentHashMap as a Set of keys, in which additions may optionally be enabled by mapping to a common value.

1:        ConcurrentHashMap.KeySetView<String, Integer> keys = chm.keySet(10);  
2:        System.out.println(keys.isEmpty());  
3:        System.out.println(keys.toString());  
4:        keys.add("ten");  
5:          
6:        keys.forEach((s) -> System.out.println(s));  
7:        System.out.println(chm.toString());  
8:          
9:        ConcurrentHashMap.KeySetView<String, Boolean> keys1 = chm.newKeySet();  
10:        System.out.println(keys1.isEmpty());  
11:        System.out.println(keys1.toString());  
12:        keys1.add("four");  
13:          
14:        keys1.forEach((s) -> System.out.println(s));  
15:        System.out.println(chm.toString());  

The above give two examples of usage of ConcurrentHashMap.KeySetView. The first one notice that changes to the keys affect the original concurrentHashMap chm whilst the second does not. So read the javadoc and pick the implementation that suit your requirements.

Now, we will take a look at two for the concurrent linked queues.

ConcurrentLinkedDeque<E>

• An unbounded concurrent deque based on linked nodes.
• Concurrent insertion, removal, and access operations execute safely across multiple threads. 
• this class does not permit the use of null elements.
• the size method is NOT a constant-time operation. Because of the asynchronous nature of these deques, determining the current number of elements requires a traversal of the elements, and so may report inaccurate results if this collection is modified during traversal.

ConcurrentLinkedQueue<E>

• An unbounded thread-safe queue based on linked nodes.
• This queue orders elements FIFO (first-in-first-out).
• The head of the queue is that element that has been on the queue the longest time. 
• The tail of the queue is that element that has been on the queue the shortest time. 
• this class does not permit the use of null elements.

  insert                              element  
  always                              oldest  
  at tail                             at head  
  and                                 and retrieve  
  youngest                            here  
    +------------------------------------+  
    |                                    |  
    |                                    |  
    +------------------------------------+  
   tail                               head  

1:    
2:        ConcurrentLinkedDeque<Integer> cldq = new ConcurrentLinkedDeque<Integer>();   
3:        cldq.add(1);  
4:        cldq.add(2);  
5:        cldq.add(3);  
6:          
7:    
8:        ConcurrentLinkedQueue<Integer> clq = new ConcurrentLinkedQueue<Integer>();  
9:        clq.add(4);  
10:        clq.add(5);  
11:        clq.add(6);  

With the examples above, the apparent benefits is not actually express as the only thread is the main adding element to the queues serially. Like the javadoc mentioned, you should really use these queues on multithreaded situation.

ConcurrentSkipListMap<K,V>

• A scalable concurrent ConcurrentNavigableMap implementation.
• The map is sorted according to the natural ordering of its keys, or by a Comparator provided at map creation time, depending on which constructor is used.
• providing expected average log(n) time cost for the containsKey, get, put and remove operations and their variants. 
• Insertion, removal, update, and access operations safely execute concurrently by multiple threads.
• Ascending key ordered views and their iterators are faster than descending ones.
• the size method is not a constant-time operation.

ConcurrentSkipListSet<E>

• A scalable concurrent NavigableSet implementation based on a ConcurrentSkipListMap.
• The elements of the set are kept sorted according to their natural ordering, or by a Comparator provided at set creation time, depending on which constructor is used.
• expected average log(n) time cost for the contains, add, and remove operations and their variants.
• Insertion, removal, and access operations safely execute concurrently by multiple threads.
• Ascending ordered views and their iterators are faster than descending ones.
• the size method is not a constant-time operation. 

1:        ConcurrentSkipListMap<String,Integer> cslm = new ConcurrentSkipListMap<String, Integer>();  
2:        cslm.put("one", 1);  
3:        cslm.put("two", 2);  
4:        cslm.put("six", 6);  
5:          
6:        ConcurrentSkipListSet<Integer> csls = new ConcurrentSkipListSet<Integer>();  
7:        csls.add(1);  
8:        csls.add(1);  
9:        System.out.println("set size " + csls.size());  

With the examples above, the apparent benefits is not actually express as the only thread is the main adding element to the collections serially. Like the javadoc mentioned, you should really use these collections on multithreaded situation.

CopyOnWriteArrayList<E>

• A thread-safe variant of ArrayList in which all mutative operations (add, set, and so on) are implemented by making a fresh copy of the underlying array.
• This is ordinarily too costly, but may be more efficient than alternatives when traversal operations vastly outnumber mutations, and is useful when you cannot or don't want to synchronize traversals, yet need to preclude interference among concurrent threads.
• All elements are permitted, including null.

CopyOnWriteArraySet<E>

• A Set that uses an internal CopyOnWriteArrayList for all of its operations.
• It is best suited for applications in which set sizes generally stay small, read-only operations vastly outnumber mutative operations, and you need to prevent interference among threads during traversal.
• It is thread-safe.
• Mutative operations (add, set, remove, etc.) are expensive since they usually entail copying the entire underlying array.

1:        CopyOnWriteArrayList<Integer> cowal = new CopyOnWriteArrayList<Integer>();  
2:        cowal.add(1);  
3:        cowal.add(2);  
4:        cowal.add(3);  
5:          
6:        CopyOnWriteArraySet<Integer> cowas = new CopyOnWriteArraySet<Integer>();  
7:        cowas.add(1);  
8:        cowas.add(1);  
9:        System.out.println("set size " + cowas.size());  

Just like the four collections above, these beneifts best shown on multithreaded applications.

CountDownLatch

• A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.
• A CountDownLatch is initialized with a given count. The await methods block until the current count reaches zero due to invocations of the countDown() method, after which all waiting threads are released and any subsequent invocations of await return immediately. This is a one-shot phenomenon -- the count cannot be reset

1:        int N = 10;  
2:        CountDownLatch startSignal = new CountDownLatch(1);  
3:        CountDownLatch doneSignal = new CountDownLatch(N);  
4:          
5:        for (int i = 0; i < N; ++i) // create and start threads  
6:           new Thread(new Worker(startSignal, doneSignal)).start();  
7:    
8:        doSomethingElse();     // don't let run yet  
9:        startSignal.countDown();  // let all threads proceed  
10:        doSomethingElse();  
11:        doneSignal.await();    // wait for all to finish  
12:    
13:    
14:     private static void doSomethingElse() throws InterruptedException {  
15:          Thread.sleep(3000);  
16:        System.out.println(Thread.currentThread().getName() + " doing something else");  
17:    
18:     }  
19:    
20:    
21:     static class Worker implements Runnable {  
22:          private final CountDownLatch startSignal;  
23:          private final CountDownLatch doneSignal;  
24:          Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {  
25:           this.startSignal = startSignal;  
26:           this.doneSignal = doneSignal;  
27:          }  
28:          public void run() {  
29:           try {  
30:            startSignal.await();  
31:            doWork();  
32:            doneSignal.countDown();  
33:           } catch (InterruptedException ex) {} // return;  
34:          }  
35:    
36:          void doWork() { System.out.println(Thread.currentThread().getName() + " doing work"); try {  
37:           Thread.sleep(200);  
38:      }  

we see that ten worker threads were started but it was in waiting state in the run method. Until the startSignal started to count down, then only all the workers thread started. In the individual worker threads, we will see doneSignal is counting down one by one for 10 tens for each worker thread respectively. In the main thread, doneSignal is in waiting state before all the worker thread done all the signals.

CountedCompleter<T>

• A ForkJoinTask with a completion action performed when triggered and there are no remaining pending actions.
• Sample Usages.
• Parallel recursive decomposition.
• Searching. 
• Recording subtasks. 
• Completion Traversals. 
• Triggers.

1:        // CountedCompleter<T>  
2:        Integer[] numbers = {1,2,3,4,5};  
3:        // null ?  
4:        MapReducer<Integer> numbersReducer = new MapReducer<Integer>(null, numbers, new MyMapper(), new MyReducer(), 1, 10);  
5:        Integer result = numbersReducer.getRawResult();  
6:        System.out.println(result);  

CyclicBarrier

• A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point.
• CyclicBarriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. 
• The barrier is called cyclic because it can be re-used after the waiting threads are released. 
• The CyclicBarrier uses an all-or-none breakage model for failed synchronization attempts: If a thread leaves a barrier point prematurely because of interruption, failure, or timeout, all other threads waiting at that barrier point will also leave abnormally via BrokenBarrierException (or InterruptedException if they too were interrupted at about the same time). 

1:        // CyclicBarrier  
2:        float[][] matrix = {{1,2}, {2,3}};  
3:        new Solver(matrix);  
4:    
5:  public class Solver {  
6:    
7:     final int N;  
8:     final float[][] data;  
9:     final CyclicBarrier barrier;  
10:    
11:     class Worker implements Runnable {  
12:        int myRow;  
13:        boolean done;  
14:    
15:        Worker(int row) {  
16:           myRow = row;  
17:        }  
18:    
19:        public void run() {  
20:           while (!done()) {  
21:              processRow(myRow);  
22:    
23:              try {  
24:                 barrier.await();  
25:              } catch (InterruptedException ex) {  
26:                 return;  
27:              } catch (BrokenBarrierException ex) {  
28:                 return;  
29:              }  
30:           }  
31:        }  
32:          
33:        public boolean done() {  
34:           return done;  
35:        }  
36:          
37:        private void processRow(int row) {  
38:           System.out.println(Thread.currentThread().getName() + " processing row " + row );  
39:           done = true;  
40:        }  
41:     }  
42:    
43:     public Solver(float[][] matrix) {  
44:        data = matrix;  
45:        N = matrix.length;  
46:        Runnable barrierAction = new Runnable() {   
47:           public void run() {   
48:              //mergeRows(...);   
49:              System.out.println("merging row");  
50:           }  
51:        };  
52:        barrier = new CyclicBarrier(N, barrierAction);  
53:    
54:        List<Thread> threads = new ArrayList<Thread>(N);  
55:        for (int i = 0; i < N; i++) {  
56:         Thread thread = new Thread(new Worker(i));  
57:         threads.add(thread);  
58:         thread.start();  
59:        }  
60:    
61:        // wait until done  
62:        for (Thread thread : threads)  
63:           try {  
64:              thread.join();  
65:           } catch (InterruptedException e) {  
66:              e.printStackTrace();  
67:           }  
68:       }  
69:  }  

In the example above, we see the main class initialized a new solver object passing a two dimentional floating matrix for process. In the solver class, we see that a cyclicbarrier is initialized with a runnable barrier action. Depending on the matrix length, the length shall be used to initialize the workers threads. The solver thread waits until all the workers thread done.

In the worker thread, I simplified the processRow to just printout and set done to true, you can of cause process data[myRow] to make the sample code near to the real world problem. Noticed that barrier in each individual worker thread is call method await. IN this example, if two workers are done process the row and barrier await is executed, then the final barrierAction object will run the final merging row.


DelayQueue<E extends Delayed>

• An unbounded blocking queue of Delayed elements, in which an element can only be taken when its delay has expired.
• The head of the queue is that Delayed element whose delay expired furthest in the past.
• If no delay has expired there is no head and poll will return null.
• Expiration occurs when an element's getDelay(TimeUnit.NANOSECONDS) method returns a value less than or equal to zero.
• the size method returns the count of both expired and unexpired elements.
• This queue does not permit null elements.

1:        DelayQueue<SalaryDelay> delayQueue = new DelayQueue<SalaryDelay>();  
2:        delayQueue.add(new SalaryDelay("August", 1));  
3:        delayQueue.add(new SalaryDelay("September", 2));  
4:        delayQueue.add(new SalaryDelay("October", 3));  
5:          
6:        System.out.println(delayQueue.size());  
7:        System.out.println(delayQueue.poll());  

Like this queue before, you add the class that implement delayed and be place in this delayqueue.

Exchanger<V>

• A synchronization point at which threads can pair and swap elements within pairs.

1:        Exchanger<?> exchanger = new Exchanger<>();  
2:        ExchangerRunnable exchangerRunnable1 = new ExchangerRunnable(exchanger, "keychain");  
3:        ExchangerRunnable exchangerRunnable2 = new ExchangerRunnable(exchanger, "chocalate");  
4:          
5:        new Thread(exchangerRunnable1).start();  
6:        new Thread(exchangerRunnable2).start();  
7:          
8:  public class ExchangerRunnable implements Runnable {  
9:    
10:     Exchanger exchanger = null;  
11:     Object object = null;  
12:    
13:     public ExchangerRunnable(Exchanger exchanger, Object object) {  
14:        this.exchanger = exchanger;  
15:        this.object = object;  
16:     }  
17:    
18:     public void run() {  
19:        try {  
20:           Object previous = this.object;  
21:    
22:           this.object = this.exchanger.exchange(this.object);  
23:    
24:           System.out.println(Thread.currentThread().getName() + " exchanged "  
25:                 + previous + " for " + this.object);  
26:        } catch (InterruptedException e) {  
27:           e.printStackTrace();  
28:        }  
29:     }  
30:    
31:  }  

With the code above, there are two thread that exchange a string object to each other.

ExecutorCompletionService<V>

• A CompletionService that uses a supplied Executor to execute tasks.

1:        ExecutorService executorService = Executors.newFixedThreadPool(1);  
2:        CompletionService<Integer> longRunningCompletionService = new ExecutorCompletionService<Integer>(executorService);  
3:        longRunningCompletionService.submit(() -> {System.out.println("done"); return 1;});  
4:        longRunningCompletionService.take();  
5:        executorService.shutdown();  

Moving onto our last 2 classes in this lesson.

Executors
- Factory and utility methods for Executor, ExecutorService, ScheduledExecutorService, ThreadFactory, and Callable classes defined in this package.

1:        Executors.newCachedThreadPool();  
2:        Executors.defaultThreadFactory();  
3:        Executors.newFixedThreadPool(10);  
4:        Executors.newScheduledThreadPool(1);  
5:        Executors.newSingleThreadExecutor();  
6:        Executors.privilegedThreadFactory();  
7:        Executors.newWorkStealingPool();  

Just try different pools implementation in java to get some idea the specific pools.

ForkJoinPool

• An ExecutorService for running ForkJoinTasks.
• This implementation restricts the maximum number of running threads to 32767. 

1:        ForkJoinPool fjPool = new ForkJoinPool();  
2:        Future<Integer> sum = fjPool.submit(new Summer(11, 89));  
3:        System.out.println(sum.get());  
4:        fjPool.shutdown();  

A trivial example of using ForkJoinPool to submit a runnable task which return a result.

That's it for this long but brief lesson. This article end with the source code where you can get from the links below.

https://github.com/jasonwee/videoOnCloud/commit/0e7549249490ef5f9e528e101354440b7ad5cbdb
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/ExchangerRunnable.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/LearnConcurrentClassP3.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/MapReducer.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/MyMapper.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/MyReducer.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/Solver.java

Learn java util concurrent part3

This series is the next learning series of java.util.concurrent. You should read the first part and second part too. Today we will learn remaining ten interfaces in package java.util.concurrent.

Okay, let's start with ForkJoinPool.ForkJoinWorkerThreadFactory. For programming wise, you should not be worry as the class ForkJoinPool takes care of this implementation. Within class ForkJoinPool, we see that, there are two classes

• DefaultForkJoinWorkerThreadFactory
• InnocuousForkJoinWorkerThreadFactory

which implement ForkJoinWorkerThreadFactory with access modifier to default. So unless you know what you want and you know how ForkJoinPool work, then subclass ForkJoinPool away. For beginner in this article, it is sufficient to just use ForkJoinPool.

ManagedBlocker is an interface for extending managed parallelism for tasks running in ForkJoinPools. There are two methods to be implemented, block() and isReleasable()

1:  public class QueueManagedBlocker<T> implements ManagedBlocker {  
2:       
3:     final BlockingQueue<T> queue;  
4:     volatile T value = null;  
5:       
6:     QueueManagedBlocker(BlockingQueue<T> queue) {  
7:        this.queue = queue;  
8:     }  
9:    
10:     @Override  
11:     public boolean block() throws InterruptedException {  
12:        if (value == null)  
13:           value = queue.take();  
14:        return true;  
15:     }  
16:    
17:     @Override  
18:     public boolean isReleasable() {  
19:        return value != null || (value = queue.poll()) != null;  
20:     }  
21:       
22:     public T getValue() {  
23:        return value;  
24:     }  
25:    
26:  }  

Next, we have interface Future<V> which we have see before in the previous learning series.

1:  ExecutorService executorService = Executors.newFixedThreadPool(1);  
2:  Future<Integer> future = executorService.submit(new Summer(11,22));  

It's very clear you can obtain the result via future variable above. Interface RejectedExecutionHandler is mostly for error handling.

1:  RejectedExecutionHandler executionHandler = new MyRejectedExecutionHandlerImpl();  
2:  ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 3, 10, TimeUnit.SECONDS, worksQueue, executionHandler);  
3:    
4:  public class MyRejectedExecutionHandlerImpl implements RejectedExecutionHandler {  
5:    
6:     @Override  
7:     public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {  
8:        System.out.println(r.toString() + " : I've been rejected !");  
9:     }  
10:    
11:  }  

So you can set the implementation classed to ThreadPoolExecutor and if a task cannot be executor by the ThreadPoolExecutor, rejectedExecution will be executed. Moving onto the next interface, RunnableFuture<V> .

1:  RunnableFuture<Integer> rf = new FutureTask<Integer>(new Summer(22,33));  

so we see an initialization of object FutureTask with a callable class Summer class which we created in the previous learning series. Interface RunnableScheduledFuture which extend the previous interface RunnableFuture has another additional method to implement upon on.

1:  RunnableScheduledFuture<Integer> rsf = new Summer1();  
2:  System.out.println(rsf.isPeriodic());RunnableFuture<Integer> rf = new FutureTask<Integer>(new Summer(22,33));  

In the class Summer1, you should determine if the class is periodic or not. ScheduledExecutorService is pretty common if you google this interface and given the code below.

1:  ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);  
2:  scheduler.scheduleAtFixedRate(() -> System.out.println("hihi"), 1, 1, TimeUnit.SECONDS);  
3:  Thread.sleep(3000);  
4:  scheduler.shutdown();  

so we see a thread is executed every second.

1:  ScheduledFuture<Integer> sf = new ScheduledFutureImpl();  
2:  sf.isCancelled();  

ScheduledFuture<V> is a delayed result-bearing action that can be cancelled. Usually a scheduled future is the result of scheduling a task with a ScheduledExecutorService. This class is pretty common if you have a future task which get delay for whatever reason or it may get cancel, you want to look further into this class.

ThreadFactory is another interface which creates new threads on demand. Using thread factories removes hardwiring of calls to new Thread, enabling applications to use special thread subclasses, priorities, etc.

1:  ThreadFactory tf = Executors.defaultThreadFactory();  
2:  tf.newThread(()->System.out.println("ThreadFactory")).start();  

In this last series, we take a look at the last interface, TransferQueue.  A TransferQueue may be useful for example in message passing applications in which producers sometimes (using method transfer(E)) await receipt of elements by consumers invoking take or poll, while at other times enqueue elements (via method put) without waiting for receipt.

1:  TransferQueue<Integer> tq = new LinkedTransferQueue<Integer>();  

That's it for this learning series. Thank you. Oh, and the source code.

https://github.com/jasonwee/videoOnCloud/commit/ce479e5befaf7abe84d3d85930d5196a639e2643

https://github.com/jasonwee/videoOnCloud/blob/master/src/java/org/just4fun/concurrent/MyRejectedExecutionHandlerImpl.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/org/just4fun/concurrent/ExampleThreadPoolExecutor.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/org/just4fun/concurrent/DemoExecutor.java

https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/LearnConcurrentInterfaceP2.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/QueueManagedBlocker.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/ScheduledFutureImpl.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/Summer1.java

Learn java util concurrent part2

This series is the next learning series of java.util.concurrent. You should read the first part here too. Today we will learn ten interfaces in package java.util.concurrent.

Okay, let's start on the queue interface, BlockingDeque. Some characteristics of this interface including

• blocking
• thread safe
• does not permit null elements
• may (or may not) be capacity-constrained.

and we can do adding/removing item to this queue. Example below.

1:  bd.add(1);  
2:  bd.add(2);  
3:  System.out.println("size: " + bd.size());  
4:          
5:  bd.add(3);  
6:  System.out.println("size: " + bd.size());  
7:  //bd.add(4); // exception  
8:          
9:  bd.forEach(s -> System.out.println(s));  

Try play around this class with different methods to get a basic understanding on it. Next, we have a similar queue called BlockingQueue. It's characters same as BlockingDeque, not sure where is the different. But official java has many classes implement this BlockingQueue.

1:  BlockingQueue<Integer> bq = new ArrayBlockingQueue<Integer>(10);  
2:  bq = new DelayQueue();  
3:  bq = new LinkedBlockingDeque<Integer>();  
4:  bq = new LinkedTransferQueue<Integer>();  
5:  bq = new PriorityBlockingQueue<Integer>();  
6:  bq = new SynchronousQueue<Integer>();  

Next we have Callable interface which is similar to Runnable with a clear distinction. Callable return a value.

1:  ExecutorService executorService = Executors.newFixedThreadPool(1);  
2:    
3:  Future<Integer> future = executorService.submit(new Summer(11,22));  
4:    
5:  try {  
6:     Integer total = future.get();  
7:     System.out.println("sum " + total);  
8:  } catch (Exception e) {  
9:     e.printStackTrace();  
10:  }  
11:    

As can be read above, Summer is a implementation of interface Callable and it is submitted to an executor service to be execute upon on. CompletableFuture.AsynchronousCompletionTask is a interesting interface and its official documentation said "A marker interface identifying asynchronous tasks produced by async methods. This may be useful for monitoring, debugging, and tracking asynchronous activities."

1:  CompletableFuture<Integer> cf = new CompletableFuture<Integer>();  
2:  ForkJoinPool.commonPool().submit(  
3:        (Runnable & CompletableFuture.AsynchronousCompletionTask)()->{  
4:      try {  
5:         cf.complete(1);  
6:      } catch (Exception e) {  
7:         cf.completeExceptionally(e);  
8:      }  
9:   });  

As can be read above, we submit a anonymous function to the ForkJoinPool where this anonymous function cast into the intersection of interface Runnable and CompletableFuture.AsynchronousCompletionTask. Moving on, we have Interface CompletionService. Now if you have a long running service, you might want to look into this interface. Example as can be read below.

1:  CompletionService<Integer> longRunningCompletionService = new ExecutorCompletionService<Integer>(executorService);  
2:    
3:  longRunningCompletionService.submit(() -> {System.out.println("done"); return 1;});  
4:    
5:  try {  
6:     Future<Integer> result = longRunningCompletionService.take();  
7:     System.out.println(result.get());  
8:  } catch (Exception e) {  
9:     // TODO Auto-generated catch block  
10:     e.printStackTrace();  
11:  }  

Persist the object longRunningCompletionService throughout your application and the result can be retrieve in the future. Pretty handy. Moving on, we have a new Interface CompletionStage which debut on jdk8. From CompletionStage javadoc, A stage of a possibly asynchronous computation, that performs an action or computes a value when another CompletionStage completes. A stage completes upon termination of its computation, but this may in turn trigger other dependent stages.

Example code using CompletionStage as of following.

1:  ListenableFuture<String> springListenableFuture = createSpringListenableFuture();  
2:    
3:  CompletableCompletionStage<Object> completionStage = factory.createCompletionStage();  
4:  springListenableFuture.addCallback(new ListenableFutureCallback<String>() {  
5:    @Override  
6:    public void onSuccess(String result) {  
7:       System.out.println("onSuccess called");  
8:      completionStage.complete(result);  
9:    }  
10:    @Override  
11:    public void onFailure(Throwable t) {  
12:       System.out.println("onFailure called");  
13:      completionStage.completeExceptionally(t);  
14:    }  
15:  });  
16:    
17:  completionStage.thenAccept(System.out::println);  

Until here, if you don't understand, you should start to take the code and start to work on it. In this concurrent package, we have two Maps to use, that is ConcurrentMap and ConcurrentNavigableMap.

1:  ConcurrentMap<String, String> cm = new ConcurrentHashMap();  
2:  cm = new ConcurrentSkipListMap<String, String>();  
3:    
4:  ConcurrentNavigableMap<String, String> cnm = new ConcurrentSkipListMap<String, String>();  

ConcurrentMap providing thread safety and atomicity guarantees whilst ConcurrentNavigableMap support additional supporting NavigableMap operations, and recursively so for its navigable sub-maps. Then we have interface Delayed.

1:  Random random = new Random();  
2:  int delay = random.nextInt(10000);  
3:  Delayed employer = new SalaryDelay("a lot of bs reasons", delay);  
4:  System.out.println("bullshit delay this time " + employer.getDelay(TimeUnit.SECONDS));  

Delayed is an interface where you should implemented two require methods, getDelay and compareTo. As you can read fundamentally is you have a few object which feed to the executor to process upon but not immmediate, it may get delay for any reasons. As an example, pretty common in the working world, emplayer delay salary for any reasons.

Last two interfaces is related to each other where ExecutorService is a sub interface of Executor. For ExecutorService, we have seen an example above and below is for Executor,

1:  Executor executor = new ForkJoinPool();  
2:  executor = new ScheduledThreadPoolExecutor(1);  
3:    
4:  BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<Runnable>(4);  
5:  executor = new ThreadPoolExecutor(1, 1, 1000, TimeUnit.SECONDS, blockingQueue);  

We see there are many classes implemented Executor interface. Executor interface guaranteed

• An object that executes submitted Runnable tasks. 
• Executor interface does not strictly require that execution be asynchronous

That's it for this article, we continue the rest in the next article!

Oh before that, you can download the full source at the follow links

https://github.com/jasonwee/videoOnCloud/commit/3d291610df89c610215b8ecdbcc59cbb028ba14b
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/LearnConcurrentInterfaceP1.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/SalaryDelay.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/concurrent/Summer.java
https://github.com/jasonwee/videoOnCloud/blob/master/src/java/play/learn/java/CompletableFuture/LearnCompletionStage.java

Learn java util concurrent part1

Today we are going to learn classes in java package java.util.concurrent. Because there are many classes within java.util.concurrent package, there will be several articles covering classes in this package. Let's start with a simple class first, TimeUnit.

1:  package play.learn.java.concurrent;  
2:    
3:  import java.util.concurrent.BrokenBarrierException;  
4:  import java.util.concurrent.TimeUnit;  
5:    
6:  public class LearnTimeUnit {  
7:       
8:     public LearnTimeUnit() throws InterruptedException {  
9:          
10:        // assuming we have a long running apps which ran for 2 days 7hours 35minutes 6 seconds   
11:        long longRunningApplicationDuration = 200102l;  
12:          
13:        System.out.println("duration in nanos " + TimeUnit.SECONDS.toNanos(longRunningApplicationDuration));  
14:        System.out.println("duration in days " + TimeUnit.SECONDS.toDays(longRunningApplicationDuration));  
15:        System.out.println("duration in hours " + TimeUnit.SECONDS.toHours(longRunningApplicationDuration));  
16:        System.out.println("duration in micros " + TimeUnit.SECONDS.toMicros(longRunningApplicationDuration));  
17:        System.out.println("duration in millis " + TimeUnit.SECONDS.toMillis(longRunningApplicationDuration));  
18:        System.out.println("duration in minutes " + TimeUnit.SECONDS.toMinutes(longRunningApplicationDuration));  
19:        System.out.println("duration in seconds " + TimeUnit.SECONDS.toSeconds(longRunningApplicationDuration));  
20:          
21:          
22:        TimeUnit[] var = TimeUnit.values();  
23:        System.out.println("size " + var.length);  
24:          
25:        for (TimeUnit elem : var) {  
26:           System.out.println(elem.name());  
27:        }  
28:          
29:        TimeUnit.SECONDS.sleep(10);  
30:     }  
31:    
32:     public static void main(String[] args) throws InterruptedException {  
33:        new LearnTimeUnit();  
34:     }  
35:    
36:  }  

TimeUnit provides several helpful methods to convert the time to different unit. You can also download the above source code here.

Next, we will take a look at concurrent exceptions. This exception will become meaningful when we try catch it in the class. For now, we will go through the definition to get a basic understanding of them. Below is a summarization.

BrokenBarrierException
Exception thrown when a thread tries to wait upon a barrier that is in a broken state, or which enters the broken state while the thread is waiting.

CancellationException
Exception indicating that the result of a value-producing task, such as a FutureTask, cannot be retrieved because the task was cancelled.

CompletionException
Exception thrown when an error or other exception is encountered in the course of completing a result or task.

ExecutionException
Exception thrown when attempting to retrieve the result of a task that aborted by throwing an exception. This exception can be inspected using the Throwable.getCause() method.

RejectedExecutionException
Exception thrown by an Executor when a task cannot be accepted for execution.

TimeoutException
Exception thrown when a blocking operation times out. Blocking operations for which a timeout is specified need a means to indicate that the timeout has occurred. For many such operations it is possible to return a value that indicates timeout; when that is not possible or desirable then TimeoutException should be declared and thrown.


BrokenBarrierException example, for full source code, you can download it here.

1:  package play.learn.java.concurrent;  
2:    
3:  import java.util.concurrent.BrokenBarrierException;  
4:  import java.util.concurrent.CyclicBarrier;  
5:    
6:  public class LearnBrokenBarrierException {  
7:       
8:     private CyclicBarrier cibai;  
9:     public static int count = 0;  
10:       
11:     private void manageThread() {  
12:        cibai = new CyclicBarrier(3);  
13:          
14:        for (int i = 0; i < 3; i++) {  
15:           new Thread(new Worker(cibai)).start();  
16:        }  
17:     }  
18:       
19:     public static void barrierComplete(CyclicBarrier cb) {  
20:        System.out.println("collating task");  
21:          
22:        if (count == 3) {  
23:           System.out.println("Exit from system");  
24:           // comment for finite  
25:           System.exit(0);  
26:        }  
27:        count++;  
28:          
29:        for (int i = 0; i < 3; i++) {  
30:        new Thread(new Worker(cb)).start();  
31:        }  
32:     }  
33:       
34:     public static void main(String[] args) {  
35:        new LearnBrokenBarrierException().manageThread();   
36:     }  
37:       
38:     static class Worker implements Runnable {  
39:          
40:        CyclicBarrier cibai;  
41:          
42:        public Worker(CyclicBarrier cb) {  
43:           this.cibai = cb;  
44:        }  
45:          
46:        @Override  
47:        public void run() {  
48:           doSomeWork();  
49:           try {  
50:              if (cibai.await() == 0)  
51:                 LearnBrokenBarrierException.barrierComplete(cibai);  
52:           } catch (InterruptedException e) {  
53:              e.printStackTrace();  
54:           } catch (BrokenBarrierException e) {  
55:              e.printStackTrace();  
56:           }  
57:        }  
58:    
59:        private void doSomeWork() {  
60:           System.out.println("Doing some work");  
61:        }  
62:          
63:     }  
64:    
65:  }  
66:    

CancellationException, ExecutionException, RejectedExecutionException and TimeoutException example, see below. Full source code can be download here.

1:  package play.learn.java.concurrent;  
2:    
3:  import java.util.concurrent.Callable;  
4:  import java.util.concurrent.CancellationException;  
5:  import java.util.concurrent.ExecutionException;  
6:  import java.util.concurrent.ExecutorService;  
7:  import java.util.concurrent.Executors;  
8:  import java.util.concurrent.TimeUnit;  
9:  import java.util.concurrent.TimeoutException;  
10:  import java.util.concurrent.FutureTask;  
11:    
12:  public class LearnCancellationException {  
13:    
14:     public static void main(String[] args) {  
15:        MyCallable callable1 = new MyCallable(1000);  
16:        MyCallable callable2 = new MyCallable(2000);  
17:    
18:        FutureTask<String> futureTask1 = new FutureTask<String>(callable1);  
19:        FutureTask<String> futureTask2 = new FutureTask<String>(callable2);  
20:    
21:        ExecutorService executor = Executors.newFixedThreadPool(2);  
22:        executor.execute(futureTask1);  
23:        executor.execute(futureTask2);  
24:    
25:        while (true) {  
26:           try {  
27:              if(futureTask1.isDone() && futureTask2.isDone()){  
28:                 System.out.println("Done");  
29:                 //shut down executor service  
30:                 executor.shutdown();  
31:                 return;  
32:              }  
33:                
34:              // uncomment for cancel  
35:              //futureTask2.cancel(true);  
36:    
37:              if(!futureTask1.isDone()){  
38:              //wait indefinitely for future task to complete  
39:              System.out.println("FutureTask1 output="+futureTask1.get());  
40:              }  
41:    
42:              System.out.println("Waiting for FutureTask2 to complete");  
43:              // set a samll range to get timedout exception.  
44:              String s = futureTask2.get(2000L, TimeUnit.MILLISECONDS);  
45:              if(s !=null){  
46:                 System.out.println("FutureTask2 output="+s);  
47:              }  
48:           } catch (CancellationException e) {  
49:              e.printStackTrace();  
50:           } catch (InterruptedException | ExecutionException e) {  
51:              e.printStackTrace();  
52:           } catch(TimeoutException e){  
53:              e.printStackTrace();  
54:           }  
55:        }  
56:    
57:     }  
58:       
59:     static class MyCallable implements Callable<String> {  
60:          
61:        private long waitTime;  
62:          
63:        public MyCallable(int timeInMillis) {  
64:           this.waitTime = timeInMillis;  
65:        }  
66:    
67:        @Override  
68:        public String call() throws Exception {  
69:           Thread.sleep(waitTime);  
70:           return Thread.currentThread().getName();  
71:        }  
72:          
73:     }  
74:    
75:  }  
76:    

CompletionException example, for full source code, you can download it here.

1:  public class LearnCompletableFuture {  
2:    
3:     public void learnCompletionException() {  
4:        try {  
5:           List<String> list = Arrays.asList("A", "B", "C", "D");  
6:           list.stream().map(s->CompletableFuture.supplyAsync(() -> s+s))  
7:           .map(f->f.getNow("Not Done")).forEach(s->System.out.println(s));  
8:    
9:        } catch (CompletionException e) {  
10:           e.printStackTrace();  
11:        }  
12:    
13:     }  
14:    
15:     public static void main(String[] args) {  
16:        LearnCompletableFuture c = new LearnCompletableFuture();  
17:        c.learnCompletionException();  
18:     }  
19:  }  

That's it for this article, for the incoming interface and classed until java.util.concurrent which will be publish in the next few articles, until then.

Learning Java Eden Space

If you have been a java developer and you should came across java garbage collection that free the object created by your application from occupied all the java heap. In today article, we will look into java heap and particular into java eden space. First, let's look at the general java heap.

From this StackOverflow

Heap memory

The heap memory is the runtime data area from which the Java VM allocates memory for all class instances and arrays. The heap may be of a fixed or variable size. The garbage collector is an automatic memory management system that reclaims heap memory for objects.

Eden Space: The pool from which memory is initially allocated for most objects.

Survivor Space: The pool containing objects that have survived the garbage collection of the Eden space.

Tenured Generation: The pool containing objects that have existed for some time in the survivor space.

When you created a new object, jvm allocate a part of the heap for your object. Visually, it is something as of following.

                   +-----+  
                   |     |  
   <-minor gc->    v     v   <------------- major gc---------------------->  
   +------------+-----+-----+----------------------------------------------+-------------+  
   |            |     |     |                                              |             |
   | Eden       | S0  | S1  |  Tenure Generation                           | Perm gen    |
   |            |     |     |                                              |             |
   +------------+-----+-----+----------------------------------------------+-------------+  
    <---------------------jvm heap (-Xms -Xmx)----------------------------> -XX:PermSize  
    <-- young gen(-Xmn)---->                                                -XX:MaxPermSize  

When eden space is fill with object and minor gc is performed, survive objects will copy to either survivor spaces; s0 or s1. At a time, one of the survivor space is empty. Because the eden space are relatively small in comparison to the tenure generation, hence, the gc that happened in eden space is quick.  Eden and both survivors spaces are also known as young or new generation.

To understand into how young generation heap get free, this article provided detail explanation.

The Sun/Oracle HotSpot JVM further divides the young generation into three sub-areas: one large area named "Eden" and two smaller "survivor spaces" named "From" and "To". As a rule, new objects are allocated in "Eden" (with the exception that if a new object is too large to fit into "Eden" space, it will be directly allocated in the old generation). During a GC, the live objects in "Eden" first move into the survivor spaces and stay there until they have reached a certain age (in terms of numbers of GCs passed since their creation), and only then they are transferred to the old generation. Thus, the role of the survivor spaces is to keep young objects in the young generation for a little longer than just their first GC, in order to be able to still collect them quickly should they die soon afterwards.

Based on the assumption that most of the young objects may be deleted during a GC, a copying strategy ("copy collection") is being used for young generation GC. At the beginning of a GC, the survivor space "To" is empty and objects can only exist in "Eden" or "From". Then, during the GC, all objects in "Eden" that are still being referenced are moved into "To". Regarding "From", the still referenced objects in this space are handled depending on their age. If they have not reached a certain age ("tenuring threshold"), they are also moved into "To". Otherwise they are moved into the old generation. At the end of this copying procedure, "Eden" and "From" can be considered empty (because they only contain dead objects), and all live objects in the young generation are located in "To". Should "to" fill up at some point during the GC, all remaining objects are moved into the old generation instead (and will never return). As a final step, "From" and "To" swap their roles (or, more precisely, their names) so that "To" is empty again for the next GC and "From" contains all remaining young objects.

As you can observed based on the visual diagram above, you can set the amount of heap for the eden and survivor space using -Xmn in the java parameter. There is also -XX:SurvivorRatio=ratio and you can find further information here for java8. Note that in the diagram above, Perm gen has been removed in java8, hence always refer find out what java run your application and refer to the right version of java documentation.

If you want to monitor the statistics of eden , you can use jstats. Previously I have written an article about jstat and you can read here what is jstat and how to use it. You can also enable gc log statistics and so jvm will write the gc statistics into a file, you can further read more here.

Till then we meet again in the next article. Please consider donate, thank you!