Pipeline di elaborazione dei file: producer–consumer

1. Come coordinare più thread durante l'elaborazione dei file

Quando si lavora con file di grandi dimensioni o con elaborazioni complesse dei dati, spesso si presenta l'esigenza: ripartire il lavoro tra più thread. Ad esempio, un thread legge le righe da un file, mentre gli altri — elaborano queste righe (cercano parole, calcolano statistiche, ecc.).

Qual è il problema?

• Se tutti i thread lavorano con la stessa risorsa (ad esempio con un file o una collezione condivisa), è facile incorrere in errori: alcuni thread possono “sorpassare” gli altri, si creano race condition, perdite di dati, saturazione della memoria.
• Se i thread sono molti, ma non c'è coordinamento — alcuni resteranno inattivi, altri saranno sovraccarichi.

Serve una soluzione che:

• Consenta uno scambio sicuro di dati tra thread.
• Eviti l'esaurimento della memoria (se l'elaborazione procede più lentamente della lettura).
• Permetta di terminare facilmente il lavoro di tutti i thread quando i dati sono finiti.

2. Pattern “Producer–Consumer” (Produttore–Consumatore)

Producer–Consumer è un pattern classico che aiuta i thread a lavorare in modo coordinato, senza intralciarsi.

In esso ci sono due ruoli: il Producer (produttore) crea i dati — ad esempio, legge righe da un file o riceve messaggi dalla rete — e li inserisce in una coda condivisa. Il Consumer (consumatore) preleva i dati da questa coda e li elabora: conta le parole, salva nel database o scrive in un altro file.

L'idea principale è che questi due tipi di thread lavorano in modo indipendente. Il produttore può leggere più velocemente di quanto il consumatore riesca a elaborare, o viceversa — e nessuno blocca l'altro. Tra loro c'è un buffer — una coda che uniforma il ritmo di lavoro.

Schema visivo

[File] --(legge)--> [Producer] --(inserisce in coda)--> [BlockingQueue] --(preleva)--> [Consumer] --(elabora)

3. Implementazione con BlockingQueue

In Java, per organizzare questo scambio è ideale l'interfaccia BlockingQueue (ad esempio l'implementazione ArrayBlockingQueue).

Che cos'è BlockingQueue?

BlockingQueue è una coda thread-safe a capacità limitata che si occupa autonomamente della sincronizzazione. Se il producer tenta di aggiungere un elemento e la coda è già piena, il thread si blocca automaticamente e attende finché non si libera spazio. Analogamente, se il consumer tenta di prelevare un elemento e la coda è vuota, attende semplicemente finché qualcuno non inserisce qualcosa nella coda.

Questo meccanismo risolve automaticamente il problema classico del “sorpasso” tra thread e dell'esaurimento della memoria: i producer non intasano la coda con elementi in eccesso e i consumer non cercano di lavorare “nel vuoto”. Tutti i thread eseguono il proprio lavoro in modo regolare e coordinato.

Esempio di creazione della coda

import java.util.concurrent.*;

BlockingQueue<String> queue = new ArrayBlockingQueue<>(100); // buffer di 100 elementi

100 è il numero massimo di elementi nella coda. Se il producer lavora più velocemente, attenderà finché il consumer non avrà elaborato parte dei dati.

4. Coordinamento dei thread: backpressure e terminazione

Limitazione della dimensione della coda (backpressure)

Backpressure è un meccanismo che impedisce al producer di saturare la memoria se il consumer non riesce a elaborare i dati abbastanza velocemente.

• Se la coda è piena, il producer “rallenta” automaticamente (il metodo put() si blocca).
• Se la coda è vuota, il consumer attende (il metodo take() si blocca).

Questo consente al sistema di lavorare in modo stabile anche con velocità dei thread diverse.

Terminazione: “poison pill” (pillola avvelenata)

Quando il producer ha finito di leggere il file, bisogna in qualche modo informare i consumer che non arriveranno più dati e che è ora di terminare.

Soluzione:

• Nella coda si inserisce un oggetto speciale — una “poison pill” (ad esempio la stringa "__END__" o un altro valore speciale, ma non null).
• Il consumer, ricevendo la “poison pill”, capisce che è il momento di terminare.

Se i consumer sono più di uno, occorre inserire tante “poison pill” quanti sono i consumer!

5. Esempio di pipeline: lettura del file ed elaborazione delle righe

Realizziamo una pipeline semplice:

• Un thread legge le righe da un file e le inserisce nella coda.
• Più thread prelevano le righe dalla coda, contano il numero di parole e stampano il risultato.

Passo 1. Producer — lettore del file

import java.io.*;
import java.util.concurrent.*;

public class FileProducer implements Runnable {
    private final BlockingQueue<String> queue;
    private final File file;
    private final int consumerCount;
    private final String POISON_PILL = "__END__";

    public FileProducer(BlockingQueue<String> queue, File file, int consumerCount) {
        this.queue = queue;
        this.file = file;
        this.consumerCount = consumerCount;
    }

    @Override
    public void run() {
        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                queue.put(line); // se la coda è piena — attendiamo
            }
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            // Inseriamo una poison pill per ogni consumer
            try {
                for (int i = 0; i < consumerCount; i++) {
                    queue.put(POISON_PILL);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

Passo 2. Consumer — elaboratore di righe

import java.util.concurrent.BlockingQueue;

public class LineConsumer implements Runnable {
    private final BlockingQueue<String> queue;
    private final String POISON_PILL = "__END__";

    public LineConsumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                String line = queue.take(); // se la coda è vuota — attendiamo
                if (POISON_PILL.equals(line)) {
                    break; // terminazione
                }
                int wordCount = line.trim().isEmpty() ? 0 : line.trim().split("\\s+").length;
                System.out.println(Thread.currentThread().getName() + ": " + wordCount + " parole nella riga: " + line);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

Passo 3. Avvio della pipeline tramite Executors

import java.io.File;
import java.util.concurrent.*;

public class PipelineDemo {
    public static void main(String[] args) throws Exception {
        int consumerCount = 3;
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);

        File file = new File("input.txt"); // il tuo file

        // Avviamo il producer
        Thread producer = new Thread(new FileProducer(queue, file, consumerCount));
        producer.start();

        // Avviamo i consumer tramite ExecutorService
        ExecutorService consumers = Executors.newFixedThreadPool(consumerCount);
        for (int i = 0; i < consumerCount; i++) {
            consumers.submit(new LineConsumer(queue));
        }

        // Attendiamo che il producer termini
        producer.join();

        // Attendiamo la terminazione dei consumer
        consumers.shutdown();
        consumers.awaitTermination(1, TimeUnit.MINUTES);

        System.out.println("Elaborazione completata!");
    }
}

6. Visualizzazione: schema di funzionamento della pipeline

flowchart LR
    A[Producer: legge il file] -- put() --> Q[BlockingQueue]
    Q -- take() --> C1[Consumer 1: conta le parole]
    Q -- take() --> C2[Consumer 2: conta le parole]
    Q -- take() --> C3[Consumer 3: conta le parole]
    A -.->|poison pill| Q
    Q -.->|poison pill| C1
    Q -.->|poison pill| C2
    Q -.->|poison pill| C3

7. Dettagli utili e best practice

• Limitate la dimensione della coda — protegge dall'esaurimento della memoria.
• Usate la “poison pill” per la terminazione — altrimenti i consumer possono rimanere bloccati per sempre.
• Non usate null come “poison pill”, se nella coda possono esserci valori null reali. Meglio una stringa o un oggetto speciale.
• Gestite InterruptedException — è importante per terminare correttamente i thread.
• Usate ExecutorService per gestire il pool di thread — è più semplice e sicuro che creare i thread manualmente.

8. Errori tipici nell'implementazione della pipeline

Errore n. 1: Coda non limitata → OutOfMemoryError. Se si utilizza LinkedBlockingQueue senza limitare la capacità, il producer può saturare tutta la memoria se il consumer non tiene il passo.

Errore n. 2: Nessuna “poison pill” → i consumer restano bloccati. Se si dimentica di inserire le “pillole avvelenate”, i thread-consumer resteranno in attesa di nuovi dati all'infinito.

Errore n. 3: Inserita una sola “poison pill”, ma i consumer sono più di uno. Ogni thread-consumer ha bisogno della propria “pillola” — altrimenti non termineranno tutti.

Errore n. 4: InterruptedException non gestita. Se un thread viene interrotto, bisogna terminare correttamente il lavoro (ripristinare il flag di interruzione), altrimenti sono possibili thread “bloccati”.

Errore n. 5: Uso di una variabile condivisa tra thread senza sincronizzazione. Non reinventare la ruota — usa BlockingQueue, non liste o array normali.