Heading

Guida al framework Fork/Join in Java

Instanziare ExecutorService

2.1. Metodi di fabbrica della classe Executors

Il modo più semplice per creare ExecutorService è quello di utilizzare uno dei metodi di fabbrica della classe Executors.

Per esempio, la seguente linea di codice creerà un pool di thread con 10 thread:

ExecutorService executor = Executors.newFixedThreadPool(10);

Ci sono diversi altri metodi di fabbrica per creare un ExecutorService predefinito che soddisfa casi d’uso specifici. Per trovare il metodo migliore per le vostre esigenze, consultate la documentazione ufficiale di Oracle.

2.2. Creare direttamente un ExecutorService

Perché ExecutorService è un’interfaccia, si può usare un’istanza di qualsiasi sua implementazione. Ci sono diverse implementazioni tra cui scegliere nel pacchetto java.util.concurrent, oppure si può creare la propria.

Per esempio, la classe ThreadPoolExecutor ha alcuni costruttori che possiamo usare per configurare un servizio executor e il suo pool interno:

ExecutorService executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

Si può notare che il codice sopra è molto simile al codice sorgente del metodo factory newSingleThreadExecutor(). Per la maggior parte dei casi, una configurazione manuale dettagliata non è necessaria.

Assegnare i compiti a ExecutorService

ExecutorService può eseguire compiti Runnable e Callable. Per mantenere le cose semplici in questo articolo, saranno usati due compiti primitivi. Notate che qui usiamo espressioni lambda invece di classi interne anonime:

Runnable runnableTask = () -> { try { TimeUnit.MILLISECONDS.sleep(300); } catch (InterruptedException e) { e.printStackTrace(); }};Callable<String> callableTask = () -> { TimeUnit.MILLISECONDS.sleep(300); return "Task's execution";};List<Callable<String>> callableTasks = new ArrayList<>();callableTasks.add(callableTask);callableTasks.add(callableTask);callableTasks.add(callableTask);

Possiamo assegnare compiti all’ExecutorService usando diversi metodi tra cui execute(), che è ereditato dall’interfaccia Executor, e anche submit(), invokeAny() e invokeAll().

Il metodo execute() è nullo e non dà alcuna possibilità di ottenere il risultato dell’esecuzione di un task o di controllare lo stato del task (è in esecuzione):

executorService.execute(runnableTask);

submit() sottopone un task Callable o Runnable a un ExecutorService e restituisce un risultato di tipo Future:

Future<String> future = executorService.submit(callableTask);

invokeAny() assegna un insieme di task a un ExecutorService, causando l’esecuzione di ciascuno di essi, e restituisce il risultato di un’esecuzione riuscita di un task (se c’è stata un’esecuzione riuscita):

String result = executorService.invokeAny(callableTasks);

invokeAll() assegna un insieme di compiti a un ExecutorService, facendo eseguire ciascuno di essi, e restituisce il risultato di tutte le esecuzioni dei compiti sotto forma di una lista di oggetti di tipo Future:

List<Future<String>> futures = executorService.invokeAll(callableTasks);

Prima di andare avanti, dobbiamo discutere altri due punti: chiudere un ExecutorService e trattare con i tipi di ritorno Future.

Spegnere un ExecutorService

In generale, l’ExecutorService non sarà automaticamente distrutto quando non ci sono compiti da processare. Rimarrà in vita e aspetterà un nuovo lavoro da fare.

In alcuni casi questo è molto utile, come quando un’applicazione ha bisogno di elaborare compiti che appaiono su base irregolare o la quantità di compiti non è nota al momento della compilazione.

D’altra parte, un’applicazione potrebbe raggiungere la sua fine ma non essere fermata perché un ExecutorService in attesa farà sì che la JVM continui a funzionare.

Per chiudere correttamente un ExecutorService, abbiamo le API shutdown() e shutdownNow().

Il metodo shutdown() non causa la distruzione immediata dell’ExecutorService. Farà sì che l’ExecutorService smetta di accettare nuovi compiti e si spenga dopo che tutti i thread in esecuzione hanno finito il loro lavoro corrente:

executorService.shutdown();

Il metodo shutdownNow() cerca di distruggere immediatamente l’ExecutorService, ma non garantisce che tutti i thread in esecuzione vengano fermati nello stesso momento:

List<Runnable> notExecutedTasks = executorService.shutDownNow();

Questo metodo restituisce una lista di compiti che sono in attesa di essere processati. Sta allo sviluppatore decidere cosa fare con questi compiti.

Un buon modo per chiudere l’ExecutorService (che è anche raccomandato da Oracle) è quello di utilizzare entrambi questi metodi combinati con il metodo awaitTermination():

executorService.shutdown();try { if (!executorService.awaitTermination(800, TimeUnit.MILLISECONDS)) { executorService.shutdownNow(); } } catch (InterruptedException e) { executorService.shutdownNow();}

Con questo approccio, l’ExecutorService smetterà prima di prendere nuovi compiti e poi aspetterà fino ad un determinato periodo di tempo che tutti i compiti siano completati. Se questo tempo scade, l’esecuzione viene fermata immediatamente.

L’interfaccia Future

I metodi submit() e invokeAll() restituiscono un oggetto o una collezione di oggetti di tipo Future, che ci permette di ottenere il risultato dell’esecuzione di un task o di controllare lo stato del task (è in esecuzione).

L’interfaccia Future fornisce uno speciale metodo bloccante get(), che restituisce un risultato effettivo dell’esecuzione del task Callable o nullo nel caso di un task Runnable:

Future<String> future = executorService.submit(callableTask);String result = null;try { result = future.get();} catch (InterruptedException | ExecutionException e) { e.printStackTrace();}

Chiamando il metodo get() mentre il task è ancora in esecuzione, l’esecuzione si bloccherà fino a quando il task non verrà eseguito correttamente e il risultato sarà disponibile.

Con un blocco molto lungo causato dal metodo get(), le prestazioni di un’applicazione possono degradare. Se i dati risultanti non sono cruciali, è possibile evitare tale problema usando i timeout:

String result = future.get(200, TimeUnit.MILLISECONDS);

Se il periodo di esecuzione è più lungo di quello specificato (in questo caso, 200 millisecondi), verrà lanciata una TimeoutException.

Possiamo usare il metodo isDone() per controllare se il task assegnato è già stato elaborato o no.

L’interfaccia Future prevede anche l’annullamento dell’esecuzione del compito con il metodo cancel() e la verifica della cancellazione con il metodo isCancelled():

boolean canceled = future.cancel(true);boolean isCancelled = future.isCancelled();

L’interfaccia ScheduledExecutorService

Lo ScheduledExecutorService esegue i compiti dopo un certo ritardo predefinito e/o periodicamente.

Ancora una volta, il modo migliore per istanziare uno ScheduledExecutorService è usare i metodi factory della classe Executors.

Per questa sezione, usiamo uno ScheduledExecutorService con un thread:

ScheduledExecutorService executorService = Executors .newSingleThreadScheduledExecutor();

Per programmare l’esecuzione di un singolo task dopo un ritardo fisso, usiamo il metodo scheduled() dello ScheduledExecutorService.

Due metodi scheduled() permettono di eseguire task Runnable o Callable:

Future<String> resultFuture = executorService.schedule(callableTask, 1, TimeUnit.SECONDS);

Il metodo scheduleAtFixedRate() ci permette di eseguire un task periodicamente dopo un ritardo fisso. Il codice qui sopra ritarda di un secondo prima di eseguire callableTask.

Il seguente blocco di codice eseguirà un task dopo un ritardo iniziale di 100 millisecondi. E dopo questo, eseguirà lo stesso compito ogni 450 millisecondi:

Future<String> resultFuture = service .scheduleAtFixedRate(runnableTask, 100, 450, TimeUnit.MILLISECONDS);

Se il processore ha bisogno di più tempo per eseguire un compito assegnato rispetto al parametro period del metodo scheduleAtFixedRate(), lo ScheduledExecutorService aspetterà che il compito corrente sia completato prima di iniziare quello successivo.

Se è necessario avere un ritardo di lunghezza fissa tra le iterazioni del task, si dovrebbe usare scheduleWithFixedDelay().

Per esempio, il seguente codice garantirà una pausa di 150 millisecondi tra la fine dell’esecuzione corrente e l’inizio di un’altra:

service.scheduleWithFixedDelay(task, 100, 150, TimeUnit.MILLISECONDS);

Secondo i contratti dei metodi scheduleAtFixedRate() e scheduleWithFixedDelay(), l’esecuzione periodica del task terminerà al termine dell’ExecutorService o se viene lanciata un’eccezione durante l’esecuzione del task.

ExecutorService vs Fork/Join

Dopo il rilascio di Java 7, molti sviluppatori hanno deciso di sostituire il framework ExecutorService con il framework fork/join.

Questa però non è sempre la decisione giusta. Nonostante la semplicità e i frequenti guadagni di prestazioni associati al fork/join, esso riduce il controllo dello sviluppatore sull’esecuzione concorrente.

ExecutorService dà allo sviluppatore la possibilità di controllare il numero di thread generati e la granularità dei compiti che dovrebbero essere eseguiti da thread separati. Il miglior caso d’uso di ExecutorService è l’elaborazione di compiti indipendenti, come transazioni o richieste secondo lo schema “un thread per un compito”.

Al contrario, secondo la documentazione di Oracle, fork/join è stato progettato per accelerare il lavoro che può essere suddiviso in pezzi più piccoli in modo ricorsivo.

Conclusione

Nonostante la relativa semplicità di ExecutorService, ci sono alcune insidie comuni.

Riassumiamole:

Mantenere in vita un ExecutorService inutilizzato: Vedere la spiegazione dettagliata nella sezione 4 su come spegnere un ExecutorService.

Capacità errata del pool di thread mentre si usa un pool di thread a lunghezza fissa: È molto importante determinare quanti thread l’applicazione avrà bisogno per eseguire i compiti in modo efficiente. Un pool di thread troppo grande causerà un inutile overhead solo per creare thread che saranno per lo più in modalità di attesa. Troppo pochi possono far sembrare un’applicazione poco reattiva a causa dei lunghi periodi di attesa per i compiti in coda.

Chiamare il metodo get() di un futuro dopo la cancellazione del compito: Il tentativo di ottenere il risultato di un task già cancellato scatena una CancellationException.

Blocco inaspettatamente lungo con il metodo get() di Future: Dovremmo usare i timeout per evitare attese inaspettate.

Come sempre, il codice di questo articolo è disponibile nel repository GitHub.