Contenu du cours

Multithreading en Java

1. Notions de Base du Multithreading

Qu'est-ce que le Multithreading Différence Entre le Multithreading et le Monothreading Threads en Java Synchronisation des Threads Performance et Défis Atomicité Tâche de Multithreading

2. Collections Synchronisées

Collections avec Support Multithreading API de Flux Parallèle BlockingQueue et Ses Implémentations Défi BlockingQueue ConcurrentMap et Ses Implémentations Défi ConcurrentMap Collection Copy-On-Write Exemples d'Utilisation des Méthodes de Collections

3. Mécanismes de Synchronisation de Haut Niveau

Verrou et Condition Défi Verrou et Condition Sémaphore et Barrière Défi Sémaphore Variables Atomiques Exécuteurs et Pool de Threads Défi Exécuteurs

4. Meilleures Pratiques de Multithreading

Producteur-Consommateur ForkJoinPool Défi ForkJoinPool ThreadLocal Principes de Base de la Programmation Asynchrone CompletableFuture Défi CompletableFuture Résumé Multithreading

Sémaphore et Barrière

Dans les programmes multithread, il est souvent nécessaire de contrôler l'accès aux ressources ou de synchroniser l'exécution des threads. Semaphore et Barrier sont des mécanismes de synchronisation de haut niveau qui aident à relever ces défis.

Aujourd'hui, nous allons explorer chacun de ces mécanismes en séquence et comprendre leurs différences. Commençons par Semaphore.

Semaphore en Java sont implémentés via la classe java.util.concurrent.Semaphore.

Constructeurs

Semaphore(int permits): Constructeur qui crée un semaphore avec un certain nombre de permissions. Les permissions représentent le nombre d'accès à la ressource partagée.

Main


              1
            
 Semaphore semaphore = new Semaphore(20);

Semaphore(int permits, boolean fair): Constructeur qui fournit une résolution premier arrivé, premier servi.

Main


              1
            
Semaphore semaphore = new Semaphore(20, true);

Si fair est défini sur true, le semaphore accordera les permissions dans l'ordre premier entré, premier sorti (FIFO), ce qui peut aider à éviter la famine. Par défaut - false.

Méthodes Principales

La méthode acquire() demande une seule permission. Si une permission est disponible, elle est accordée immédiatement ; sinon, le thread est bloqué jusqu'à ce qu'une permission devienne disponible. Une fois qu'une tâche est terminée, la méthode release() est utilisée pour libérer la permission, la retournant au semaphore. Si d'autres threads attendaient une permission, l'un d'eux sera débloqué.

Imaginez un parking avec un nombre limité de places. Le semaphore fonctionne comme un contrôleur, gardant une trace des places disponibles et refusant l'accès une fois le parking plein.

Main


              1234567891011121314151617181920212223242526272829303132
            
package com.example;

import java.util.concurrent.Semaphore;

public class Main {
    private final Semaphore semaphore;

    public Main(int slots) {
        semaphore = new Semaphore(slots);
    }

    public void parkCar() {
        try {
            semaphore.acquire(); // Request a parking spot
            System.out.println("Car parked. Available slots: " + semaphore.availablePermits());
            Thread.sleep(2000); // Simulate parking time
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            semaphore.release(); // Release the parking spot
            System.out.println("Car left. Available slots: " + semaphore.availablePermits());
        }
    }

    public static void main(String[] args) {
        Main parking = new Main(3); // Parking lot with 3 spots

        for (int i = 0; i < 5; i++) {
            new Thread(parking::parkCar).start();
        }
    }
}

Vous pouvez également découvrir combien de permissions sont actuellement disponibles dans Semaphore en utilisant la méthode int availablePermits(). Vous pouvez également essayer d'obtenir une permission en utilisant la méthode boolean tryAcquire(), qui renvoie true si une permission a été obtenue et false sinon.

Main


              1234567891011121314151617181920212223242526272829303132333435363738394041
            
package com.example;

import java.util.concurrent.Semaphore;

public class Main {
    // Define the maximum number of permits available
    private static final int MAX_PERMITS = 3;
    private static Semaphore semaphore = new Semaphore(MAX_PERMITS);

    public static void main(String[] args) {
        // Create and start 5 worker threads
        for (int i = 1; i <= 5; i++) {
            new Thread(new Worker(), "Worker-" + i).start();
        }
    }

    static class Worker implements Runnable {
        @Override
        public void run() {
            String name = Thread.currentThread().getName();
            System.out.println(name + " trying to acquire a permit...");

            // Try to acquire a permit
            if (semaphore.tryAcquire()) {
                try {
                    System.out.println(name + " acquired a permit! Available permits: " + semaphore.availablePermits());
                    Thread.sleep(1000); // Simulate work
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    // Release the permit after the work is done
                    semaphore.release();
                    System.out.println(name + " released a permit. Available permits: " + semaphore.availablePermits());
                }
            } else {
                // Inform if the permit could not be acquired
                System.out.println(name + " could not acquire a permit. Available permits: " + semaphore.availablePermits());
            }
        }
    }
}

Résultats

En d'autres termes, un Semaphore est utile lorsque vous devez fournir un accès limité simultané à un segment de code spécifique. Le seul inconvénient est le potentiel de blocage si les threads sont bloqués dans le mauvais ordre.

Passons maintenant au prochain mécanisme de synchronisation qui est encore plus simple à utiliser mais sera 100 pour cent précieux pour vos besoins.

CyclicBarrier

Barrier en Java est représenté par la classe java.util.concurrent.CyclicBarrier. Les principales méthodes de CyclicBarrier incluent :

Constructeurs CyclicBarrier

CyclicBarrier(int parties): Constructeur qui crée une barrière qui bloque les threads jusqu'à ce qu'un certain nombre de threads (parties) arrivent.

Main


              1
            
CyclicBarrier cyclicBarrier = new CyclicBarrier(10);

CyclicBarrier(int parties, Runnable barrierAction): Constructeur qui crée une barrière avec un nombre donné de parties et une action (barrierAction) qui est exécutée lorsque toutes les parties arrivent à la barrière.

Main


              1234567
            
Runnable task = () -> {
    // This task will be executed when all parties have reached the barrier
    System.out.println("Hello))");
};

// Create a `CyclicBarrier` for 10 parties with a barrier action
CyclicBarrier cyclicBarrier = new CyclicBarrier(10, task);

Méthodes CyclicBarrier

La méthode principale await() qui est utilisée comme une barrière et ne laisse pas le fil aller plus loin jusqu'à ce que tous les fils atteignent cette méthode. Retourne un numéro de séquence indiquant l'ordre d'arrivée des participants.

Main


              1234567891011121314151617181920212223242526272829303132333435363738394041424344
            
package com.example;

import java.util.concurrent.CyclicBarrier;

public class Main {
    public static void main(String[] args) {
        // Create a `CyclicBarrier` for 3 parties with a barrier action
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {
            System.out.println("All parties have reached the barrier. Barrier action executed.");
        });

        // Create and start 3 worker threads
        for (int i = 1; i <= 3; i++) {
            new Thread(new Worker(barrier), "Worker-" + i).start();
        }
    }

    static class Worker implements Runnable {
        private CyclicBarrier barrier;

        Worker(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {
            String name = Thread.currentThread().getName();
            System.out.println(name + " is working...");

            try {
                // Simulate work
                Thread.sleep((int) (Math.random() * 1000));

                System.out.println(name + " is waiting at the barrier.");
                barrier.await(); // Wait at the barrier

                // This code will execute after all parties have reached the barrier
                System.out.println(name + " has crossed the barrier.");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Il peut arriver que tous les threads ne parviennent pas à atteindre la barrière et que le programme se bloque, à cet effet, la méthode await(long timeout, TimeUnit unit) est utilisée, qui est similaire à await(), mais avec un délai d'attente. Si le délai d'attente expire avant l'arrivée de tous les participants, la méthode génère une exception TimeoutException.

Vous pouvez également connaître le nombre de participants requis pour compléter la barrière int getParties() et sa méthode similaire int getNumberWaiting() qui renvoie le nombre de participants actuellement en attente à la barrière.

Main


              12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849
            
package com.example;

import java.util.concurrent.CyclicBarrier;

public class Main {
    public static void main(String[] args) {
        // Create a `CyclicBarrier` for 3 parties with a barrier action
        CyclicBarrier barrier = new CyclicBarrier(3, () -> {
            System.out.println("All parties have reached the barrier. Barrier action executed.");
        });

        System.out.println("Total number of parties required to complete the barrier: " + barrier.getParties());

        // Create and start 3 worker threads
        for (int i = 1; i <= 3; i++) {
            new Thread(new Worker(barrier), "Worker-" + i).start();
        }
    }

    static class Worker implements Runnable {
        private CyclicBarrier barrier;

        Worker(CyclicBarrier barrier) {
            this.barrier = barrier;
        }

        @Override
        public void run() {
            String name = Thread.currentThread().getName();
            System.out.println(name + " is working...");

            try {
                // Simulate work
                Thread.sleep((int) (Math.random() * 1000));

                System.out.println(name + " is waiting at the barrier.");
                barrier.await(); // Wait at the barrier

                // This code will execute after all parties have reached the barrier
                System.out.println(name + " has crossed the barrier.");
            } catch (Exception e) {
                e.printStackTrace();
            }

            // Print the number of participants currently waiting at the barrier
            System.out.println("Number of participants currently waiting at the barrier: " + barrier.getNumberWaiting());
        }
    }
}

Il est également possible de vérifier si la barrière a été détruite si l'un des threads est interrompu ou si le délai d'attente a expiré en utilisant la méthode boolean isBroken(). Si elle a été brisée, vous pouvez utiliser la méthode void reset() qui restaurera simplement la barrière.

Main


              12345
            
// Check if the barrier is broken and reset it if necessary
if (barrier.isBroken()) {
    System.out.println("Barrier is broken. Resetting the barrier.");
    barrier.reset();
}

Remarque

Il faut prendre en compte que certains flux peuvent ne pas atteindre la barrière en raison d'une erreur ou autre chose et il est alors clair que la barrière n'autorisera pas les flux qui attendent actuellement à la barrière

Tout était clair ?

Merci pour vos commentaires !

Section 3. Chapitre 3