Introduzione

Il signor Doug Lea è un emerito professore che può vantarsi, oltre che per i molti successi professionali, di aver contribuito alla creazione di un'intera libreria di utility che sono finite nella distribuzione ufficiale di Java, dalla versione 1.5 (o 5.0 che dir si voglia): stiamo parlando delle utility del package java.util.concurrent

Frustrato (come molti!) dalle limitazioni e dalla poca flessibilità dei costrutti Java per il supporto alla programmazione concorrente, invece che lamentarsi si è rimboccato le mani e ha creato un'intera libreria di classi per gestire la concorrenza, includendo tutta una serie di utilità come:

• diverse implementazioni di lock;
• barriere;
• variabili atomiche;
• collezioni studiate per il multi-threading;
• ...

Gli ingegneri della Sun hanno evidentemente capito la lezione e hanno deciso di includerle nei package standard, con le dovute modifiche: utilizzo dei Generics (che sennò li avevano fatti per niente), alcune modifiche a qualche nome e cosette così.

Vi risparmierò tutta la storia, che potete leggere benissmo in molti siti, per mostrarvi alcuni utilizzi pratici di queste classi.

Un semplice thread pool

Realizzare a mano un thread pool è, nel migliore dei casi, un vero e proprio incubo! Non basta saper creare un Thread e lanciarlo, occorrono tutta una serie di controlli, sincronizzazioni, gestione degli errori che prima o poi, se non siete un professore della statura di mr. Lea, si rivelerà bacato e vi impallerà l'applicazione. Garantito.

Per rimanere sul pratico vorrei proporre un esempio molto concreto: un servizio di aggregazione feed RSS. Le caratteristiche del servizio saranno queste:

• nome della classe: FeedAggregator;
• un unico metodo (sincrono) aggregateFeeds() , che restituisce una lista di stringhe, il risultato della query RSS;
• il servizio è capace di processare contemporaneamente n feed, specificati in configurazione;
• chiamate contemporanee al servizio non "intasano" la JVM di thread, ma in ogni momento possono essere in esecuzione solamente gli n thread specificati in configurazione.

Tutto ciò si può realizzare utilizzando direttamente la classe java.util.concurrent.ThreadPoolExecutor, ma esiste un metodo ancora più semplice, c'è infatti una classe di utilità per sveltire la fase di creazione delle varie implementazioni di classi concorrenti: java.util.concurrent.Executors

Il metodo Executors.newFixedThreadPool(int nThreads) è proprio quello che fa al caso nostro: crea un'istanza di ExecutorService basata proprio su ThreadPoolExecutor, configurata per eseguire al massimo n threads.

Poiché ogni task dovrà restituire un risultato (una stringa contenente i dati del feed) dovremo implementare l'interfaccia java.util.concurrent.Callable invece che java.lang.Runnable. Daremo in pasto i nostri Callable al metodo ExecutorService.submit(), che ci restituirà un oggetto java.util.concurrent. Future: questo oggetto sarà il punto di sincronizzazione finale, perché la chiamata al suo metodo get() aspetterà la fine dell'esecuzione. Questa funzionlità è simile al costrutto Thread.join(), ma slegata dall'implementazione a Thread e molto più flessibile.

Ecco quindi l'implementazione del nostro feed aggregator:

package it.megadix.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class FeedAggregator {

  private ExecutorService executorService;

  public FeedAggregator(int threadPoolSize) {
    executorService = Executors.newFixedThreadPool(threadPoolSize);
  }

  public List<String> aggregateFeeds() throws Exception {
    List<String> result = new ArrayList<String>();
    List<Future<String>> futures = new ArrayList<Future<String>>();
    for (int i = 0; i < 20; i++) {
      futures.add(executorService.submit(new LongRunningTask(i)));
    }
    for (Future<String> future : futures) {
      result.add(future.get());
    }
    return result;
  }

  private void destroy() {
    executorService.shutdown();
  }

  public static void main(String[] args) {
    final FeedAggregator ag = new FeedAggregator(5);

    Thread th1, th2, th3;
    try {
      // chiama aggregateFeeds() da diversi thread contemporaneamente
      Runnable r = new Runnable() {
        public void run() {
          try {
            System.out.println(ag.aggregateFeeds());
          } catch (Exception e) {
            e.printStackTrace();
          }
        }
      };
      th1 = new Thread(r);
      th2 = new Thread(r);
      th3 = new Thread(r);

      th1.start();
      th2.start();
      th3.start();

      th1.join();
      th2.join();
      th3.join();

    } catch (Exception e) {
      e.printStackTrace();
    }
    ag.destroy();
  }
}

class LongRunningTask implements Callable<String> {
  private int id;

  public LongRunningTask(int id) {
    this.id = id;
  }

  public String call() throws Exception {
    System.out.println("[" + id + "] Start...");

    StringBuffer sb = new StringBuffer();
    sb.append("Feed [");
    sb.append(id);
    sb.append("] Bla bla bla");

    try {
      Thread.sleep((long) (Math.random() * 5000));
    } catch (InterruptedException e) {
    }
    System.out.println("RESULT:\n[" + id + "] ... finish!");

    return sb.toString();
  }
}

Alcune osservazioni:

• il costruttore FeedAggregator() richiede che venga specificato il numero di thread che utilizzerà;
• il tempo di esecuzione viene simulato con Thread.sleep();
• chiamate concorrenti allo stesso metodo non creano più degli n thread specificati.
• il codice dentro il main() utilizza i Thread alla vecchia maniera solamente per scopi didattici!