Cours : Concurrence en Java

1. Thread Lifecycle

1.1 États d'un Thread

                  ┌──────────┐
                  │   NEW    │ (créé mais non démarré)
                  └────┬─────┘
                       │ start()
                       ↓
                  ┌──────────┐
         ┌───────│ RUNNABLE │────────┐
         │       └────┬─────┘        │
         │            │              │
         ↓            ↓              ↓
  ┌───────────┐ ┌──────────┐ ┌──────────┐
  │  BLOCKED  │ │ WAITING  │ │TIMED_WAIT│
  │(synchronized)│(join,wait)│ │(sleep)   │
  └───────────┘ └──────────┘ └──────────┘
         │            │              │
         └────────────┴──────────────┘
                        ↓
                  ┌──────────┐
                  │TERMINATED│
                  └──────────┘

1.2 Création de Threads

// 1. Hériter de Thread
class MonThread extends Thread {
    @Override
    public void run() {
        System.out.println("Dans mon thread");
    }
}
new MonThread().start();

// 2. Implémenter Runnable
Thread t = new Thread(() -> System.out.println("Runnable"));
t.start();

// 3. Factory methods (Java 21+)
Thread vThread = Thread.ofVirtual()
    .name("virtual-")
    .start(() -> System.out.println("Virtual thread"));

Thread pThread = Thread.ofPlatform()
    .name("platform-")
    .daemon(true)
    .start(() -> System.out.println("Platform thread"));

2. synchronized et volatile

2.1 synchronized

public class Compteur {
    private int count = 0;
    
    // Méthode synchronisée (verrou sur l'instance)
    public synchronized void incrementer() {
        count++;
    }
    
    // Bloc synchronisé
    public void incrementer2() {
        synchronized (this) {
            count++;
        }
    }
    
    // Méthode statique synchronisée (verrou sur la classe)
    public static synchronized void reset() {
        // ...
    }
}

2.2 volatile

public class Flag {
    // volatile garantit la visibilité entre threads
    private volatile boolean running = true;
    
    public void arreter() { running = false; }
    
    public void travailler() {
        while (running) {
            // Traitement
        }
    }
}

Différence clé :

• volatile : visibilité seulement (pas d'atomicité)
• synchronized : visibilité + atomicité + exclusion mutuelle

3. Locks

3.1 ReentrantLock

private final Lock lock = new ReentrantLock();

public void methode() {
    lock.lock();
    try {
        // Section critique
    } finally {
        lock.unlock(); // Toujours dans finally !
    }
}

// Avec tryLock (non-bloquant)
if (lock.tryLock(1, TimeUnit.SECONDS)) {
    try {
        // Section critique
    } finally {
        lock.unlock();
    }
} else {
    System.out.println("Verrou non obtenu");
}

3.2 ReadWriteLock

private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
private final Lock readLock = rwLock.readLock();
private final Lock writeLock = rwLock.writeLock();
private Map<String, String> cache = new HashMap<>();

public String lire(String key) {
    readLock.lock();
    try {
        return cache.get(key);
    } finally {
        readLock.unlock();
    }
}

public void ecrire(String key, String value) {
    writeLock.lock();
    try {
        cache.put(key, value);
    } finally {
        writeLock.unlock();
    }
}

3.3 StampedLock (Java 8+)

private final StampedLock stampedLock = new StampedLock();
private double x, y;

public void move(double deltaX, double deltaY) {
    long stamp = stampedLock.writeLock();
    try {
        x += deltaX;
        y += deltaY;
    } finally {
        stampedLock.unlockWrite(stamp);
    }
}

// Optimistic read (non-bloquant)
public double distanceFromOrigin() {
    long stamp = stampedLock.tryOptimisticRead();
    double currentX = x, currentY = y;
    if (!stampedLock.validate(stamp)) {
        stamp = stampedLock.readLock();
        try {
            currentX = x;
            currentY = y;
        } finally {
            stampedLock.unlockRead(stamp);
        }
    }
    return Math.sqrt(currentX * currentX + currentY * currentY);
}

4. ExecutorService

4.1 Types de Pools

// Nombre fixe de threads
ExecutorService fixedPool = Executors.newFixedThreadPool(4);

// Pool extensible
ExecutorService cachedPool = Executors.newCachedThreadPool();

// Un seul thread
ExecutorService singleThread = Executors.newSingleThreadExecutor();

// Pool avec scheduling
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

// Virtual Threads (Java 21+)
ExecutorService vtPool = Executors.newVirtualThreadPerTaskExecutor();

// Work-stealing pool (ForkJoin)
ExecutorService workStealing = Executors.newWorkStealingPool();

4.2 Soumission de Tâches

ExecutorService executor = Executors.newFixedThreadPool(4);

// execute() : Runnable, pas de retour
executor.execute(() -> System.out.println("Tâche"));

// submit() : Callable, retourne Future<T>
Future<String> future = executor.submit(() -> {
    Thread.sleep(1000);
    return "Résultat";
});

// invokeAll() : plusieurs tâches
List<Callable<String>> tasks = List.of(
    () -> "Tâche 1",
    () -> "Tâche 2"
);
List<Future<String>> results = executor.invokeAll(tasks);

// invokeAny() : retourne le premier résultat
String first = executor.invokeAny(tasks);

4.3 Arrêt de l'ExecutorService

ExecutorService executor = Executors.newFixedThreadPool(4);

// Arrêt progressif (n'accepte plus de nouvelles tâches)
executor.shutdown();

// Attend la fin des tâches en cours (avec timeout)
if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
    executor.shutdownNow(); // Force l'arrêt
}

// Java 19+ : try-with-resources
try (var exec = Executors.newVirtualThreadPerTaskExecutor()) {
    exec.submit(() -> System.out.println("Auto-fermeture"));
} // shutdown() automatique

5. CompletableFuture

5.1 Création

// Future simple
CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "Hello");

// Avec Executor personnalisé
var executor = Executors.newFixedThreadPool(4);
CompletableFuture<String> future2 = CompletableFuture
    .supplyAsync(() -> "Hello", executor);

// Future terminé
CompletableFuture<String> completed = CompletableFuture
    .completedFuture("Valeur");

// runAsync (Runnable)
CompletableFuture<Void> run = CompletableFuture
    .runAsync(() -> System.out.println("Run"));

5.2 Transformation

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "Hello")
    .thenApply(s -> s + " World")       // sync transformation
    .thenApplyAsync(s -> s.toUpperCase()); // async transformation

// thenAccept : Consumer (pas de retour)
future.thenAccept(System.out::println);

// thenRun : Runnable
future.thenRun(() -> System.out.println("Terminé"));

5.3 Composition

// thenCompose : flatMap (Future<Future> → Future)
CompletableFuture<String> composed = CompletableFuture
    .supplyAsync(() -> "Hello")
    .thenCompose(s -> CompletableFuture
        .supplyAsync(() -> s + " World"));

// thenCombine : combine deux Future indépendants
CompletableFuture<String> combined = 
    CompletableFuture.supplyAsync(() -> "Hello")
    .thenCombine(
        CompletableFuture.supplyAsync(() -> " World"),
        (s1, s2) -> s1 + s2
    );

// allOf : attend plusieurs Future
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "A");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "B");
CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2);

// anyOf : premier terminé
CompletableFuture<Object> any = CompletableFuture.anyOf(f1, f2);

5.4 Gestion d'Erreurs

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        if (Math.random() > 0.5) throw new RuntimeException("Erreur");
        return "Succès";
    })
    .exceptionally(ex -> "Fallback: " + ex.getMessage())
    .handle((result, ex) -> 
        ex == null ? result : "Géré: " + ex.getMessage());

// whenComplete : callback quel que soit le résultat
future.whenComplete((result, ex) -> {
    if (ex != null) System.err.println("Erreur: " + ex);
    else System.out.println("Succès: " + result);
});

5.5 Exemple Complet

public CompletableFuture<UserData> fetchUserData(Long userId) {
    return CompletableFuture
        .supplyAsync(() -> fetchUser(userId))
        .thenCompose(user -> {
            CompletableFuture<Address> address = 
                CompletableFuture.supplyAsync(() -> fetchAddress(user));
            CompletableFuture<List<Order>> orders = 
                CompletableFuture.supplyAsync(() -> fetchOrders(user));
            return address.thenCombine(orders, 
                (addr, ords) -> new UserData(user, addr, ords));
        })
        .exceptionally(ex -> {
            log.error("Erreur chargement user {}", userId, ex);
            return UserData.DEFAULT;
        });
}

6. ForkJoinPool

6.1 Principe (Divide & Conquer)

// Tâche de somme récursive
class SumTask extends RecursiveTask<Long> {
    private static final int SEUIL = 10_000;
    private final long[] array;
    private final int start, end;
    
    SumTask(long[] array, int start, int end) {
        this.array = array;
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= SEUIL) {
            long sum = 0;
            for (int i = start; i < end; i++) sum += array[i];
            return sum;
        }
        int mid = start + length / 2;
        var left = new SumTask(array, start, mid);
        var right = new SumTask(array, mid, end);
        left.fork();  // Exécute en parallèle
        return right.compute() + left.join();
    }
}

// Utilisation
long[] data = new long[1_000_000];
// ... remplir le tableau
long sum = ForkJoinPool.commonPool()
    .invoke(new SumTask(data, 0, data.length));

7. Virtual Threads

7.1 Création et Utilisation

// Création directe
Thread vt = Thread.startVirtualThread(() -> {
    System.out.println("Virtual Thread: " + Thread.currentThread());
});

// Avec nom et factory
ThreadFactory factory = Thread.ofVirtual()
    .name("worker-", 0)
    .factory();

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 10_000; i++) {
        executor.submit(() -> {
            Thread.sleep(100); // I/O simulé
            return 42;
        });
    }
} // Attend la fin de toutes les tâches

// Détection
System.out.println(Thread.currentThread().isVirtual());

7.2 Pinnage (synchronized)

// ❌ Problème : synchronized + Virtual Thread
private final Object lock = new Object();

synchronized (lock) { // Le VT capture le thread porteur
    Thread.sleep(100); // Thread porteur bloqué !
}

// ✅ Solution : ReentrantLock
private final Lock lock = new ReentrantLock();

lock.lock();
try {
    Thread.sleep(100); // VT libère le thread porteur
} finally {
    lock.unlock();
}

8. Structured Concurrency (Preview Java 21+)

// ShutdownOnFailure : si une tâche échoue, toutes sont annulées
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> user = scope.fork(() -> fetchUser(id));
    Future<Integer> orders = scope.fork(() -> fetchOrderCount(id));
    
    scope.join();           // Attend toutes les tâches
    scope.throwIfFailed();  // Lance si une a échoué
    
    return new Response(user.resultNow(), orders.resultNow());
}

// ShutdownOnSuccess : premier résultat
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    scope.fork(() -> fetchFrom("server1"));
    scope.fork(() -> fetchFrom("server2"));
    scope.fork(() -> fetchFrom("server3"));
    
    return scope.join().result(); // Retourne le premier succès
}

9. Collections Concurrentes

9.1 Map Concurrentes

// HashMap non thread-safe
Map<String, String> bad = new HashMap<>(); // ❌ ConcurrentModification

// HashTable (thread-safe mais lent)
Map<String, String> old = new Hashtable<>();

// ConcurrentHashMap (performant)
ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();

map.putIfAbsent("key", "value");
map.computeIfAbsent("key", k -> computeValue(k));
map.computeIfPresent("key", (k, v) -> v + " modifié");
map.merge("key", "value", (v1, v2) -> v1 + v2);
map.forEach(100, (k, v) -> System.out.println(k + "=" + v)); // parallélisme
map.search(100, (k, v) -> v.startsWith("A") ? k : null);
map.reduce(100, (k, v) -> v.length(), Integer::sum);

9.2 List/Set Concurrentes

// CopyOnWriteArrayList : lecture rapide, écriture lente
// Idéale pour : lectures > écritures
List<String> list = new CopyOnWriteArrayList<>();
list.add("a");
list.add("b");
for (var s : list) { // Pas de ConcurrentModificationException
    System.out.println(s);
}

// CopyOnWriteArraySet
Set<String> set = new CopyOnWriteArraySet<>();

// ConcurrentLinkedQueue / Deque
Queue<String> queue = new ConcurrentLinkedQueue<>();
Deque<String> deque = new ConcurrentLinkedDeque<>();

// BlockingQueue (pour producteur-consommateur)
BlockingQueue<String> blocking = new LinkedBlockingQueue<>(100);
// producteur
blocking.put("item");    // Bloque si pleine
// consommateur
String item = blocking.take(); // Bloque si vide

9.3 Synchronizers

// CountDownLatch : attendre que N opérations soient terminées
CountDownLatch latch = new CountDownLatch(3);
// Dans chaque thread : latch.countDown();
latch.await(); // Attend que le compteur atteigne 0

// CyclicBarrier : N threads s'attendent mutuellement
CyclicBarrier barrier = new CyclicBarrier(3, () -> 
    System.out.println("Tous prêts !"));
// Dans chaque thread : barrier.await();

// Semaphore : contrôle d'accès
Semaphore semaphore = new Semaphore(3); // 3 accès simultanés
semaphore.acquire();  // Bloque si complet
try { /* section */ }
finally { semaphore.release(); }

// Exchanger : échange de données entre 2 threads
Exchanger<String> exchanger = new Exchanger<>();
// Thread A : String data = exchanger.exchange("dataFromA");
// Thread B : String data = exchanger.exchange("dataFromB");

10. Bonnes Pratiques

// 1. Toujours utiliser ExecutorService, jamais new Thread()
ExecutorService exec = Executors.newFixedThreadPool(4);

// 2. Toujours fermer ExecutorService
try (var exec = Executors.newVirtualThreadPerTaskExecutor()) { }

// 3. Lock toujours dans try/finally
lock.lock();
try { /* ... */ } finally { lock.unlock(); }

// 4. Préférer CompletableFuture à Future
CompletableFuture.supplyAsync(() -> service.call());

// 5. Collections concurrentes
ConcurrentHashMap<String, Data> map = new ConcurrentHashMap<>();

// 6. Atomic pour les compteurs simples
private final AtomicInteger counter = new AtomicInteger(0);

// 7. Immutabilité
public final class ImmutableData {
    private final List<String> items;
    public ImmutableData(List<String> items) {
        this.items = List.copyOf(items); // Défensive
    }
    public List<String> getItems() { return items; }
}

11. Résumé