ForkJoinPool Sınıfı - Fork/Join Framework'ün En Temel Sınıfı

Giriş
Şu satırı dahil ederiz. Java 7 ile geldi.

import java.util.concurrent.ForkJoinPool;

Açıklaması şöyle.

Unfortunately, ForkJoinPool does not work well in the face of Thread.sleep(), because it designed for many short tasks that finish quickly, rather than tasks that block for a long time.

Şeklen şöyle

Önemli arayüzler şöyle

RecursiveAction : Sonuç dönmez

RecursiveTask : Sonuç döner

ExecutorService'ten Farkı Nedir?
Açıklaması şöyle. İşleri daha küçük parçalara (smaller recursive tasks) bölmek için tasarlanmıştır

A ForkJoinPool works on the work-stealing principle and was added in Java 7. The ForkJoinPool is similar to the ExecutorService, but with one difference being that the ForkJoinPool splits work units into smaller tasks (fork process) and then is submitted to the thread pool.

This is called the forking step and is a recursive process. The forking process continues until a limit is reached when it is impossible to split into further sub-tasks. All the sub-tasks are executed, and the main task waits until all the sub-tasks have finished their execution. The main task joins all the individual results and returns a final single result. This is the joining process, where the results are collated and a single data is built as an end result.

Work Stealing Nedir?

ForkJoinPool Work Stealing Nedir? yazısına taşıdım

Callable veya Runnable İş Ekleme

Açıklaması şöyle. Yani executor'dan çok farkı yok. 

Also be aware that ForkJoinPool not only allows the submission of ForkJoinTasks, it also allows the submission of Callable or Runnable tasks, so you can use ForkJoinPool in the same way that you could use the existing Executors.
The only difference would be that your task won’t split itself, but you could benefit from work stealing performance improvements if multiple tasks are submitted and there are some threads with less work than others.

ForkJoinTask İş Ekleme - İş Bölünebilir Olmalı

Buradaki avantaj eğer iş bölünebilir bir şeyse bir sürü thread tarafından çalıştırılıyor

execute()

invoke()

submit() 

metodları kullanılabilir. 

Hepsi parametre olarak ForkJoinTask nesnesi alırlar. ForkJoinTask soyut bir sınıftır bundan kalıtan 

RecursiveAction ve RecursiveTask yazılarına bakabilirsiniz. Açıklaması şöyle. 

There are two ways to submit a task to a ForkJoinPool:

RecursiveAction - A task which does not return any value. It does some work (e.g. copying a file from disk to a remote location and then exit). It may still need to break up its work into smaller chunks, which can be executed by independent threads or CPUs. A RecursiveAction can be implemented by sub-classing it.

RecursiveTask - A task that returns a result to the ForkJoinPool. It may split its work up into smaller tasks and merge the result of the smaller tasks into one result. The splitting of work into sub-tasks and merging may take place at several levels.,ScheduledThreadPoolExecutor,ForkJoinPool

constructor

Şöyle kurarız.

ForkJoinPool pool = new ForkJoinPool();  

constructor - int
Şöyle yaparız.

ForkJoinPool pool = new ForkJoinPool(3);

constructor - int parallelism, ForkJoinWorkerThreadFactory factory,

                        UncaughtExceptionHandler handler,

                        boolean asyncMode

Örnek ver

commonPool - Hazır Bir ForkJoinPool
Java 8'den itibaren JVM içinde ortak bir thread pool kullanılmaya başlandı. Açıklaması şöyle.

ForkJoinPool#commonPool() is a static thread-pool, which is lazily initialized when is actually needed. Two major concepts use the commonPool inside JDK: CompletableFuture and  Parallel Streams. There is one small difference between those two features: with  CompletableFuture, you are able to specify your own thread-pool and don't use the threads from the commonPool, you cannot in case of  Parallel Streams.

Pool içinde normalde işlemci sayısı - 1 kadar thread bulunur. Bu sayıyı değiştirmek için şöyle yaparırz

-Djava.util.concurrent.ForkJoinPool.common.parallelism=1000

Örnek
Şöyle yaparız

ForkJoinPool pool = ForkJoinPool.commonPool();  

execute metodu

İmzası şöyle.

public void execute(Runnable task)  
public void execute(ForkJoinTask<?> task)  

invoke metodu

İmzası şöyle. Aynı callable gibi bir sonuç döndürür. 

public <T> T invoke(ForkJoinTask<T> task)  

managedBlock metodu

Açıklaması şöyle

With managed blocking, the thread tells the thread pool that it may be blocked before it calls the potentially blocking method, and also informs the pool when the blocking method is finished. The thread pool then knows that there is a risk of a starvation deadlock, and may spawn additional threads if all of its threads are currently in some blocking operation and there are still other tasks to execute. 

Örnek

Şöyle yaparız

class BlockingGetUnicorns implements ForkJoinPool.ManagedBlocker {
  @Override
  public boolean block() {
    ...
  }
  @Override
  public boolean isReleasable() { return false; }
}

CompletableFuture<List<String>> fetchUnicorns  = 
  CompletableFuture.supplyAsync(() -> {
    BlockingGetUnicorns getThem = new BlockingGetUnicorns();
    try {
      ForkJoinPool.managedBlock(getThem);
     } catch (InterruptedException ex) {
       throw new AssertionError();
     }
     return getThem.unicorns;
});

shutdown metodu - Durdurma

Şöyle yaparız.

pool.shutdown();  
pool.awaitTermination(1, TimeUnit.MINUTES);  

submit metodu - Runnable veya Callable veya RecursiveAction veya RecursiveTask

İmzası şöyle. Future döndürür. 

public ForkJoinTask<?> submit(Runnable task)  
  
public <T> ForkJoinTask<T> submit(Runnable task, T result)  
  
public <T> ForkJoinTask<T> submit(Callable<T> task)  
  
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task)  

Örnek - RecursiveTask

Şöyle yaparız.

pool.submit(new RecursiveTaskA()); 

Örnek - RecursiveTask
Şöyle yaparız.

ForkJoinPool pool = new ForkJoinPool(3);
ForkJoinTask<F.Promise<Void>> result = pool
  .submit(() -> {
    ...
    return F.Promise.<Void>pure(null);
});