Multithreading in Java

public class DownloadThread extends Thread {
    private final String url;

    public DownloadThread(String url) {
        this.url = url;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " downloading: " + url);
    }
}

DownloadThread thread = new DownloadThread("https://example.com/file.zip");
thread.start();

public class DownloadTask implements Runnable {
    private final String url;

    public DownloadTask(String url) {
        this.url = url;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + " downloading: " + url);
    }
}

Thread thread = new Thread(new DownloadTask("https://example.com/file.zip"));
thread.start();

// Lambda style (Java 8+)
Thread thread = new Thread(() -> System.out.println("Running in: " + Thread.currentThread().getName()));
thread.start();

public class PriceCalculator implements Callable<Double> {
    private final String ticker;

    public PriceCalculator(String ticker) {
        this.ticker = ticker;
    }

    @Override
    public Double call() throws Exception {
        Thread.sleep(1000);
        return Math.random() * 1000;
    }
}

ExecutorService executor = Executors.newFixedThreadPool(4);
Future<Double> future = executor.submit(new PriceCalculator("GOOG"));
Double price = future.get();  // blocks until result is available
executor.shutdown();

Thread t = new Thread(() -> System.out.println("Running in: " + Thread.currentThread().getName()));

t.start();  // Creates a NEW thread, executes run() in that thread
t.run();    // Does NOT create a new thread — runs in caller's thread

try {
    Thread.sleep(2000);
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();  // Restore interrupt flag
}

Thread worker = new Thread(() -> computeResult());
worker.start();
worker.join();  // Current thread waits until worker finishes

Thread worker = new Thread(() -> {
    while (!Thread.currentThread().isInterrupted()) {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            break;
        }
    }
});
worker.start();
worker.interrupt();  // Request cancellation — sets flag

Thread daemon = new Thread(() -> {
    while (true) {
        cleanupExpiredSessions();
        Thread.sleep(60_000);
    }
});
daemon.setDaemon(true);  // Must be set BEFORE start()
daemon.start();

// Thread 1                       // Thread 2
flag = true;                      while (!flag) { }  // May loop forever!
data = 42;                        print(data);       // May print 0!

// Source code             // After reordering (legal for single thread)
x = 1;                    y = 2;  // moved up!
y = 2;                    x = 1;

private volatile boolean shutdownRequested = false;

// Thread 1 (writer)
shutdownRequested = true;  // StoreStore + StoreLoad barrier after write

// Thread 2 (reader)
while (!shutdownRequested) {  // LoadLoad + LoadStore barrier before read
    doWork();
}
// Guaranteed to see the write from Thread 1

private volatile int counter = 0;

// Thread 1                    // Thread 2
counter++;                     counter++;

// counter++ is: read → increment → write (3 operations)
// volatile makes each individual read/write visible, but the compound
// operation is NOT atomic. Final counter could be 1, not 2.

public class Singleton {
    private static volatile Singleton instance;  // volatile is critical here

    public static Singleton getInstance() {
        if (instance == null) {                   // first check (no lock)
            synchronized (Singleton.class) {
                if (instance == null) {           // second check (with lock)
                    instance = new Singleton();   // without volatile, partially
                }                                 // constructed object may be visible
            }
        }
        return instance;
    }
}

public class BankAccount {
    private double balance;
    private final Object lock = new Object();

    public void deposit(double amount) {
        synchronized (lock) {
            balance += amount;
        }
        notifyObservers();  // outside lock — better throughput
    }
}

public class BoundedBuffer<T> {
    private final Queue<T> queue = new LinkedList<>();
    private final int capacity;

    public BoundedBuffer(int capacity) {
        this.capacity = capacity;
    }

    public synchronized void produce(T item) throws InterruptedException {
        while (queue.size() == capacity) {
            wait();  // Buffer full — release lock and wait
        }
        queue.add(item);
        notifyAll();  // Wake up consumers
    }

    public synchronized T consume() throws InterruptedException {
        while (queue.isEmpty()) {
            wait();  // Buffer empty — release lock and wait
        }
        T item = queue.poll();
        notifyAll();  // Wake up producers
        return item;
    }
}

private final ReentrantLock lock = new ReentrantLock();

public void transfer(Account from, Account to, double amount) {
    lock.lock();
    try {
        from.debit(amount);
        to.credit(amount);
    } finally {
        lock.unlock();  // ALWAYS in finally — prevents lock leak on exception
    }
}

public boolean transferWithTimeout(Account from, Account to, double amount) 
        throws InterruptedException {
    if (from.lock.tryLock(1, TimeUnit.SECONDS)) {
        try {
            if (to.lock.tryLock(1, TimeUnit.SECONDS)) {
                try {
                    from.debit(amount);
                    to.credit(amount);
                    return true;
                } finally {
                    to.lock.unlock();
                }
            }
        } finally {
            from.lock.unlock();
        }
    }
    return false;  // Could not acquire both locks — caller retries
}

private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();

public void produce(T item) throws InterruptedException {
    lock.lock();
    try {
        while (count == capacity) notFull.await();
        enqueue(item);
        notEmpty.signal();  // Only wake consumers, not other producers
    } finally {
        lock.unlock();
    }
}

public T consume() throws InterruptedException {
    lock.lock();
    try {
        while (count == 0) notEmpty.await();
        T item = dequeue();
        notFull.signal();  // Only wake producers
        return item;
    } finally {
        lock.unlock();
    }
}

private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
private final Map<String, Object> cache = new HashMap<>();

public Object get(String key) {
    rwLock.readLock().lock();  // Multiple readers can hold simultaneously
    try {
        return cache.get(key);
    } finally {
        rwLock.readLock().unlock();
    }
}

public void put(String key, Object value) {
    rwLock.writeLock().lock();  // Exclusive — blocks all readers and writers
    try {
        cache.put(key, value);
    } finally {
        rwLock.writeLock().unlock();
    }
}

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

public double distanceFromOrigin() {
    long stamp = sl.tryOptimisticRead();  // No lock acquired — just a stamp
    double currentX = x, currentY = y;
    if (!sl.validate(stamp)) {  // Check if a write occurred meanwhile
        stamp = sl.readLock();  // Fall back to pessimistic read
        try {
            currentX = x;
            currentY = y;
        } finally {
            sl.unlockRead(stamp);
        }
    }
    return Math.sqrt(currentX * currentX + currentY * currentY);
}

// What AtomicInteger.incrementAndGet() does internally:
public final int incrementAndGet() {
    int prev, next;
    do {
        prev = get();              // volatile read
        next = prev + 1;
    } while (!compareAndSet(prev, next));  // CAS retry loop
    return next;
}

// AtomicLong: all threads CAS on single variable — high contention, many retries
private final AtomicLong counter = new AtomicLong();

// LongAdder: internally striped across cells — threads update different cells
// sum() aggregates all cells (slightly higher read cost, much lower write contention)
private final LongAdder counter = new LongAdder();
counter.increment();      // write — distributed across cells
counter.sum();            // read — aggregates (eventual consistency for reads)

AtomicStampedReference<Node> head = new AtomicStampedReference<>(node, 0);

int[] stampHolder = new int[1];
Node current = head.get(stampHolder);
int stamp = stampHolder[0];

head.compareAndSet(current, newNode, stamp, stamp + 1);

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    4,                              // corePoolSize
    16,                             // maximumPoolSize
    60L, TimeUnit.SECONDS,          // keepAliveTime for idle threads > core
    new ArrayBlockingQueue<>(1000), // bounded queue — backpressure!
    new ThreadFactory() {           // custom thread naming
        private final AtomicInteger count = new AtomicInteger();
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "order-processor-" + count.incrementAndGet());
            t.setDaemon(false);
            return t;
        }
    },
    new ThreadPoolExecutor.CallerRunsPolicy()  // rejection policy
);

public class MergeSortTask extends RecursiveAction {
    private final int[] array;
    private final int left, right;
    private static final int THRESHOLD = 1024;

    @Override
    protected void compute() {
        if (right - left < THRESHOLD) {
            Arrays.sort(array, left, right);
            return;
        }
        int mid = (left + right) / 2;
        invokeAll(
            new MergeSortTask(array, left, mid),
            new MergeSortTask(array, mid, right)
        );
        merge(array, left, mid, right);
    }
}

ForkJoinPool pool = new ForkJoinPool();
pool.invoke(new MergeSortTask(array, 0, array.length));

int serviceCount = 5;
CountDownLatch latch = new CountDownLatch(serviceCount);

for (int i = 0; i < serviceCount; i++) {
    executor.submit(() -> {
        try {
            initializeService();
        } finally {
            latch.countDown();  // Decrement count
        }
    });
}

latch.await(30, TimeUnit.SECONDS);  // Block until count reaches 0
// All services initialized — start accepting requests

int threadCount = 4;
CyclicBarrier barrier = new CyclicBarrier(threadCount, () -> {
    mergePartialResults();  // Runs when all threads arrive
});

for (int i = 0; i < threadCount; i++) {
    final int partition = i;
    executor.submit(() -> {
        while (hasMoreData()) {
            processPartition(partition);
            barrier.await();  // Wait for all threads to finish this phase
            // barrier resets — ready for next phase
        }
    });
}

// Connection pool with max 10 concurrent connections
Semaphore semaphore = new Semaphore(10, true);  // fair = true

public Connection getConnection() throws InterruptedException {
    semaphore.acquire();  // blocks if 10 connections already in use
    try {
        return pool.borrowConnection();
    } catch (Exception e) {
        semaphore.release();
        throw e;
    }
}

public void releaseConnection(Connection conn) {
    pool.returnConnection(conn);
    semaphore.release();  // permit becomes available
}

Phaser phaser = new Phaser(1);  // register self

for (int i = 0; i < workerCount; i++) {
    phaser.register();  // dynamic registration
    executor.submit(() -> {
        // Phase 0: load data
        loadData();
        phaser.arriveAndAwaitAdvance();

        // Phase 1: process
        process();
        phaser.arriveAndAwaitAdvance();

        // Phase 2: write results
        writeResults();
        phaser.arriveAndDeregister();  // done — leave the phaser
    });
}

phaser.arriveAndDeregister();  // deregister self

Exchanger<List<Order>> exchanger = new Exchanger<>();

// Producer thread
List<Order> buffer = new ArrayList<>();
while (running) {
    buffer.add(fetchOrder());
    if (buffer.size() >= BATCH_SIZE) {
        buffer = exchanger.exchange(buffer);  // swap full buffer for empty one
        buffer.clear();
    }
}

// Consumer thread
List<Order> buffer = new ArrayList<>();
while (running) {
    buffer = exchanger.exchange(buffer);  // swap empty buffer for full one
    processBatch(buffer);
}

// Excellent for: listeners, event handlers, configuration (read >> write)
private final CopyOnWriteArrayList<EventListener> listeners = new CopyOnWriteArrayList<>();

public void addListener(EventListener l) { listeners.add(l); }  // copies entire array

public void fireEvent(Event e) {
    for (EventListener l : listeners) {  // no lock needed — iterates snapshot
        l.onEvent(e);
    }
}

// Run async, no return value
CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> sendEmail());

// Run async, return value
CompletableFuture<User> cf = CompletableFuture.supplyAsync(() -> fetchUser(id));

// With custom executor
CompletableFuture<User> cf = CompletableFuture.supplyAsync(
    () -> fetchUser(id), ioExecutor
);

CompletableFuture<OrderConfirmation> pipeline = 
    CompletableFuture.supplyAsync(() -> validateOrder(order))
        .thenApply(valid -> enrichOrder(valid))          // sync transform
        .thenCompose(enriched -> chargePayment(enriched)) // async flatMap
        .thenApply(charged -> createConfirmation(charged))
        .exceptionally(ex -> handleFailure(ex));

// Wait for ALL to complete
CompletableFuture<Void> allOf = CompletableFuture.allOf(
    fetchUser(id),
    fetchOrders(id),
    fetchRecommendations(id)
);

// Wait for FIRST to complete (racing)
CompletableFuture<Object> anyOf = CompletableFuture.anyOf(
    callServiceA(),
    callServiceB()  // hedged request
);

CompletableFuture<Price> price = CompletableFuture
    .supplyAsync(() -> callPricingService(item))
    .orTimeout(2, TimeUnit.SECONDS)                     // Java 9
    .exceptionally(ex -> getCachedPrice(item));          // fallback

public UserProfile buildProfile(String userId) {
    var userFuture = CompletableFuture.supplyAsync(() -> fetchUser(userId), ioPool);
    var ordersFuture = CompletableFuture.supplyAsync(() -> fetchOrders(userId), ioPool);
    var prefsFuture = CompletableFuture.supplyAsync(() -> fetchPrefs(userId), ioPool);

    return userFuture.thenCombine(ordersFuture, (user, orders) -> 
        new PartialProfile(user, orders)
    ).thenCombine(prefsFuture, (partial, prefs) -> 
        new UserProfile(partial, prefs)
    ).join();  // block for final result
}

// Single virtual thread
Thread.ofVirtual().name("worker-1").start(() -> {
    String result = blockingHttpCall();  // OK to block!
    processResult(result);
});

// Virtual thread per task executor (replaces CachedThreadPool)
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<String>> futures = IntStream.range(0, 100_000)
        .mapToObj(i -> executor.submit(() -> fetchData(i)))
        .toList();
    for (Future<String> f : futures) {
        process(f.get());
    }
}

// BAD: synchronized pins the virtual thread to the carrier
synchronized (lock) {
    connection.query(sql);  // blocks while pinned — wastes carrier
}

// GOOD: ReentrantLock does NOT pin
private final ReentrantLock lock = new ReentrantLock();
lock.lock();
try {
    connection.query(sql);  // virtual thread unmounts while waiting for I/O
} finally {
    lock.unlock();
}

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Subtask<User> userTask = scope.fork(() -> fetchUser(userId));
    Subtask<List<Order>> ordersTask = scope.fork(() -> fetchOrders(userId));

    scope.join();           // Wait for all
    scope.throwIfFailed();  // Propagate exceptions

    return new UserProfile(userTask.get(), ordersTask.get());
}
// If any subtask fails, all others are cancelled automatically

private static final ScopedValue<RequestContext> CONTEXT = ScopedValue.newInstance();

ScopedValue.where(CONTEXT, new RequestContext(requestId))
    .run(() -> {
        handleRequest();  // CONTEXT.get() available in this scope and child tasks
    });

private static final ThreadLocal<SimpleDateFormat> dateFormat = 
    ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd"));

public String formatDate(Date date) {
    return dateFormat.get().format(date);  // each thread gets its own instance
}

// LEAK: ThreadLocal not removed in thread pool
executorService.submit(() -> {
    CONTEXT.set(expensiveObject);  // stored in worker thread's map
    doWork();
    // Missing CONTEXT.remove() — object lives until thread dies
    // In a thread pool, thread never dies → permanent leak!
});

// FIX: always remove in finally
executorService.submit(() -> {
    CONTEXT.set(expensiveObject);
    try {
        doWork();
    } finally {
        CONTEXT.remove();  // critical in thread pools
    }
});

// Child threads inherit parent's ThreadLocal value (copy at creation time)
private static final InheritableThreadLocal<String> traceId = 
    new InheritableThreadLocal<>();

traceId.set("trace-12345");
new Thread(() -> {
    System.out.println(traceId.get());  // "trace-12345" — inherited
}).start();

// BUG: counter++ is NOT atomic (read → increment → write)
private int counter = 0;
public void increment() { counter++; }

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

// FIX 2: AtomicInteger (better performance)
private final AtomicInteger counter = new AtomicInteger();
public void increment() { counter.incrementAndGet(); }

// Thread 1: lock A → tries lock B
// Thread 2: lock B → tries lock A
// DEADLOCK!

// Prevention: always acquire locks in consistent global order

// BAD: counters[0] and counters[1] on same cache line (64 bytes)
// Writing to counters[0] invalidates counters[1] in other CPU's cache
private long[] counters = new long[NUM_THREADS];

// Thread i increments counters[i] — but all threads slow each other down
// due to cache line bouncing between CPUs

// FIX: pad to separate cache lines (@Contended in JMH)
@jdk.internal.vm.annotation.Contended
private volatile long counter;  // padded to own cache line

public class TokenBucketRateLimiter {
    private final int maxTokens;
    private final int refillRate;  // tokens per second
    private double availableTokens;
    private long lastRefillTime;
    private final ReentrantLock lock = new ReentrantLock();

    public boolean tryAcquire() {
        lock.lock();
        try {
            refill();
            if (availableTokens >= 1) {
                availableTokens--;
                return true;
            }
            return false;
        } finally {
            lock.unlock();
        }
    }

    private void refill() {
        long now = System.nanoTime();
        double elapsed = (now - lastRefillTime) / 1_000_000_000.0;
        availableTokens = Math.min(maxTokens, availableTokens + elapsed * refillRate);
        lastRefillTime = now;
    }
}

public class StampedeProtectedCache<K, V> {
    private final ConcurrentHashMap<K, V> cache = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<K, CompletableFuture<V>> inflight = new ConcurrentHashMap<>();
    private final Function<K, V> loader;

    public V get(K key) {
        V cached = cache.get(key);
        if (cached != null) return cached;

        // Only ONE thread loads; others wait on the same future
        CompletableFuture<V> future = inflight.computeIfAbsent(key, k ->
            CompletableFuture.supplyAsync(() -> {
                V value = loader.apply(k);
                cache.put(k, value);
                inflight.remove(k);
                return value;
            })
        );
        return future.join();
    }
}

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, r -> {
    Thread t = new Thread(r, "health-check");
    t.setDaemon(true);
    return t;
});

ScheduledFuture<?> task = scheduler.scheduleAtFixedRate(
    () -> {
        try { checkHealth(); }
        catch (Exception e) { log.warn("Health check failed", e); }
    },
    0, 30, TimeUnit.SECONDS  // initial delay, period
);

// Graceful shutdown
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    task.cancel(false);
    scheduler.shutdown();
    try { scheduler.awaitTermination(5, TimeUnit.SECONDS); }
    catch (InterruptedException e) { scheduler.shutdownNow(); }
}));

// No synchronized, no volatile, no double-checked locking
// JVM guarantees class initialization is thread-safe
public class ExpensiveService {
    private static class Holder {
        static final ExpensiveService INSTANCE = new ExpensiveService();
    }

    public static ExpensiveService getInstance() {
        return Holder.INSTANCE;  // class loaded on first access
    }
}