// Java: Producer-Consumer with BlockingQueue (preferred approach)
import java.util.concurrent.*;

public class ProducerConsumer {
    private static final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);

    static class Producer implements Runnable {
        public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    queue.put(i);  // blocks if queue full
                    System.out.println("Produced: " + i);
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }
        }
    }

    static class Consumer implements Runnable {
        public void run() {
            while (true) {
                try {
                    int val = queue.take();  // blocks if queue empty
                    System.out.println("Consumed: " + val);
                    Thread.sleep(50);
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; }
            }
        }
    }

    public static void main(String[] args) {
        new Thread(new Producer()).start();
        new Thread(new Consumer()).start();
        new Thread(new Consumer()).start(); // multiple consumers
    }
}

synchronized(monitor) {
    while (queue.isEmpty()) monitor.wait();   // always use while not if (spurious wakeup)
    item = queue.poll();
    monitor.notifyAll();                      // wake up producers
}

// BAD: Race condition — counter++ is NOT atomic
// It's: read counter, increment, write counter — 3 steps!
private int counter = 0;
public void increment() { counter++; } // NOT THREAD SAFE

// FIX 1: synchronized
public synchronized void increment() { counter++; }

// FIX 2: AtomicInteger (lock-free, CAS-based)
private AtomicInteger counter = new AtomicInteger(0);
public void increment() { counter.incrementAndGet(); }

// Memory Visibility: volatile
private volatile boolean running = true;  // write from one thread visible to all
// Without volatile: thread may read stale cached value from register/L1 cache

// Thread Pool: ALWAYS use instead of creating raw threads
ExecutorService pool = Executors.newFixedThreadPool(
    Runtime.getRuntime().availableProcessors() * 2  // rule of thumb for I/O-bound
);

// For CPU-bound tasks: nCPU cores
// For I/O-bound tasks: nCPU * (1 + wait_time / compute_time), often 2x-10x cores

// Submit tasks
Future<String> future = pool.submit(() -> {
    Thread.sleep(1000);
    return "result";
});
String result = future.get(); // blocks; use get(timeout, unit) in prod

// Custom thread pool
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4,           // corePoolSize — always kept alive
    16,          // maximumPoolSize — max threads
    60L, TimeUnit.SECONDS,  // keepAliveTime for excess threads
    new LinkedBlockingQueue<>(1000),  // work queue (bounded!)
    new ThreadPoolExecutor.CallerRunsPolicy() // rejection policy
);
// Rejection policies: AbortPolicy(default), CallerRunsPolicy, DiscardPolicy, DiscardOldestPolicy

// Always shutdown properly!
pool.shutdown();
pool.awaitTermination(5, TimeUnit.SECONDS);

// ReentrantLock — more flexible than synchronized
ReentrantLock lock = new ReentrantLock(true); // fair=true → FIFO order (prevents starvation)

lock.lock();
try {
    // critical section
} finally {
    lock.unlock(); // ALWAYS in finally!
}

// tryLock with timeout (avoids deadlock)
if (lock.tryLock(500, TimeUnit.MILLISECONDS)) {
    try { /* ... */ } finally { lock.unlock(); }
} else {
    // couldn't acquire — handle gracefully
}

// ReadWriteLock: many readers OR one writer
ReadWriteLock rwLock = new ReentrantReadWriteLock();
// Multiple readers can hold read lock simultaneously
rwLock.readLock().lock();
try { return map.get(key); } finally { rwLock.readLock().unlock(); }

// Exclusive write lock
rwLock.writeLock().lock();
try { map.put(key, val); } finally { rwLock.writeLock().unlock(); }

// StampedLock (Java 8+): optimistic read without locking
StampedLock sl = new StampedLock();
long stamp = sl.tryOptimisticRead();
int x = this.x; int y = this.y;           // read without lock
if (!sl.validate(stamp)) {                 // check if write happened
    stamp = sl.readLock();
    try { x = this.x; y = this.y; } finally { sl.unlockRead(stamp); }
}
// ~3x faster than ReadWriteLock for read-heavy workloads

// Atomic types — based on CPU CAS (Compare-And-Swap) instruction
AtomicInteger counter = new AtomicInteger(0);
counter.incrementAndGet();                    // atomic i++
counter.compareAndSet(expected, newVal);      // CAS — update only if current == expected
counter.getAndUpdate(x -> x * 2);

AtomicLong, AtomicBoolean, AtomicReference<T>

// LongAdder — better than AtomicLong under high contention
// Maintains per-CPU-stripe counters, sums on read
LongAdder adder = new LongAdder();
adder.increment();  // contention spread across stripes
long total = adder.sum(); // slightly less precise but much faster

// ConcurrentHashMap — segment-level locking (Java 8: CAS + synchronized per bucket)
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
map.putIfAbsent(key, 0);
map.compute(key, (k, v) -> v == null ? 1 : v + 1); // atomic update

// CopyOnWriteArrayList — writes create new copy; reads never block
// Good for: event listener lists (rare writes, many reads)
CopyOnWriteArrayList<Listener> listeners = new CopyOnWriteArrayList<>();

// CompletableFuture — non-blocking async composition
CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(
    () -> userService.findById(userId),  // runs on ForkJoinPool.commonPool()
    customExecutor                        // or specify your executor
);

// Chain operations (no blocking!)
CompletableFuture<String> result = userFuture
    .thenApply(user -> user.getEmail())           // transform
    .thenCompose(email -> sendEmail(email))        // chain another async op
    .thenAccept(res -> log.info("Sent: " + res))  // side effect, returns Void
    .exceptionally(ex -> { log.error(ex); return null; }); // error handling

// Run multiple in parallel, wait for all
CompletableFuture<User>    cf1 = CompletableFuture.supplyAsync(() -> getUser(id));
CompletableFuture<Profile> cf2 = CompletableFuture.supplyAsync(() -> getProfile(id));

CompletableFuture.allOf(cf1, cf2).thenRun(() -> {
    User user = cf1.join();
    Profile profile = cf2.join();
    // both ready here
});

// First one to complete wins
CompletableFuture.anyOf(cf1, cf2).thenAccept(result -> { /* handle */ });

// With timeout (Java 9+)
cf1.orTimeout(2, TimeUnit.SECONDS)
   .exceptionally(ex -> DEFAULT_USER);

// Create virtual threads — Java 21
Thread.ofVirtual().start(() -> {
    // blocking I/O here doesn't waste OS thread!
    String result = httpClient.get("https://api.example.com/data");
});

// Virtual thread executor — best for I/O-bound server workloads
ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor();
// Submit 100,000 tasks — no problem, each gets own virtual thread
for (int i = 0; i < 100_000; i++) {
    executor.submit(() -> processRequest()); // cheap!
}

// Spring Boot 3.2+: enable virtual threads
// application.properties: spring.threads.virtual.enabled=true