Interfața ExecutorService

ExecutorService este o interfață în Java ce permite executarea de task-uri asincrone, în background, în mod concurent, pe baza modelului Replicated Workers. De exemplu, putem avea nevoie să trimitem un număr de cereri, dar ar fi ineficient să le trimitem pe rând, secvențial, așteptând după fiecare să se termine. Soluția ar fi să lucrăm asincron, adică să trimitem o cerere, să nu așteptăm "după ea" (deci să se trimită în background) și să folosim thread-uri pentru a împărți numărul de cereri și a trimite mai multe deodată (concurent). Acesta ar fi cazul când nu știm în avans dimensiunea problemei pe care o rezolvăm, și fie avem nevoie de toate soluțiile problemei (pentru că trebuie să trimitem toate cererile), fie nu vrem să găsim toate soluțiile, ci minim una.

În continuare, se prezintă un exemplu de program care folosește ExecutorService pentru a afișa toate fișierele dintr-un director (inclusiv din fiecare subdirector în parte). Pentru că nu știm în avans câte fișiere și directoare va trebui să analizăm, putem spune că nu cunoaștem dimensiunea problemei.

import java.io.File;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
 
public class Example1 {
    public static void main(String[] args) {
        AtomicInteger inQueue = new AtomicInteger(0);
        ExecutorService tpe = Executors.newFixedThreadPool(4);
 
        inQueue.incrementAndGet();
        tpe.submit(new MyRunnable("files", tpe, inQueue));
    }
}
 
class MyRunnable implements Runnable {
    String path;
    ExecutorService tpe;
    AtomicInteger inQueue;
 
    public MyRunnable(String path, ExecutorService tpe, AtomicInteger inQueue) {
        this.path = path;
        this.tpe = tpe;
        this.inQueue = inQueue;
    }
 
    @Override
    public void run() {
        File file = new File(path);
        if (file.isFile()) {
            System.out.println(file.getPath());
        } else if (file.isDirectory()) {
            File[] files = file.listFiles();
 
            if (files != null) {
                for (File f : files) {
                    inQueue.incrementAndGet();
                    tpe.submit(new MyRunnable(f.getPath(), tpe, inQueue));
                }
            }
        }
 
        int left = inQueue.decrementAndGet();
        if (left == 0) {
            tpe.shutdown();
        }
    }
}

Așa cum se poate observa în codul de mai sus, instanțierea unui ExecutorService cu 4 workeri se realizează în thread-ul principal. Prin metoda submit(), se introduce un task nou în pool, corespunzător directorului părinte. Metoda submit() poate primi un obiect ce implementează interfața Runnable și conține metoda run() sau unul care implementează interfața Callable și conține metoda call(). Diferența dintre ele este că run() returnează void, iar call() returnează un obiect și poate arunca și o excepție. Cu alte cuvinte, Callable este potrivit când ne interesează rezultatul final și eventualele erori la executarea task-ului. În cadrul laboratorului, vom lucra cu interfața Runnable.

În exemplul de mai sus, clasa noastră ce reprezintă un task se numește MyRunnable. În metoda run, se verifică dacă avem de-a face cu un fișier sau cu un director. Dacă lucrăm pe un fișier, doar se printează numele acestuia. Dacă lucrăm cu un director, înseamnă că trebuie create task-uri noi, pentru fiecare fișier sau subdirector în parte conținut de directorul părinte. Astfel, se creează câte o instanță de MyRunnable pentru fiecare din ele și se adaugă în pool-ul de ExecutorService.

În cazul exemplului prezentat în acest laborator, ne trebuie toate soluțille problemei (toate fișierele și subdirectoarele), dar nu știm dinainte dimensiunea problemei, așa că o metodă prin care ne putem opri este să ținem evidența task-urilor din pool și să finalizăm execuția atunci când acesta se golește. Altfel, programul nostru va rula la infinit. Modul prin care se finalizează execuția unui ExecutorService poate fi observat în codul de mai sus. Mai precis, se folosește metoda shutdown() pentru a opri ExecutorService-ul din a primi task-uri noi. Dacă ne-ar fi interesat o singură soluție a problemei noastre (și nu toate, ca în exemplul de mai sus), am fi putut opri execuția în momentul în care ajungeam la acea soluție.