import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class CorruptedHashMap {

    public static void main(String[] args) throws InterruptedException {

        for (int attempt = 1; attempt <= 5; attempt++) {

            Map<Integer, String> map = new HashMap<>();
            final int PER_THREAD = 100_000;

            Thread t1 = new Thread(() -> {
                for (int i = 0; i < PER_THREAD; i++) map.put(i, "A" + i);
            }, "writer-1");

            Thread t2 = new Thread(() -> {
                for (int i = PER_THREAD; i < PER_THREAD * 2; i++) map.put(i, "B" + i);
            }, "writer-2");

            t1.start();
            t2.start();

            // Deadline: if the HashMap gets corrupted, a thread can end up
            // in an infinite loop inside put() or get().
            t1.join(5000);
            t2.join(5000);

            int expected = PER_THREAD * 2;
            if (t1.isAlive() || t2.isAlive()) {
                System.out.printf("Attempt %d: INFINITE LOOP. "
                        + "t1=%s t2=%s  <-- corrupted structure%n",
                        attempt, t1.getState(), t2.getState());
                System.out.println("  (look at the CPU usage: one core at 100%)");
                System.exit(1);
            }
            System.out.printf("Attempt %d: expected %d, actual %d, lost %d%n",
                    attempt, expected, map.size(), expected - map.size());
        }
    }
}

Attempt 1: expected 200000, actual 187341, lost 12659
Attempt 2: expected 200000, actual 193028, lost 6972
Attempt 3: INFINITE LOOP. t1=RUNNABLE t2=RUNNABLE  <-- corrupted structure
  (look at the CPU usage: one core at 100%)

List<String> list = new ArrayList<>(List.of("a", "b", "c"));
for (String s : list) {
    list.remove(s);       // ConcurrentModificationException
}

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class ConcurrentFailFast {

    public static void main(String[] args) throws InterruptedException {

        List<String> catalog = new ArrayList<>();
        for (int i = 0; i < 10_000; i++) catalog.add("978-" + i);

        Thread reader = new Thread(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    int n = 0;
                    for (String isbn : catalog) {     // iteration
                        n += isbn.length();
                    }
                }
            } catch (java.util.ConcurrentModificationException e) {
                System.out.println("[reader] ConcurrentModificationException: "
                        + "another thread modified the catalogue while iterating");
            }
        }, "bibliotech-reader");

        Thread writer = new Thread(() -> {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    catalog.add("978-new");
                    TimeUnit.MILLISECONDS.sleep(1);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "bibliotech-writer");

        reader.start();
        writer.start();

        TimeUnit.SECONDS.sleep(2);
        reader.interrupt();
        writer.interrupt();
    }
}

[reader] ConcurrentModificationException: another thread modified the catalogue while iterating

import java.util.*;

Map<String, Material> map = Collections.synchronizedMap(new HashMap<>());
List<Material> list       = Collections.synchronizedList(new ArrayList<>());
Set<String> set           = Collections.synchronizedSet(new HashSet<>());

// Roughly, what Collections.synchronizedMap does:
public V get(Object key) {
    synchronized (mutex) { return m.get(key); }
}
public V put(K key, V value) {
    synchronized (mutex) { return m.put(key, value); }
}

List<Material> list = Collections.synchronizedList(new ArrayList<>());

// WRONG: every iterator call is synchronized, but the
// WHOLE ITERATION is not. ConcurrentModificationException guaranteed.
for (Material m : list) {
    process(m);
}

// RIGHT: synchronise the entire iteration on the wrapper itself.
// It is what the javadoc of Collections.synchronizedList explicitly
// requires, and hardly anybody reads that part.
synchronized (list) {
    for (Material m : list) {
        process(m);       // CAREFUL: 'process' runs WITH THE LOCK HELD
    }                     // (rule 3 of 08-04: no foreign code here)
}

Map<String, Integer> loansByIsbn = Collections.synchronizedMap(new HashMap<>());

// WRONG: check-then-act. Each call is synchronized,
// but the GAP between them is not.
if (!loansByIsbn.containsKey(isbn)) {           // (1) thread A: not there
    loansByIsbn.put(isbn, 1);                   // (3) thread A writes 1
}                                               // (2) thread B also saw "not there"
                                                // (4) thread B writes 1 -> one is lost

// WRONG: read-modify-write, the same problem as 'counter++' (08-04)
Integer n = loansByIsbn.get(isbn);              // (1) reads 5
loansByIsbn.put(isbn, n + 1);                   // (3) writes 6
                                                // another thread also read 5 and
                                                // also writes 6: one loan lost

// Correct but clumsy: an external lock over the collection.
synchronized (loansByIsbn) {
    Integer n = loansByIsbn.get(isbn);
    loansByIsbn.put(isbn, n == null ? 1 : n + 1);
}

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

public class MapComparison {

    static long measure(Map<Integer, String> map, int threads, int opsPerThread,
                        double readRatio) throws InterruptedException {

        // Preload so the reads hit.
        for (int i = 0; i < 10_000; i++) map.put(i, "value-" + i);

        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch finishLine = new CountDownLatch(threads);

        for (int h = 0; h < threads; h++) {
            new Thread(() -> {
                try {
                    startGate.await();
                    ThreadLocalRandom random = ThreadLocalRandom.current();
                    for (int i = 0; i < opsPerThread; i++) {
                        int key = random.nextInt(10_000);
                        if (random.nextDouble() < readRatio) {
                            map.get(key);
                        } else {
                            map.put(key, "new-" + i);
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    finishLine.countDown();
                }
            }, "access-" + h).start();
        }

        long start = System.nanoTime();
        startGate.countDown();       // they all start at once (08-05)
        finishLine.await();
        return (System.nanoTime() - start) / 1_000_000;
    }

    public static void main(String[] args) throws InterruptedException {

        final int THREADS = 16;
        final int OPS = 200_000;

        System.out.printf("%d threads x %d operations%n%n", THREADS, OPS);
        System.out.printf("%-28s | %10s | %10s%n", "Implementation", "90% reads", "50% reads");
        System.out.println("-----------------------------|------------|------------");

        long s90 = measure(Collections.synchronizedMap(new HashMap<>()), THREADS, OPS, 0.9);
        long s50 = measure(Collections.synchronizedMap(new HashMap<>()), THREADS, OPS, 0.5);
        System.out.printf("%-28s | %8d ms | %8d ms%n", "synchronizedMap", s90, s50);

        long c90 = measure(new ConcurrentHashMap<>(), THREADS, OPS, 0.9);
        long c50 = measure(new ConcurrentHashMap<>(), THREADS, OPS, 0.5);
        System.out.printf("%-28s | %8d ms | %8d ms%n", "ConcurrentHashMap", c90, c50);

        System.out.printf("%nImprovement: %.1fx (90%% reads), %.1fx (50%% reads)%n",
                (double) s90 / c90, (double) s50 / c50);
    }
}

16 threads x 200000 operations

Implementation               |  90% reads |  50% reads
-----------------------------|------------|------------
synchronizedMap              |     3187 ms |     3402 ms
ConcurrentHashMap            |      184 ms |      271 ms

Improvement: 17.3x (90% reads), 12.6x (50% reads)

import java.util.concurrent.ConcurrentHashMap;
import java.util.Map;

public class AtomicMapOperations {

    private final Map<String, Integer> loansByIsbn = new ConcurrentHashMap<>();
    private final Map<String, Card> cardCache = new ConcurrentHashMap<>();

    /** Check-then-act, solved: a single atomic operation. */
    public boolean registerFirstLoan(String isbn) {
        // Returns null if it did NOT exist (and inserts it), or the previous value.
        return loansByIsbn.putIfAbsent(isbn, 1) == null;
    }

    /** Read-modify-write, solved with merge. */
    public void countLoan(String isbn) {
        // If absent -> stores 1. If present -> applies Integer::sum
        // between the current value and the 1 we pass. All atomic.
        loansByIsbn.merge(isbn, 1, Integer::sum);
    }

    /** The same with compute, more explicit. */
    public void countLoanAlternative(String isbn) {
        loansByIsbn.compute(isbn, (key, current) ->
                current == null ? 1 : current + 1);
    }

    /**
     * A lazy cache, solved with computeIfAbsent.
     * It replaces the three phases of exercise 3 of 08-04:
     * the function runs AT MOST ONCE per key, even if
     * ten threads ask for it at once. The other nine wait and
     * receive the same object: there is no duplicated computation.
     */
    public Card getCard(String isbn) {
        return cardCache.computeIfAbsent(isbn, this::buildCard);
    }

    /** Remove only if the value is the expected one: compare-and-remove. */
    public boolean returnIfLast(String isbn) {
        return loansByIsbn.remove(isbn, 1);
    }

    /** Download counter with no NullPointerException. */
    public int downloadsOf(String isbn) {
        return loansByIsbn.getOrDefault(isbn, 0);
    }

    private Card buildCard(String isbn) { /* expensive query */ return null; }
}

// FORBIDDEN 1: modifying the SAME map inside the function.
// It can cause a deadlock or corrupt the structure.
map.computeIfAbsent(k, key -> {
    map.put("another", "thing");     // NEVER
    return compute(key);
});

// FORBIDDEN 2: long or blocking operations.
// It locks the whole bucket, and the whole application suffers
// (rule 2 of 08-04: no I/O inside the lock).
map.computeIfAbsent(k, key -> readFromDisk(key));       // WRONG if it is slow

// FORBIDDEN 3: calling foreign code (listeners, callbacks).
// Rule 3 of 08-04, applied here.

// CORRECT when the computation is expensive: compute outside and publish with putIfAbsent.
Card computed = readFromDisk(isbn);          // with no lock at all
Card existing = map.putIfAbsent(isbn, computed);
Card result = (existing != null) ? existing : computed;
// There may be duplicated computation if two threads coincide, but
// the result is correct and the bucket is not locked.

Map<String, Material> catalog = new ConcurrentHashMap<>();

// SAFE: it throws no exception, blocks nobody, and no writer
// sits waiting for you to finish walking 100,000 entries.
for (Map.Entry<String, Material> e : catalog.entrySet()) {
    process(e.getValue());
}
// If another thread adds an entry while you iterate, you may see it
// and you may not. What will NOT happen is the loop failing.

// WRONG: check-then-act on an approximate size.
if (catalog.size() < LIMIT) {
    catalog.put(isbn, material);     // the size may have changed in between
}

// The size of a concurrent structure is a STATISTICAL figure,
// useful for logs and metrics, not for control decisions.
LOG.log(Level.INFO, "Catalogue with ~{0} materials", catalog.size());

package com.nexussoftware.bibliotech.service;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * Registry of catalogue event listeners.
 *
 * A PERFECT use case for CopyOnWriteArrayList:
 *  - ~10 listeners register at start-up and almost never change.
 *  - They are walked on EVERY catalogue operation: thousands of times a minute.
 *  - Read/write ratio: 100,000 to 1.
 */
public class ListenerRegistry {

    private final List<CatalogListener> listeners = new CopyOnWriteArrayList<>();

    public void register(CatalogListener l)   { listeners.add(l); }
    public void unregister(CatalogListener l) { listeners.remove(l); }

    /**
     * Notifies every listener.
     *
     * TWO decisive advantages over a synchronized list:
     *  1. There is NO lock during the walk: listeners can
     *     register or unregister listeners without deadlocking.
     *  2. It is rule 3 of 08-04 —never call foreign code holding a
     *     lock— honoured automatically by the collection.
     */
    public void notifyAdded(Material m) {
        for (CatalogListener l : listeners) {
            try {
                l.materialAdded(m);
            } catch (RuntimeException e) {
                // A faulty listener must not stop the others
                // being told (degradation policy of 06-07).
                LOG.log(Level.WARNING, "Listener failed: " + l, e);
            }
        }
    }
}

// DISASTER: 10,000 writes on a growing list.
// Every add() copies the whole array: 1 + 2 + ... + 10,000 ≈ 50 million
// reference copies. Seconds where it should be milliseconds.
List<Material> list = new CopyOnWriteArrayList<>();
for (int i = 0; i < 10_000; i++) {
    list.add(materials.get(i));        // O(n) each -> O(n²) in total
}

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

Queue<Reservation> pending = new ConcurrentLinkedQueue<>();

pending.offer(reservation);        // add: never blocks, never fails
Reservation r = pending.poll();    // take: returns NULL if it is empty

// ANTIPATTERN with ConcurrentLinkedQueue: polling (08-03, section 6).
while (!Thread.currentThread().isInterrupted()) {
    Reservation r = pending.poll();
    if (r == null) {
        Thread.sleep(100);       // wakes 10 times a second for nothing
        continue;
    }
    serve(r);
}

import java.util.concurrent.*;

// 1. ARRAYBLOCKINGQUEUE: bounded. If it fills up, the producer WAITS.
//    That is BACKPRESSURE: the producer slows to the consumer's pace.
//    It is what avoids the OutOfMemoryError of unbounded queues (08-05).
BlockingQueue<Reservation> reservations = new ArrayBlockingQueue<>(100);

// 2. Bounded LINKEDBLOCKINGQUEUE: two internal locks (one for the
//    head and one for the tail), so a producer and a consumer
//    can work at the same time. Better with high concurrency.
BlockingQueue<Notice> notices = new LinkedBlockingQueue<>(500);

// 3. SYNCHRONOUSQUEUE: capacity ZERO. Every put() waits for a take().
//    It is a hand-to-hand transfer, with no store. It is the queue
//    newCachedThreadPool uses (08-05), hence its unlimited threads.
BlockingQueue<Task> handoff = new SynchronousQueue<>();

// 4. PRIORITYBLOCKINGQUEUE: the "smallest" according to the
//    Comparator (05-09) comes out first. The most overdue notices first.
BlockingQueue<Loan> byUrgency = new PriorityBlockingQueue<>(
        100, Comparator.comparingLong(Loan::daysLate).reversed());

// 5. DELAYQUEUE: the elements implement Delayed and cannot be
//    taken until their delay expires. Retries with a wait.
DelayQueue<NoticeRetry> retries = new DelayQueue<>();

package com.nexussoftware.bibliotech.service;

import com.nexussoftware.bibliotech.domain.Reservation;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Reservation processor with the producer-consumer pattern.
 *
 * The employees produce reservations from the menu (several threads);
 * a group of consumers serves them (check availability,
 * notify, register).
 *
 * The BOUNDED queue gives backpressure: if reservations arrive faster
 * than they are served, the menu slows down instead of piling
 * reservations up in memory until it runs out.
 */
public class ReservationProcessor implements AutoCloseable {

    private static final Logger LOG = Logger.getLogger(ReservationProcessor.class.getName());

    private static final int CAPACITY = 50;
    private static final int CONSUMERS = 4;

    private final BlockingQueue<Reservation> queue = new ArrayBlockingQueue<>(CAPACITY);
    private final ExecutorService consumers;

    private final AtomicInteger served = new AtomicInteger();
    private final AtomicInteger rejected = new AtomicInteger();
    private volatile boolean acceptingNew = true;

    public ReservationProcessor() {
        AtomicInteger n = new AtomicInteger(1);
        this.consumers = Executors.newFixedThreadPool(CONSUMERS,
                r -> new Thread(r, "bibliotech-reservations-" + n.getAndIncrement()));

        for (int i = 0; i < CONSUMERS; i++) {
            consumers.execute(this::consumerLoop);
        }
    }

    // ---------- PRODUCER ----------

    /**
     * Enqueues a reservation. BLOCKS if the queue is full: that is
     * backpressure, and it is a feature, not a defect.
     */
    public void enqueue(Reservation r) throws InterruptedException {
        if (!acceptingNew) {
            throw new IllegalStateException("The processor is shutting down");
        }
        queue.put(r);     // waits if it is full
    }

    /**
     * A variant that does not wait indefinitely: if there is no room
     * within 2 seconds, it rejects. That is right for a user interface:
     * better to say "the system is saturated" than to leave the menu hanging.
     */
    public boolean enqueueWithDeadline(Reservation r) throws InterruptedException {
        boolean accepted = queue.offer(r, 2, TimeUnit.SECONDS);
        if (!accepted) {
            rejected.incrementAndGet();
            LOG.log(Level.WARNING, "Reservation rejected through saturation: {0}", r.id());
        }
        return accepted;
    }

    // ---------- CONSUMER ----------

    private void consumerLoop() {
        String me = Thread.currentThread().getName();
        LOG.log(Level.INFO, "[{0}] consumer ready", me);
        try {
            while (true) {
                // take() BLOCKS without consuming CPU if the queue is empty.
                // It is the correct alternative to polling with sleep (08-03).
                Reservation r = queue.take();

                // POISON PILL: the shutdown signal (section 12).
                if (r == Reservation.END) {
                    LOG.log(Level.INFO, "[{0}] pill received, finishing", me);
                    return;
                }

                try {
                    serve(r);
                    served.incrementAndGet();
                } catch (Exception e) {
                    // An individual failure must NOT kill the consumer:
                    // if it dies, the pool silently loses capacity.
                    LOG.log(Level.WARNING, "[" + me + "] reservation failed: " + r.id(), e);
                }
            }
        } catch (InterruptedException e) {
            LOG.log(Level.INFO, "[{0}] consumer interrupted", me);
            Thread.currentThread().interrupt();      // restore (08-02)
        }
    }

    private void serve(Reservation r) throws InterruptedException {
        TimeUnit.MILLISECONDS.sleep(120);            // simulated I/O
        if (r.id().hashCode() % 23 == 0) {
            throw new IllegalStateException("material already on loan");
        }
    }

    // ---------- SHUTDOWN ----------

    /**
     * ORDERLY shutdown: stop accepting, insert one pill per
     * consumer and wait. Everything already queued gets processed.
     */
    @Override
    public void close() {
        acceptingNew = false;
        try {
            for (int i = 0; i < CONSUMERS; i++) {
                queue.put(Reservation.END);  // one per consumer
            }
            consumers.shutdown();
            if (!consumers.awaitTermination(30, TimeUnit.SECONDS)) {
                consumers.shutdownNow();
            }
        } catch (InterruptedException e) {
            consumers.shutdownNow();
            Thread.currentThread().interrupt();
        }
        LOG.log(Level.INFO, "Processor closed: {0} served, {1} rejected",
                new Object[] { served.get(), rejected.get() });
    }

    public int served()   { return served.get(); }
    public int rejected() { return rejected.get(); }
    public int pending()  { return queue.size(); }
}

public class ReservationDemo {

    public static void main(String[] args) throws InterruptedException {

        try (ReservationProcessor processor = new ReservationProcessor()) {

            // Three producers: three employees using the menu at once.
            Thread[] producers = new Thread[3];
            for (int p = 0; p < 3; p++) {
                final String employee = switch (p) {
                    case 0 -> "Marta Ruiz";
                    case 1 -> "Diego Alonso";
                    default -> "Nuria Vidal";
                };
                producers[p] = new Thread(() -> {
                    try {
                        for (int i = 1; i <= 40; i++) {
                            processor.enqueue(new Reservation(
                                    employee + "-R" + i, "978-000000000" + (i % 3 + 1)));
                            TimeUnit.MILLISECONDS.sleep(20);
                        }
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }, "menu-" + employee.split(" ")[0]);
                producers[p].start();
            }

            // Progress while they work.
            for (int t = 0; t < 10; t++) {
                System.out.printf("  served=%d  pending=%d%n",
                        processor.served(), processor.pending());
                TimeUnit.MILLISECONDS.sleep(400);
            }

            for (Thread p : producers) p.join();
            System.out.println("All the producers have finished");

        }   // close(): poison pills and wait for the drain
    }
}

  served=0  pending=3
  served=12  pending=14
  served=25  pending=26
  served=38  pending=39
  served=51  pending=50   <-- queue FULL: the producers slow down
  served=64  pending=50
  served=77  pending=43
  served=90  pending=30
  served=103  pending=17
  served=113  pending=7
All the producers have finished
INFO: Processor closed: 120 served, 0 rejected

package com.nexussoftware.bibliotech.domain;

public record Reservation(String id, String isbn) {

    /**
     * POISON PILL: a sentinel instance meaning
     * "there will be no more work, finish".
     *
     * It is compared with == (identity), not with equals: it is a
     * unique, unrepeatable object, and no real data can match it.
     */
    public static final Reservation END = new Reservation("__END__", "__END__");
}

for (int i = 0; i < CONSUMERS; i++) {
    queue.put(Reservation.END);
}

// Alternative: a single pill each consumer re-injects.
if (r == Reservation.END) {
    queue.put(Reservation.END);   // pass it to the next one
    return;
}

import java.util.concurrent.ConcurrentSkipListMap;
import java.util.NavigableMap;

// Loans sorted by due date (in milliseconds).
NavigableMap<Long, Loan> byDueDate = new ConcurrentSkipListMap<>();

byDueDate.put(dueMs, loan);

// All those overdue before now, in order, blocking nobody.
NavigableMap<Long, Loan> overdue =
        byDueDate.headMap(System.currentTimeMillis(), true);

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicCounter {

    private final AtomicInteger value = new AtomicInteger(0);

    public void increment() {
        value.incrementAndGet();     // ATOMIC, lock-free
    }

    public int value() { return value.get(); }

    public static void main(String[] args) throws InterruptedException {
        final int ROUNDS = 1_000_000;
        AtomicCounter c = new AtomicCounter();

        Thread t1 = new Thread(() -> { for (int i = 0; i < ROUNDS; i++) c.increment(); });
        Thread t2 = new Thread(() -> { for (int i = 0; i < ROUNDS; i++) c.increment(); });
        t1.start(); t2.start(); t1.join(); t2.join();

        System.out.println("Expected: " + ROUNDS * 2 + ", actual: " + c.value());
    }
}

Expected: 2000000, actual: 2000000

AtomicInteger a = new AtomicInteger(10);

// "If it is 10, set it to 11". Returns true if it did.
boolean success = a.compareAndSet(10, 11);

/**
 * What incrementAndGet() does inside, simplified.
 * It is the CAS LOOP pattern, the basis of all lock-free programming.
 */
public int incrementAndGet() {
    while (true) {
        int current = get();                     // 1. read
        int next = current + 1;                  // 2. compute
        if (compareAndSet(current, next)) {      // 3. try to write
            return next;                         //    success: leave
        }
        // Failure: another thread changed the value between 1 and 3.
        // Nothing has been lost: read again and retry.
    }
}

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.atomic.*;

/**
 * BiblioTech statistics with atomic counters.
 * Without a single lock, and with reads that block nothing.
 */
public class BiblioTechStatistics {

    private final AtomicLong totalLoans   = new AtomicLong();
    private final AtomicLong totalReturns = new AtomicLong();
    private final AtomicLong finesCollectedCents = new AtomicLong();
    private final AtomicInteger activeLoans = new AtomicInteger();
    private final AtomicBoolean maintenanceMode = new AtomicBoolean(false);
    private final AtomicLong maxConcurrent = new AtomicLong();

    public void registerLoan() {
        totalLoans.incrementAndGet();
        int active = activeLoans.incrementAndGet();

        // ALL-TIME MAXIMUM with accumulateAndGet: it combines the current
        // value with 'active' using Math::max, atomically.
        // Writing it with get()+set() would be a classic race.
        maxConcurrent.accumulateAndGet(active, Math::max);
    }

    public void registerReturn(long fineCents) {
        totalReturns.incrementAndGet();
        activeLoans.decrementAndGet();
        if (fineCents > 0) {
            finesCollectedCents.addAndGet(fineCents);
        }
    }

    /**
     * Applies a 10% discount to the total collected.
     * updateAndGet applies the function ATOMICALLY, with a CAS loop
     * underneath: if another thread modifies the value between the read and the
     * write, the function is RE-EXECUTED with the new value.
     *
     * IMPORTANT: the function must be PURE and fast, because it can
     * run several times. No side effects.
     */
    public long applyDiscount() {
        return finesCollectedCents.updateAndGet(v -> (long) (v * 0.9));
    }

    /**
     * Enter maintenance ONLY IF we were not already in it.
     * compareAndSet guarantees that, even if ten threads ask at once,
     * exactly ONE receives true and runs the preparation.
     * It is the "only once" idiom without locks.
     */
    public boolean enterMaintenance() {
        if (maintenanceMode.compareAndSet(false, true)) {
            prepareMaintenance();          // only one thread gets here
            return true;
        }
        return false;                      // somebody beat us to it
    }

    public String summary() {
        return String.format(
                "loans=%d returns=%d active=%d max=%d fines=%.2f EUR",
                totalLoans.get(), totalReturns.get(),
                activeLoans.get(), maxConcurrent.get(),
                finesCollectedCents.get() / 100.0);
    }

    private void prepareMaintenance() { /* ... */ }
}

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.atomic.AtomicReference;

public class BiblioTechState {

    /** The complete state, immutable (a record from 04-07). */
    public record State(long loans, long returns,
                        long fineCents, boolean maintenance) {

        State withLoan() {
            return new State(loans + 1, returns, fineCents, maintenance);
        }
        State withReturn(long fine) {
            return new State(loans, returns + 1,
                    fineCents + fine, maintenance);
        }
    }

    private final AtomicReference<State> state =
            new AtomicReference<>(new State(0, 0, 0, false));

    /**
     * A completely lock-free read that is ALWAYS CONSISTENT: the four
     * fields come from the same instant, because they are a single
     * immutable object. That is the advantage over four separate counters.
     */
    public State snapshot() {
        return state.get();
    }

    /** Atomic update of all four fields at once. */
    public void registerLoan() {
        state.updateAndGet(State::withLoan);
    }

    public void registerReturn(long fineCents) {
        state.updateAndGet(s -> s.withReturn(fineCents));
    }
}

Thread 1: reads A ............................. CAS(A -> C): SUCCESS
Thread 2:      reads A, CAS(A->B), CAS(B->A)
                                                 ^
        Thread 1 has no idea two changes happened.
        If its decision depended on NOTHING having happened, it is a bug.

import java.util.concurrent.atomic.AtomicStampedReference;

// Every modification increments the STAMP, even if the value goes back to
// being the same. The CAS checks value AND stamp, so A-B-A is detected.
AtomicStampedReference<Node> head =
        new AtomicStampedReference<>(initialNode, 0);

int[] currentStamp = new int[1];
Node current = head.get(currentStamp);

head.compareAndSet(current, newNode,
                   currentStamp[0], currentStamp[0] + 1);   // stamp + 1

// A simpler variant with a single boolean bit:
// AtomicMarkableReference<Node>

import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.*;

public class CounterComparison {

    static long measureAtomicLong(int threads, int ops) throws InterruptedException {
        AtomicLong c = new AtomicLong();
        return measure(threads, ops, c::incrementAndGet, c::get);
    }

    static long measureLongAdder(int threads, int ops) throws InterruptedException {
        LongAdder c = new LongAdder();
        return measure(threads, ops, c::increment, c::sum);
    }

    static long measure(int threads, int ops, Runnable increment,
                        java.util.function.LongSupplier read) throws InterruptedException {
        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch finishLine = new CountDownLatch(threads);
        for (int h = 0; h < threads; h++) {
            new Thread(() -> {
                try {
                    startGate.await();
                    for (int i = 0; i < ops; i++) increment.run();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally { finishLine.countDown(); }
            }, "counter-" + h).start();
        }
        long start = System.nanoTime();
        startGate.countDown();
        finishLine.await();
        long ns = System.nanoTime() - start;
        // Verification: both must give the exact result.
        if (read.getAsLong() != (long) threads * ops) {
            throw new AssertionError("incorrect result");
        }
        return ns / 1_000_000;
    }

    public static void main(String[] args) throws InterruptedException {
        final int OPS = 2_000_000;
        System.out.printf("%-8s | %14s | %14s | %8s%n",
                "Threads", "AtomicLong ms", "LongAdder ms", "Gain");
        System.out.println("---------|----------------|----------------|---------");
        for (int threads : new int[] { 1, 2, 4, 8, 16 }) {
            long a = measureAtomicLong(threads, OPS / threads);
            long l = measureLongAdder(threads, OPS / threads);
            System.out.printf("%-8d | %14d | %14d | %7.1fx%n",
                    threads, a, l, (double) a / Math.max(l, 1));
        }
    }
}

Threads  |  AtomicLong ms |   LongAdder ms |     Gain
---------|----------------|----------------|---------
1        |             12 |             18 |     0.7x
2        |             41 |             21 |     2.0x
4        |             96 |             19 |     5.1x
8        |            213 |             22 |     9.7x
16       |            487 |             26 |    18.7x

package com.nexussoftware.bibliotech.service;

import com.nexussoftware.bibliotech.domain.Material;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

/**
 * BiblioTech's concurrent catalogue.
 *
 * Compared with the ReadWriteLock version of 08-04:
 *  - There is no explicit lock to release in a finally.
 *  - Reads block NOTHING, not even other reads.
 *  - Writes on different buckets happen in parallel.
 *  - Compound operations are solved with the map's own atomic
 *    methods, not by hand-encapsulating critical sections.
 *
 * HONEST LIMIT: an invariant spanning SEVERAL structures
 * (like LoanRegistry's between byId and byEmployee) still
 * needs a lock. Concurrent collections guarantee the atomicity
 * of ONE operation on ONE collection, not of two.
 */
public class ConcurrentCatalog {

    /** Main index by ISBN. Lock-free reads. */
    private final ConcurrentMap<String, Material> byIsbn = new ConcurrentHashMap<>();

    /** Materials grouped by type. The value is a concurrent list. */
    private final ConcurrentMap<MaterialType, List<Material>> byType =
            new ConcurrentHashMap<>();

    /** Event listeners: many reads, almost no writes. */
    private final List<CatalogListener> listeners = new CopyOnWriteArrayList<>();

    // --- Statistics: LongAdder because of their high write frequency ---
    private final LongAdder lookups = new LongAdder();
    private final LongAdder hits    = new LongAdder();
    private final LongAdder insertions = new LongAdder();
    private final AtomicLong lastModifiedMs = new AtomicLong();

    // ---------- WRITING ----------

    /**
     * Adds a material if its ISBN was absent.
     * putIfAbsent is ATOMIC: even if ten threads add the same ISBN
     * at once, exactly one receives true. It is check-then-act
     * solved without locks.
     */
    public boolean add(Material m) {
        if (byIsbn.putIfAbsent(m.isbn(), m) != null) {
            return false;                       // it already existed
        }
        // computeIfAbsent creates the list only if absent, atomically.
        // CopyOnWriteArrayList because these lists are walked far
        // more often than they are modified.
        byType.computeIfAbsent(m.type(), t -> new CopyOnWriteArrayList<>()).add(m);

        insertions.increment();
        lastModifiedMs.set(System.currentTimeMillis());
        notifyAdded(m);
        return true;
    }

    public boolean remove(String isbn) {
        Material m = byIsbn.remove(isbn);
        if (m == null) return false;

        List<Material> list = byType.get(m.type());
        if (list != null) list.remove(m);

        lastModifiedMs.set(System.currentTimeMillis());
        return true;
    }

    // ---------- READING ----------

    /** Lock-free. With 100 threads querying, none waits for another. */
    public Material findByIsbn(String isbn) {
        lookups.increment();
        Material m = byIsbn.get(isbn);
        if (m != null) hits.increment();
        return m;
    }

    /**
     * Returns the list for a type. Since it is a CopyOnWriteArrayList,
     * the caller can iterate it in complete safety even if another thread
     * modifies it: its iterator works over an immutable snapshot.
     * There is no need to copy defensively, unlike in 08-04.
     */
    public List<Material> byType(MaterialType type) {
        return byType.getOrDefault(type, List.of());
    }

    /**
     * A complete walk of the catalogue.
     * The iterator is WEAKLY CONSISTENT: it throws no
     * ConcurrentModificationException and does not block the writers.
     */
    public void forEach(java.util.function.Consumer<Material> action) {
        for (Material m : byIsbn.values()) {
            action.accept(m);
        }
    }

    /** CAREFUL: approximate (section 7). Fine for metrics, not for control. */
    public int approximateSize() {
        return byIsbn.size();
    }

    // ---------- LISTENERS ----------

    public void registerListener(CatalogListener l) { listeners.add(l); }

    private void notifyAdded(Material m) {
        // No lock during the notification: rule 3 of 08-04
        // ("never call foreign code holding a lock") is honoured
        // automatically thanks to CopyOnWriteArrayList.
        for (CatalogListener l : listeners) {
            try {
                l.materialAdded(m);
            } catch (RuntimeException e) {
                LOG.log(Level.WARNING, "Listener failed", e);
            }
        }
    }

    // ---------- METRICS ----------

    public String metrics() {
        long c = lookups.sum();
        long h = hits.sum();
        return String.format("lookups=%d hits=%d rate=%.1f%% insertions=%d materials~%d",
                c, h, c == 0 ? 0.0 : 100.0 * h / c, insertions.sum(), byIsbn.size());
    }
}

public class ConcurrentCatalogTest {

    public static void main(String[] args) throws InterruptedException {

        final int THREADS = 16;
        final int OPS = 100_000;

        ConcurrentCatalog catalog = new ConcurrentCatalog();
        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch finishLine = new CountDownLatch(THREADS);

        for (int h = 0; h < THREADS; h++) {
            final int id = h;
            new Thread(() -> {
                try {
                    startGate.await();
                    ThreadLocalRandom random = ThreadLocalRandom.current();
                    for (int i = 0; i < OPS; i++) {
                        if (random.nextInt(100) < 90) {
                            catalog.findByIsbn("978-" + random.nextInt(1000));
                        } else {
                            catalog.add(new Book(
                                    "978-" + random.nextInt(1000), "Title", "Author"));
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally { finishLine.countDown(); }
            }, "catalog-" + id).start();
        }

        long start = System.nanoTime();
        startGate.countDown();
        finishLine.await();
        long ms = (System.nanoTime() - start) / 1_000_000;

        System.out.println("Operations  : " + (THREADS * OPS));
        System.out.println("Time        : " + ms + " ms");
        System.out.println("Throughput  : " + (THREADS * OPS / Math.max(ms, 1)) + " ops/ms");
        System.out.println(catalog.metrics());
    }
}

Operations  : 1600000
Time        : 312 ms
Throughput  : 5128 ops/ms
Metrics     : lookups=1439871 hits=1438204 rate=99.9% insertions=1000 materials~1000

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

public class ThreeGenerations {

    static final int THREADS = 12;
    static final int OPS = 100_000;
    static final int KEYS = 5_000;

    record Result(String name, long ms, int entries, String issue) { }

    static Result measure(String name, Map<Integer, String> map)
            throws InterruptedException {

        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch finishLine = new CountDownLatch(THREADS);
        // volatile is not enough to accumulate text from several threads:
        // we use a concurrent queue to collect the issues.
        Queue<String> issues = new ConcurrentLinkedQueue<>();

        for (int h = 0; h < THREADS; h++) {
            Thread t = new Thread(() -> {
                try {
                    startGate.await();
                    ThreadLocalRandom random = ThreadLocalRandom.current();
                    for (int i = 0; i < OPS; i++) {
                        int key = random.nextInt(KEYS);
                        if (random.nextInt(100) < 80) {
                            map.get(key);
                        } else {
                            map.put(key, "v" + i);
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (RuntimeException e) {
                    issues.add(e.getClass().getSimpleName());
                } finally {
                    finishLine.countDown();
                }
            }, name + "-" + h);
            t.setDaemon(true);      // daemon: if it goes into an infinite loop,
            t.start();              // it will not stop the JVM terminating
        }

        long start = System.nanoTime();
        startGate.countDown();

        // Deadline: the unprotected HashMap may NEVER finish.
        boolean completed = finishLine.await(20, TimeUnit.SECONDS);
        long ms = (System.nanoTime() - start) / 1_000_000;

        String issue = completed
                ? (issues.isEmpty() ? "-" : issues.peek())
                : "DID NOT FINISH (infinite loop or corruption)";

        int entries;
        try {
            entries = map.size();
        } catch (RuntimeException e) {
            entries = -1;
        }
        return new Result(name, ms, entries, issue);
    }

    public static void main(String[] args) throws InterruptedException {

        System.out.printf("%d threads x %d ops (80%% reads) over %d keys%n%n",
                THREADS, OPS, KEYS);

        List<Result> results = new ArrayList<>();
        results.add(measure("HashMap", new HashMap<>()));
        results.add(measure("synchronizedMap", Collections.synchronizedMap(new HashMap<>())));
        results.add(measure("ConcurrentHashMap", new ConcurrentHashMap<>()));

        System.out.printf("%-20s | %8s | %10s | %-40s%n",
                "Implementation", "ms", "entries", "issue");
        System.out.println("---------------------|----------|------------|"
                + "------------------------------------------");
        for (Result r : results) {
            System.out.printf("%-20s | %8d | %10d | %-40s%n",
                    r.name(), r.ms(), r.entries(), r.issue());
        }

        System.out.println();
        System.out.println("Expected: " + KEYS + " entries (every key touched)");
    }
}

12 threads x 100000 ops (80% reads) over 5000 keys

Implementation       |       ms |    entries | issue
---------------------|----------|------------|------------------------------------------
HashMap              |    20003 |       4211 | DID NOT FINISH (infinite loop or corruption)
synchronizedMap      |     1876 |       5000 | -
ConcurrentHashMap    |      147 |       5000 | -

Expected: 5000 entries (every key touched)

package com.nexussoftware.bibliotech.persistence;

import java.util.concurrent.*;
import java.util.concurrent.atomic.LongAdder;

public class ImportPipeline {

    private static final String POISON = "__END__";
    private static final int CAPACITY = 20;
    private static final int PRODUCERS = 2;
    private static final int CONSUMERS = 4;
    private static final int LINES_PER_PRODUCER = 150;

    // BOUNDED queue: if the consumers cannot keep up, the producers
    // block in put(). Backpressure.
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(CAPACITY);

    private final ConcurrentMap<String, String> materials = new ConcurrentHashMap<>();

    // LongAdder: very frequent writes, occasional reads.
    private final LongAdder linesRead = new LongAdder();
    private final LongAdder created   = new LongAdder();
    private final LongAdder discarded = new LongAdder();

    private volatile boolean running = true;

    // ---------- STAGE 1: PRODUCERS ----------

    private void produce(int producerId) {
        try {
            for (int i = 0; i < LINES_PER_PRODUCER; i++) {
                String line = "978-" + producerId + String.format("%05d", i)
                        + ";Title " + i + ";Author";
                queue.put(line);      // BLOCKS if the queue is full
                linesRead.increment();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // ---------- STAGE 2: CONSUMERS ----------

    private void consume() {
        String me = Thread.currentThread().getName();
        try {
            while (true) {
                String line = queue.take();      // BLOCKS if it is empty

                // Poison pill: comparison by IDENTITY.
                if (line == POISON) {
                    System.out.println("  [" + me + "] pill received, finishing");
                    return;
                }
                try {
                    TimeUnit.MILLISECONDS.sleep(30);      // "expensive" conversion
                    String[] fields = line.split(";");
                    if (fields.length < 3) {
                        throw new IllegalArgumentException("incomplete line");
                    }
                    materials.put(fields[0], fields[1]);
                    created.increment();
                } catch (IllegalArgumentException e) {
                    discarded.increment();                 // degrade (06-07)
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    // ---------- ORCHESTRATION ----------

    public void run() throws InterruptedException {

        ExecutorService producers = Executors.newFixedThreadPool(PRODUCERS,
                new ThreadNamer("pipeline-producer"));
        ExecutorService consumers = Executors.newFixedThreadPool(CONSUMERS,
                new ThreadNamer("pipeline-consumer"));

        for (int c = 0; c < CONSUMERS; c++) consumers.execute(this::consume);

        CountDownLatch productionFinished = new CountDownLatch(PRODUCERS);
        for (int p = 0; p < PRODUCERS; p++) {
            final int id = p;
            producers.execute(() -> {
                try { produce(id); }
                finally { productionFinished.countDown(); }    // ALWAYS
            });
        }

        // Monitor: shows that the queue fills up and slows the producers.
        Thread monitor = new Thread(() -> {
            try {
                while (running) {
                    System.out.printf("      [monitor] queue=%2d/%d  read=%d  created=%d%n",
                            queue.size(), CAPACITY, linesRead.sum(), created.sum());
                    TimeUnit.MILLISECONDS.sleep(300);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, "pipeline-monitor");
        monitor.setDaemon(true);
        monitor.start();

        // 1. Wait for the producers to finish.
        productionFinished.await();
        System.out.println("  Production finished; injecting pills");

        // 2. ONE pill PER CONSUMER, at the end of the queue:
        //    everything before it gets processed first.
        for (int c = 0; c < CONSUMERS; c++) queue.put(POISON);

        // 3. Orderly two-phase shutdown.
        producers.shutdown();
        consumers.shutdown();
        boolean ok = consumers.awaitTermination(30, TimeUnit.SECONDS);
        if (!ok) consumers.shutdownNow();

        running = false;
        monitor.interrupt();

        // 4. Verification.
        long total = LINES_PER_PRODUCER * PRODUCERS;
        System.out.println();
        System.out.println("=== RESULT ===");
        System.out.println("Lines produced    : " + linesRead.sum() + " (expected " + total + ")");
        System.out.println("Materials created : " + created.sum());
        System.out.println("Discarded         : " + discarded.sum());
        System.out.println("In the map        : " + materials.size());
        System.out.println("Queue at the end  : " + queue.size() + " (must be 0)");
        System.out.println("No losses         : "
                + (created.sum() + discarded.sum() == total));
    }

    static class ThreadNamer implements ThreadFactory {
        private final String prefix;
        private int n = 1;
        ThreadNamer(String prefix) { this.prefix = prefix; }
        @Override public synchronized Thread newThread(Runnable r) {
            return new Thread(r, prefix + "-" + n++);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new ImportPipeline().run();
    }
}

      [monitor] queue=20/20  read=42  created=22
      [monitor] queue=20/20  read=82  created=62
      [monitor] queue=20/20  read=122  created=102
      [monitor] queue=20/20  read=162  created=142
      ...
  Production finished; injecting pills
  [pipeline-consumer-2] pill received, finishing
  [pipeline-consumer-1] pill received, finishing
  [pipeline-consumer-4] pill received, finishing
  [pipeline-consumer-3] pill received, finishing

=== RESULT ===
Lines produced    : 300 (expected 300)
Materials created : 300
Discarded         : 0
In the map        : 300
Queue at the end  : 0 (must be 0)
No losses         : true

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

public class StatisticsComparison {

    interface Counter {
        void registerLoan();
        long total();
        String name();
    }

    /** (a) a long guarded with synchronized. */
    static class WithSynchronized implements Counter {
        private long loans = 0;
        public synchronized void registerLoan() { loans++; }
        public synchronized long total() { return loans; }
        public String name() { return "synchronized"; }
    }

    /** (b) AtomicLong: a CAS loop underneath. */
    static class WithAtomicLong implements Counter {
        private final AtomicLong loans = new AtomicLong();
        public void registerLoan() { loans.incrementAndGet(); }
        public long total() { return loans.get(); }
        public String name() { return "AtomicLong"; }
    }

    /** (c) LongAdder: separate cells, summed on reading. */
    static class WithLongAdder implements Counter {
        private final LongAdder loans = new LongAdder();
        public void registerLoan() { loans.increment(); }
        public long total() { return loans.sum(); }
        public String name() { return "LongAdder"; }
    }

    /** (d) AtomicReference over an immutable THREE-field record. */
    static class WithAtomicReference implements Counter {

        record State(long loans, long returns, long fines) {
            State withLoan() {
                return new State(loans + 1, returns, fines);
            }
        }

        private final AtomicReference<State> state =
                new AtomicReference<>(new State(0, 0, 0));

        public void registerLoan() {
            // updateAndGet retries if another thread got in first.
            // The function must be PURE: it can run several times.
            state.updateAndGet(State::withLoan);
        }

        public long total() { return state.get().loans(); }

        /** UNIQUE ADVANTAGE: a CONSISTENT snapshot of the three fields. */
        public State snapshot() { return state.get(); }

        public String name() { return "AtomicReference+record"; }
    }

    static long measure(Counter c, int threads, int perThread) throws InterruptedException {
        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch finishLine = new CountDownLatch(threads);

        for (int h = 0; h < threads; h++) {
            new Thread(() -> {
                try {
                    startGate.await();
                    for (int i = 0; i < perThread; i++) c.registerLoan();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally { finishLine.countDown(); }
            }, "stat-" + h).start();
        }

        long start = System.nanoTime();
        startGate.countDown();
        finishLine.await();
        long ms = (System.nanoTime() - start) / 1_000_000;

        long expected = (long) threads * perThread;
        if (c.total() != expected) {
            throw new AssertionError(c.name() + " INCORRECT: "
                    + c.total() + " != " + expected);
        }
        return ms;
    }

    static Counter newCounter(int type) {
        return switch (type) {
            case 0 -> new WithSynchronized();
            case 1 -> new WithAtomicLong();
            case 2 -> new WithLongAdder();
            default -> new WithAtomicReference();
        };
    }

    public static void main(String[] args) throws InterruptedException {

        final int TOTAL = 8_000_000;

        // Warm-up (JIT).
        for (int t = 0; t < 4; t++) measure(newCounter(t), 4, 50_000);

        for (int threads : new int[] { 1, 16 }) {
            System.out.printf("%n=== %d thread(s), %d increments in total ===%n",
                    threads, TOTAL);
            System.out.printf("%-26s | %8s | %14s%n", "Implementation", "ms", "inc/ms");
            System.out.println("---------------------------|----------|---------------");

            for (int t = 0; t < 4; t++) {
                Counter c = newCounter(t);
                long ms = measure(c, threads, TOTAL / threads);
                System.out.printf("%-26s | %8d | %14d%n",
                        c.name(), ms, TOTAL / Math.max(ms, 1));
            }
        }

        // Demonstration of AtomicReference's unique advantage.
        WithAtomicReference ar = new WithAtomicReference();
        measure(ar, 8, 100_000);
        WithAtomicReference.State snap = ar.snapshot();
        System.out.printf("%nCONSISTENT snapshot: loans=%d returns=%d fines=%d%n",
                snap.loans(), snap.returns(), snap.fines());
        System.out.println("The three fields come from the SAME instant, something");
        System.out.println("three independent counters cannot guarantee.");
    }
}

=== 1 thread(s), 8000000 increments in total ===
Implementation             |       ms |         inc/ms
---------------------------|----------|---------------
synchronized               |       61 |         131147
AtomicLong                 |       48 |         166666
LongAdder                  |       72 |         111111
AtomicReference+record     |      284 |          28169

=== 16 thread(s), 8000000 increments in total ===
Implementation             |       ms |         inc/ms
---------------------------|----------|---------------
synchronized               |     1842 |           4343
AtomicLong                 |      918 |           8714
LongAdder                  |       94 |          85106
AtomicReference+record     |     1531 |           5225

CONSISTENT snapshot: loans=800000 returns=0 fines=0
The three fields come from the SAME instant, something
three independent counters cannot guarantee.