Poco tempo fa ho avuto la necessita di sviluppare un sistema multithread in grado di mantenere un pool con un numero massimo di thread in esecuzione la dove ci fossero stati processi da eseguire, con il jdk 1.4 avrei fatto questa operazione manualmente lavorando direttamente sulla classe Thread, ma dalla 1.5 ci sono venuti in aiuto una serie di oggetti che semplificano la vita in modo non indifferente.

Non riporto l’esempio dell’applicazioe che ho sviluppato ma semplifico il tutto facendo un esempio di vita comune, che comunque racchiude la logica del mio sistema, cosi da cercare di essere il piu’ chiaro possibile.

Livello:
Medio, Facile

Per chi:
Conosce i fondamentali Java e l’utilizzo dei Thread

Scenario:
Simulare il flusso di clienti in un ufficio postale utilizzando il thread pooling

Mettiamo che il nostro programma sia un ufficio postale, il quale ha 4 sportelli aperti.
Fuori dall’ ufficio postale sono presenti venti persone che vengono fatte entrare in sala di attesa, prendono il numero dalla numeratrice e aspettano il loro turno, logicamente queste persone non possono essere servite tutte assieme e tantomeno possiamo, fatte passare le prime quattro persone agli sportelli aperti, aspettare che si siano svuotati tutti per farne passare altre quattro in blocco, teniamo anche presente che ogni persona impiega un tempo differente per la propria operazione, quindi non e’ detto che la prima che arriva sia la prima a lasciare il posto.

Il flusso corretto e’ che, entrate le prime quattro, appena uno sportello si libera venga occupato dalla persona con il numero successivo, insomma, dobbiamo fare in modo che i quattro sportelli siano sempre pieni in qualunque momento sino al completo esaurimento della coda, come succede realmente :)

Ok, ora vediamo di mettere in pratica questa operazione manualmente, senza l’ausilio di librerie gia’ fatte da terzi, ma con il solo utilizzo dei thread e delle interfacce BlockingQueue e ExecutorService(presenti dalla versione Tiger).

Iniziamo con il creare un oggetto Persona, questo oggetto sara’ quello che entra in sala di attesa, prende il numero, aspetta e, venuto il suo turno, si reca allo sportello effettuando una operazione che potra’ richiedere piu’ o meno tempo.
Il tempo impiegato per l’operazione di ogni singola persona lo definisco con uno sleep di un numero di secondi casuale tar 0 e 20.

Logicamente l’oggetto Persona deve implementare l’interfaccia Runnable, quindi potremmo avere qualche cosa di simile a questo:

package posta;
 
import java.util.Random;
 
public class Persona implements Runnable {
 
private int numero;
 
public Persona(int numero) {
  System.out.println(" :: nuova persona prende il n. " + numero);
  // questo e' il numero "preso" all'ingesso
  // in posta
  this.numero = numero;
}
 
public void run() {
  // la persona si reca allo sportello 
  // per effettuare l'operazione
  effettuaOperazione();
}
 
private void effettuaOperazione() {
  String msg = String.format("[%d] si reca allo sportello", numero);
  System.out.println(msg);
 
  // qui creo un intro random non maggiore di 20
  // che utilizzo per fare in modo che ogni persona 
  // impieghi un tempo differente (1 secondo x n)
  // ad effettuare l'operazione
  int tempoOperazione = new Random().nextInt(20);
 
  try {
    // mi fermo allo sportello per il tempo
    // necessario all'operazione
    Thread.sleep(1000 * tempoOperazione);
  } catch (InterruptedException e) { }
 
  msg = String.format(" >> [%d] termina in %d secondi e abbandona " +
                "lo sportello", numero, tempoOperazione);
  System.out.println(msg);
}
}

Fatta la Persona che deve andare a pagare :) creiamo l’ Ufficio Postale, che sara’ munito di una sala di attesa e di 4 sportelli

package posta;
 
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
 
public class UfficioPostale {
 
private final int NUMERO_SPORTELLI = 4;
 
private final BlockingQueue<Runnable> salaAttesa = 
      new ArrayBlockingQueue<Runnable>(100, true);
 
private final ExecutorService sportello = 
      Executors.newFixedThreadPool(NUMERO_SPORTELLI);
 
private void aperturaSportelli() {
  System.out.println("  *** apertura sportelli ***  ");
  new Thread(new Runnable() {
    public void run() {
      for (;;) {
        try {
          // prendo e rimuovo il primo elemento
          // della coda e lo mando allo sportello
          sportello.execute(salaAttesa.take());
        } catch (InterruptedException e) {
          // ..continuo
        }
      }
    }
  }).start();
}
 
private void aperturaSalaAttesa(int persone) {
  System.out.println("  *** apertura sala di attesa ***  ");
 
    for (int i = 0; i < persone; i++) {
      try {
        // inserisco la persona nella coda
        salaAttesa.put(new Persona(i));
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }
 
private void apertura(int persone) {
  System.out.println("  *** apertura ufficio postale ***  ");
 
  aperturaSportelli();
 
  aperturaSalaAttesa(persone);
}
 
public static void main(String[] args) {
  int persone = Integer.parseInt(args[0]);
  System.out.println(" --- sono presenti n " + persone + " persone");
 
  UfficioPostale up = new UfficioPostale();
  up.apertura(persone);
}
 
}

Bene, ora eseguiamo il codice e guardiamo l’output:

$ cd /home/crx/dev/java/Thread/bin
$ java posta/UfficioPostale 20
 
 --- sono presenti n 20 persone
  *** apertura ufficio postale ***
  *** apertura sportelli ***
  *** apertura sala di attesa ***
 :: nuova persona prende il n. 0
 :: nuova persona prende il n. 1
 :: nuova persona prende il n. 2
 :: nuova persona prende il n. 3
[0] si reca allo sportello
 :: nuova persona prende il n. 4
[1] si reca allo sportello
 :: nuova persona prende il n. 5
[2] si reca allo sportello
 :: nuova persona prende il n. 6
 :: nuova persona prende il n. 7
 :: nuova persona prende il n. 8
 :: nuova persona prende il n. 9
 :: nuova persona prende il n. 10
 :: nuova persona prende il n. 11
 :: nuova persona prende il n. 12
 :: nuova persona prende il n. 13
 :: nuova persona prende il n. 14
 :: nuova persona prende il n. 15
 :: nuova persona prende il n. 16
 :: nuova persona prende il n. 17
 :: nuova persona prende il n. 18
 :: nuova persona prende il n. 19
[3] si reca allo sportello
 >> [1] termina in 2 secondi e abbandona lo sportello
[4] si reca allo sportello
 >> [0] termina in 7 secondi e abbandona lo sportello
[5] si reca allo sportello
 >> [2] termina in 9 secondi e abbandona lo sportello
[6] si reca allo sportello
 >> [4] termina in 9 secondi e abbandona lo sportello
[7] si reca allo sportello
 >> [3] termina in 12 secondi e abbandona lo sportello
[8] si reca allo sportello
 >> [8] termina in 5 secondi e abbandona lo sportello
[9] si reca allo sportello
 >> [5] termina in 11 secondi e abbandona lo sportello
[10] si reca allo sportello
 >> [7] termina in 12 secondi e abbandona lo sportello
[11] si reca allo sportello
 >> [11] termina in 1 secondi e abbandona lo sportello
[12] si reca allo sportello
 >> [6] termina in 18 secondi e abbandona lo sportello
[13] si reca allo sportello
 >> [12] termina in 5 secondi e abbandona lo sportello
[14] si reca allo sportello
 >> [9] termina in 14 secondi e abbandona lo sportello
[15] si reca allo sportello
 >> [13] termina in 5 secondi e abbandona lo sportello
[16] si reca allo sportello
 >> [10] termina in 19 secondi e abbandona lo sportello
[17] si reca allo sportello
 >> [14] termina in 10 secondi e abbandona lo sportello
[18] si reca allo sportello
 >> [15] termina in 15 secondi e abbandona lo sportello
[19] si reca allo sportello
 >> [16] termina in 17 secondi e abbandona lo sportello
 >> [19] termina in 6 secondi e abbandona lo sportello
 >> [17] termina in 18 secondi e abbandona lo sportello
 >> [18] termina in 16 secondi e abbandona lo sportello

Come potete vedere, spero di aver reso bene l’idea con l’output, i 4 soportelli sono sempre al lavoro, le persone si recano agli sportelli correttamente secondo la sequenza del numero preso e appena una esce quella con il numero successivo prende il posto dello sportollo che si e’ liberato.

Vediamo in modo sommario i due elementi chiave di questo sistema che sono:

ArrayBlockingQueue
L’ oggetto che ho utilizzato e’ in grado di creare una coda di seguendo la logica First-In-First-Out(FIFO), quindi tutti i nuovi elementi che vengono inseriti finiscono in fondo alla coda e il recupero degli elementi dalla coda parte sempre dal primo.
Per funzionare l’oggetto necessita di una dichiarazione di dimensione che non puo essere aumentata in seguito
ArrayBlockingQueue(100)
se si cercano di inserire piu’ elementi di quelli previsti questi rimarranno in attesa sino a quendo non si libera la coda inatti se nel codice precedente diminuiamo il numero massimo di persono che possono entrare in sala di salaAttesa

private final BlockingQueue salaAttesa =
new ArrayBlockingQueue(5, true);

e commentiamo il metodo aperturaSportelli ecco cosa succede:

$ java posta/UfficioPostale 20
— sono presenti n 20 persone
*** apertura ufficio postale ***
*** apertura sala di attesa ***
:: nuova persona prende il n. 0
:: nuova persona prende il n. 1
:: nuova persona prende il n. 2
:: nuova persona prende il n. 3
:: nuova persona prende il n. 4
:: nuova persona prende il n. 5

si ferma visto che gli sportelli sono “chiusi” e la sala di attesa ha una capacita’ massima di 5 persone, le altre rimangono “fuori”.

Executors.newFixedThreadPool()
Questo oggetto crea un pool di thread che non supereranno il numero definito del metodo newFixedThreadPool()
quindi:
Executors.newFixedThreadPool(5) crea un pool di massimo 5 thread contemporanei in esecuzione

Nota
Nell’esempio che ho fatto il metodo aperturaSportelli() lancia un thread che, pur avendo esaurito la coda, continua a rimanere in vita e di conseguenza l’applicazione continua a funzionare, nel caso in cui successivamente all’esaurimento della coda venisse chiamato salaAttesa.put(new Persona(i)), anche a distanza di tempo, la persona arrivata verrebbe servita ugualemnte

Per finire
Questo tipo di operazioni possono essere rese decisamente piu’ sofisticate utilizzando, e studiando, gli oggetti del package java.util.concurrent al quale, soprattutto con l’arrivo del jdk 1.5, sono stati aggiunti molti elementi che aiutano la gestione e l’esecuzione di questi flussi.

…date uno sguardo alla classe ThreadPoolExecutor che e’ molto interessante, magari piu’ avanti scrivo un articoletto anche su questo.

Come sempre spero di aver fatto cosa gradita e utile per qualcuno, ah… ricordatevi che gli uffici postali qui a Milano chiudono alle 14.00 ;)

Ciao e alla prossima,
Cristian.