JAVA CONCURRENT EXECUTORS(CALISTIRICILAR)

Yazan:Fatih KABAKCI

Java.util.Concurrent paketinin önemli yapılarından bir tanesi olan Executors(Çalıştırıcılar), bir küme thread(kanal) sınıflarının çalıştırılması ve yönetilmesini basitleştiren eş zamanlı araçlardır. Executor' lar sabit ve dinamik sayıda veya bir plana göre(scheduled) çalışması tasarlanmış kanal havuzları(thread pool) oluşturur. Bu kanal havuzları her bir thread' in load edilme ve kendi run() metotlarını çağırma gecikmelerine nazaran, daha hızlı bir çalışma şekli sağlar. En önemlisi eş zamanlılığı en zarif bir şekilde uygulamaktadır. Bu zariflik kodun daha sade ve anlaşılır olmasının yanında, performans üzerinde de önemli etkisinin olabilmesi olarak gösterilebilir.

Önemli Arabirimler(Interfaces)

Sık Kullanılan Sınıflar(Classes)

• ThreadPoolExecutor kanal çalıştırma işlemlerini sağlar.
• ScheduledThreadPoolExecutor belirli bir zamana göre ayarlanmış kanal çalıştırma işlemlerini sağlar.

Java eş zamanlı API içerisinde, Executor sınıfının dışında, içerisinde static olarak tanımlanmış bir takım thread oluşum metotları bulunduran Executors sınıfı bulunur. Bu sınıf ile farklı şekillerde kolaylıkla kanal havuzları oluşturabilirsiniz.

Bunlar aşağıda açıklanmaktadır.

public static ExecutorService newFixedThreadPool(int nThreads)
Sabit sayıda kanal(thread) oluşturmaya yarar.

public static ExecutorService newSingleThreadExecutor()
Tek bir sayıda kanal oluşturmaya yarar.

public static ExecutorService newCachedThreadPool()
Daha önceden oluşturulan kanallar ile birlikte yeni bir kanal havuzu oluşturur.

public static ScheduledExecutorService newSingleThreadScheduledExecutor()
Belirlenmiş bir zamana göre tek bir sayıda kanal oluşturur.

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
Belirlenmiş bir zaman göre belirli sayıda bir kanal oluşturur.

Yukarıdaki metotların bir de ThreadFactory parametresine sahip overload edilmiş diğer versiyonları da bulunur. Bu da verilen ThreadFactory nesnesine göre çalışma şekillerini değiştirir.

Counter Thread Sınıfı

package Concurrent.Executors;

/**
 * @author www.fatihkabakci.com
 */
public class Counter implements Runnable {

   int from;
   int to;

   public Counter(int from, int to) {
      this.from = from;
      this.to = to;
   }

   @Override
   public void run() {
      for (int i = from; i <= to; i++) {
         System.out.println(Thread.currentThread().getName() + " -> " + i);
      }
   }
}

Normal şartlarda yukarıdaki Counter sınıfından 5 adet instance oluşturarak paralel sayma işlemi aşağıdaki gibi yapılabilir.

Varsayılan Yöntem(Default Java Strategy)

/**
 * 
 */

package Concurrent.Executors;

import java.util.ArrayList;

/**
 * @author www.fatihkabakci.com
 */
public class DefaultParallelCounting {
   public static void main(String[] args) {

      Runnable counters[] = {new Counter(1, 10), new Counter(11, 20), new Counter(21, 30), new Counter(31, 40), new Counter(41, 50)};
      ArrayList<Thread> threads = new ArrayList<Thread>();
      
      System.out.println("Start");
      for (Runnable counter : counters) {
         Thread thread = new Thread(counter);
         threads.add(thread);
         thread.start();
      }
      for (Thread thread : threads) {
         while (thread.isAlive());
      }
      System.out.println("Finish");
   }
}

Concurrent Strategy

/**
 * 
 */

package Concurrent.Executors;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author www.fatihkabakci.com
 */
public class ExecutorParallelCounting {
   public static void main(String[] args) {

      Runnable counters[] = {new Counter(1, 10), new Counter(11, 20), new Counter(21, 30), new Counter(31, 40), new Counter(41, 50)};
      ExecutorService executor = Executors.newFixedThreadPool(5);
      System.out.println("Start");
      for (Runnable counter : counters)
         executor.execute(counter);
      
      executor.shutdown();
      while(!executor.isTerminated());
      System.out.println("Finish");
   }
}

5 adet thread ile oluşturulan sabit uzunluklu kanal(thread) havuzu, her bir thread' i doğrudan çalıştırarak(execute) tüm kanalların yönetilmesini kendi üzerinden yürütmektedir. Bu sayede kanalların sonlandırılması ve yönetilmesi thread pool üzerinden yapılarak, daha verimli bir yöntem oluşturulabilir.

Callable ve Future Arabirimleri

1. Sınıflarınızın Callable arabirimini uygulayarak call() metodunu override etmesi.
2. ExecutorService arabirimini uygulayan sınıf nesnesinin submit() metoduna Callable referansını göndermesi.
3. submit() metodundan dönen Future nesnesinin elde edilmesi. 

CallableCounter Sınıfı

package Concurrent.Executors;

import java.util.concurrent.Callable;

/**
 * @author www.fatihkabakci.com
 */
public class CallableCounter implements Callable<Integer> {

   int from;
   int to;

   public CallableCounter(int from, int to) {
      this.from = from;
      this.to = to;
   }

   @Override
   public Integer call() throws Exception {
      int sum = 0;
      for (int i = from; i <= to; i++) {
         System.out.println(Thread.currentThread().getName() + " -> " + i);
         sum = sum + i;
      }
      return sum;
   }
}

CallableParallelCounting Main Sınıfı

/**
 * 
 */

package Concurrent.Executors;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @author www.fatihkabakci.com
 */
public class CallableParallelCounting {
   public static void main(String[] args) throws InterruptedException, ExecutionException {

      CallableCounter counters[] = {new CallableCounter(1, 10), 
                                    new CallableCounter(11, 20), 
                                    new CallableCounter(21, 30),
                                    new CallableCounter(31, 40), 
                                    new CallableCounter(41, 50)
                                    };

      ExecutorService executor = Executors.newFixedThreadPool(5);
      System.out.println("Start");
      for (CallableCounter counter : counters) {
         Future<Integer> future = executor.submit(counter);
         Integer result = future.get();
         System.out.println("[Result]:" + result);
      }
      
      executor.shutdown();
      while (!executor.isTerminated());
      System.out.println("Finish");
   }
}

for (CallableCounter counter : counters) {
         Future<Integer> future = executor.submit(counter);
         Integer result = future.get();
         System.out.println("[Result]:" + result);
      }

TimeUnit Enum Sınıfı

Java Concurrent sınıflarının aynı zamanda yapılan sorgulara -maksimum bekleme süreleri de tayin eder. Aşağıda TimeUnit Enumeration sınıfında tanımlanan zaman değerleri verilmiştir.