wait(), notify() e notifyAll() per la sincronizzazione tra i thread

wait and notify() Methods in Java | BaeldungBaeldung

Questi metodi sono definiti nella classe java.lang.Object (invece che nella classe java.lang.Thread). Questi metodi possono essere invocati solo all'interno di codice synchronized.

I metodi wait() e notify() permettono ad un oggetto condiviso di mettere in pausa un thread con il metodo wait() e di continuare, risvegliando il thread quando si ritiene appropriato con il metodo notify() o notifyAll().

Esempio: produttore e consumatore

In questo esempio, un produttore genera un messaggio (attraverso il metodo putMessage(message)) che deve essere consumato del consumatore (attraverso il metodo getMessage()), prima che il produttore possa produrre il prossimo messaggio. Questo è chiamato il pattern del produttore-consumatore: un thread può sospendere se stesso utilizzando wait() (e rilascia così il lock) se un messaggio è già stato prodotto e non ancora consumato dal consumatore. Il produttore rimane sospeso finché un altro thread, il consumatore, non prende il messaggio e non lo risveglia utilizzando notify() o notifyAll(). A questo punto, il produttore, dopo che il primo messaggio è stato consumato, può inserire l'ulteriore messaggio.

package trythreads.prod_cons;

//Testing wait() and notify()
public class MessageBox {
	private String message;
	private boolean hasMessage;

    	// producer
	public synchronized void putMessage(String message) {
		while (hasMessage) {
			// no room for new message
			try {
				wait(); // release the lock of this object
			} catch (InterruptedException e) {
			}
		}
		// acquire the lock and continue
		hasMessage = true;
		this.message = message + " Put @ " + System.nanoTime();
		notifyAll();
	}

	// consumer
	public synchronized String getMessage() {
		while (!hasMessage) {
			// no new message
			try {
				wait(); // release the lock of this object
			} catch (InterruptedException e) {
			}
		}
		// acquire the lock and continue
		hasMessage = false;
		notifyAll();
		return message + " Get @ " + System.nanoTime();
	}
}

package trythreads.prod_cons;

public class TestMessageBox {
	public static void main(String[] args) {
		final MessageBox box = new MessageBox();

		Thread producerThread = new Thread() {
			@Override
			public void run() {
				System.out.println("Producer thread started...");
				for (int i = 1; i <= 6; ++i) {
					box.putMessage("message " + i);
					System.out.println("Put message " + i);
				}
			}
		};

		Thread consumerThread1 = new Thread() {
			@Override
			public void run() {
				System.out.println("Consumer thread 1 started...");
				for (int i = 1; i <= 3; ++i) {
					System.out.println("Consumer thread 1 Get " + box.getMessage());
				}
			}
		};

		Thread consumerThread2 = new Thread() {
			@Override
			public void run() {
				System.out.println("Consumer thread 2 started...");
				for (int i = 1; i <= 3; ++i) {
					System.out.println("Consumer thread 2 Get " + box.getMessage());
				}
			}
		};

		consumerThread1.start();
		consumerThread2.start();
		producerThread.start();
	}
}

Consumer thread 1 started...
Producer thread started...
Consumer thread 2 started...
Put message 1
Consumer thread 2 Get message 1 Put @ 12649136150600 Get @ 12649136204100
Consumer thread 1 Get message 2 Put @ 12649136264200 Get @ 12649136286100
Put message 2
Put message 3
Consumer thread 2 Get message 3 Put @ 12649136554100 Get @ 12649136585800
Consumer thread 1 Get message 4 Put @ 12649136653600 Get @ 12649136673000
Put message 4
Put message 5
Consumer thread 2 Get message 5 Put @ 12649136944000 Get @ 12649136974700
Consumer thread 1 Get message 6 Put @ 12649137051700 Get @ 12649137088800
Put message 6

I messaggi di output (con System.out) potrebbero apparire fuori ordine. Ma a una più attenta osservazione sul timestamp di put/get conferma la corretta sequenza delle operazioni.

Con il metodo synchronized putMessage() il produttore acquisisce il lock sull'oggetto, controlla se il messaggio precedente è stato consumato. Altrimenti, chiama il metodo wait(), rilascia il lock su questo oggetto, va nello stato WAITING e piazza questo thread nel "wait" set dell'oggetto. Dall'altra parte, con il metodo synchronized getMessage() acquisisce il lock su questo oggetto e controlla per un nuovo messaggio. Se c'è il nuovo messaggio, pulisce il messaggio e invia una notify(), che arbitrariamente prende un thread dal wait set dell'oggetto (che è il thread produttore in questo caso) e lo piazza nello stato BLOCKED. Il thread consumatore, se non c'è un nuovo messaggio, a sua volta va nello stato WAITING e piazza se stesso nel "wait" set di questo oggetto (dopo l'invocazione del metodo wait()). Il thread produttore poi acquisisce il lock e prosegue le sue operazioni.

La differenza tra notify() e notifyAll() è che notify() arbitrariamente preleva un thread dal pool di quelli waiting associati all'oggetto e lo piazza nello stato di quelli che cercano di acquisire il lock; mentre notifyAll() risveglia tutti i thread nel waiting pool dell'oggetto. I thread risvegliati poi completano la loro esecuzione in maniera normale.

E' interessante notare che il multithreading è all'interno del linguaggio Java giusto nella classe root java.lang.Object. Il lock di sincronizzazione è in Object. I metodi wait(), notify(), notifyAll() utilizzati per coordinare i thread sono proprio nella classe Object.

wait() con timeout

Ci sono variazioni di wait() che prendono un valore di timeout:

public final void wait() throws InterruptedException
public final void wait(long timeout) throws InterruptedException
public final void wait(long timeout, int nanos) throws InterruptedExceptionJava

Il thread andrà ANCHE nello stato BLOCKED dopo che è passato il timeout.

Produttore/Consumatore e Code Bloccanti

Una coda bloccante è uno dei classici esempi nella programmazione parallela: il problema del produttore/consumatore. Questa problematica ha a che fare con il caso in cui abbiamo uno o più "produttori" che producono qualcosa e uno o più "consumatori" che consumano queste cose. Tutti i produttori e consumatori devono poter lavorare contemporaneamente (quindi in parallelo). Se non ci sono cose pronte da essere processate, il consumatore deve aspettare finché qualcosa non viene prodotto. In molte applicazioni, i produttori anche devono aspettare delle volte: Se le cose possono essere consumate a un certo ritmo, ad esempio uno al minuto, non ha senso per i produttori produrre continuamente a due cose al minuto. Questo porterebbe a un insieme illimitato di cosa che dovranno essere processati. Quando questo limite è raggiunto, i produttori devono aspettare prima di produrre.

Noi abbiamo bisogno di un modo per mandare le cose dal produttore e consumatore. La coda è una risposta ovvia: I Produttori mettono degli elementi nella coda come sono prodotti. I Consumatori rimuovono gli elementi dall'altro capo della coda.

Noi stiamo parlando di esecuzioni parallela, quindi noi abbiamo bisogno di una coda synchronized, ma abbiamo bisogno di più di questo. Quando la coda è vuota, abbiamo bisogno per far si che il consumatore aspetti finché un termine viene messo nella coda. Se la coda invece diventa piena, abbiamo bisogno di un modo per far aspettare il produttore finché non si libera qualche posto nella coda. Nella nostra applicazione, i produttori e consumatori sono thread. Un thread che è sospeso, aspettando per qualcosa avvenga, è detto bloccato, e il tipo di coda di cui abbiamo bisogno è detta coda bloccante. In una coda bloccante, in caso di operazione di dequeueing di un elemento da una coda vuota, il thread sarà sospeso finché un nuovo termine diventi disponibile;

Qui sotto c'è una implementazione di una coda bloccante.

package produttoreconsumatore.v5;

import java.util.LinkedList;
import java.util.List;

public class BlockingQueue {
    private int maxCapability;
    private List<Integer> list = new LinkedList<>();

    public BlockingQueue(int maxCapability) {
        this.maxCapability = maxCapability;
    }

    public synchronized void addContenuto(int contenuto)
            throws InterruptedException {

        while(list.size() == maxCapability) {
            wait();
        }

        if(list.size() == 0)
            notifyAll();

        System.out.println(Thread.currentThread().getName() +  " - PUT >>>>: " + contenuto);
        list.add(contenuto);
    }

    public synchronized int getContenuto() throws InterruptedException {

        while(list.size() == 0)
            wait();

        if(list.size() == maxCapability)
            notifyAll();

        int contenuto = list.remove(0);
        System.out.println(Thread.currentThread().getName() + " - GET <<<<<: " + contenuto);
        return contenuto;
    }
}

Java ha due classi che implementano la coda bloccante: LinkedBlockingQueue e ArrayBlockingQueue. Queste sono tipi parametrizzati per permetterci di specificare il tipo di elemento che la coda può contenere. Entrambe le classi sono definite nel package java.util.concurrent ed entrambe implementano un'interfaccia chiamata BlockingQueue. Se bqueue è una variabile appartenente a una di queste classi, allora le seguenti operazioni sono definite:

• bqueue.take() - Rimuove un elemento dalla coda e lo ritorna. Se la coda è vuota quando questo metodo è invocato, il thread che ha l'invocato sarà bloccato finché un elemento non sarà inserito nella coda.

• bqueue.put(item) - Aggiunge un elemento alla coda. Se la coda ha una capacità limitata ed è piena, il thread che ha invocato il metodo sarà bloccato qualche posto non si libererà nella coda. Il metodo lancerà l'eccezione InterruptedException se il thread sarà interrotto mentre è bloccato.

• bqueue.add(item) - Aggiunge un item alla coda, se c'è spazio. Se la coda ha una capacità limitata ed è piena, una IllegalStateException è lanciata. Questo metodo non è bloccante.

• bqueue.clear() - Rimuove tutti gli elementi dalla coda e li elimina.

Le code bloccanti di Java definiscono molti metodi addizionali (per esempio, bqueue.poll(500) è simile a bqueue.take(), eccetto che non si blocca per più di 500 millisecondi), ma i quattro indicati sono sufficienti per i nostri propositi. Notiamo che abbiamo visto due metodi per aggiungere elementi alla coda: bqueue.put(item) che blocca il thread se non c'è spazio disponibile nella coda ed è più appropriato da utilizzare con code bloccanti che hanno una capacità limitata; bqueue.add(item) non blocca ed è più appropriata che sia utilizzata con code bloccanti con capacità illimitata.

Un ArrayBlockingQueue ha una capacità massima che è specificata quando è istanziata. Per esempio, per creare una coda bloccante che può tenere fino a 25 oggetti di tipo ItemType, si potrebbe scrivere:

ArrayBlockingQueue<ItemType> bqueue = 
    new ArrayBlockingQueue<>(25);

Con questa dichiarazione, bqueue.put(item) bloccherà il thread chiamante se bqueue di già contiene 25 elementi, mentre bqueue.add(item) lancerà un'eccezione in questo caso. Ricordiamo che questo garantisce che gli elementi non sono prodotti in modo indefinito più velocemente di quanto possano essere consumati. Una LinkedBlockingQueue è pensata per creare una coda bloccante con capacità illimitata. Per esempio:

LinkedBlockingQueue<ItemType> bqueue = 
        new LinkedBlockingQueue<>();

crea una coda senza limite massimo di elementi che può contenere. In questo caso, bqueue.put(item) non si bloccherà mai e bqueue.add(item) non lancerà mai IllegalStateException. Si userà una LinkedBlockingQueue quando si vuole evitare che i produttori si blocchino, e si hanno altri metodi per evitare che la coda cresca in modo indefinito. Per entrambi i tipi di coda, bqueue.take() si bloccherà se la coda è vuota.

ESEMPI RIASSUNTIVI

GitHub - checksound/EsempiProduttoreConsumatoreGitHub