Multithreading

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
// Runs every 5 seconds regardless of task duration:
scheduler.scheduleAtFixedRate(() -> doWork(), 0, 5, TimeUnit.SECONDS);
// Waits 5 seconds after previous task finishes:
scheduler.scheduleWithFixedDelay(() -> doWork(), 0, 5, TimeUnit.SECONDS);

class Node {
    volatile int state = 0;
    private static final AtomicIntegerFieldUpdater<Node> UPDATER =
        AtomicIntegerFieldUpdater.newUpdater(Node.class, "state");

    void update() { UPDATER.compareAndSet(this, 0, 1); }
}

CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "ServiceA");
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> "ServiceB");

CompletableFuture.allOf(f1, f2).thenRun(() -> {
    System.out.println("Both services responded!");
});

Thread t = Thread.currentThread();
t.interrupt();
System.out.println(t.isInterrupted()); // true
System.out.println(Thread.interrupted()); // true (clears flag)
System.out.println(t.isInterrupted()); // false

// AVOID in Virtual Threads (causes carrier thread pinning):
// synchronized (this) { makeBlockingNetworkCall(); }

// RECOMMENDED in Virtual Threads:
private final ReentrantLock lock = new ReentrantLock();
public void process() {
    lock.lock();
    try { makeBlockingNetworkCall(); } finally { lock.unlock(); }
}

// Volatile happens-before rule:
volatile boolean ready = false;
int data = 0;
// Thread 1: data = 42; ready = true; (write)
// Thread 2: if (ready) { assert data == 42; } (read guaranteed to see 42)

class SumTask extends RecursiveTask<Long> {
    protected Long compute() {
        if (size <= THRESHOLD) return computeDirectly();
        SumTask subtask = new SumTask(left);
        subtask.fork(); // Asynchronous split
        return new SumTask(right).compute() + subtask.join(); // Merge
    }
}

// Callable in ExecutorService:
Callable<String> c = () -> { if (err) throw new Exception(); return "ok"; };

// Supplier in CompletableFuture:
Supplier<String> s = () -> "ok";
CompletableFuture.supplyAsync(s);

AtomicMarkableReference<Node> node = new AtomicMarkableReference<>(targetNode, false);
// Logically mark node as deleted:
node.compareAndSet(targetNode, targetNode, false, true);

Thread t = new Thread(() -> doWork());
t.start();
t.join(); // Waits for thread 't' to finish execution and terminate

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

// In consumer:
lock.lock();
try {
    while (count == 0) notEmpty.await(); // Waits only on notEmpty
    // ... extract item ...
    notFull.signal(); // Signals only waiting producers!
} finally { lock.unlock(); }

// Obsolete legacy approach:
ThreadGroup group = new ThreadGroup("Workers");
Thread t = new Thread(group, () -> {}, "Worker-1");

// Modern approach:
ExecutorService service = Executors.newFixedThreadPool(10);

pool.shutdown(); // Refuse new tasks
try {
    if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
        pool.shutdownNow(); // Force interrupt active tasks
    }
} catch (InterruptedException e) {
    pool.shutdownNow();
}

volatile boolean eventFired = false;
// Spin-wait loop:
while (!eventFired) {
    Thread.onSpinWait(); // CPU PAUSE instruction hint
}

AtomicStampedReference<String> ref = new AtomicStampedReference<>("A", 0);
int stamp = ref.getStamp(); // 0
ref.compareAndSet("A", "B", stamp, stamp + 1); // Stamp becomes 1
ref.compareAndSet("B", "A", 1, 2);             // Stamp becomes 2
// Stale CAS expecting stamp 0 will safely FAIL even though value is 'A'!

Exchanger<List<String>> exchanger = new Exchanger<>();
// Thread A and Thread B swap populated/empty buffer lists:
// List<String> fullBuffer = getFullBuffer();
// List<String> emptyBuffer = exchanger.exchange(fullBuffer);

Phaser phaser = new Phaser(1); // Register main thread
for (int i = 0; i < 3; i++) {
    phaser.register(); // Dynamically register worker
    new Thread(() -> {
        System.out.println("Phase 1 task");
        phaser.arriveAndAwaitAdvance(); // Await phase completion
    }).start();
}
phaser.arriveAndDeregister(); // Deregister main thread

ExecutorService pool = Executors.newSingleThreadExecutor();
// execute: Fire-and-forget
pool.execute(() -> System.out.println("Executed"));

// submit: Returns Future for tracking status and exceptions
Future<String> future = pool.submit(() -> "Success");
System.out.println(future.get());

// Isolating variables to separate CPU cache lines via padding:
public class PaddedAtomicCounter {
    // @Contended (Internal JVM annotation)
    volatile long value1; // Cache line 1
    volatile long value2; // Cache line 2
}

// Livelock concept: Two polite threads continually giving way to each other
class Worker {
    private boolean active = true;
    public void work(Worker other) {
        while (active) {
            if (other.active) {
                // Yield resource back and forth infinitely
                Thread.yield();
                continue;
            }
            // Perform work
        }
    }
}

public void runTask() {
    try {
        Thread.sleep(2000);
    } catch (InterruptedException e) {
        // Proper handling: Restore interrupt flag so caller knows thread was interrupted
        Thread.currentThread().interrupt();
        System.err.println("Task interrupted cleanly");
    }
}

private static final ScopedValue<String> USER_ID = ScopedValue.newInstance();

ScopedValue.where(USER_ID, "usr_9981").run(() -> {
    System.out.println("Authenticated User: " + USER_ID.get());
});

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Supplier<String> user = scope.fork(() -> fetchUser());
    Supplier<Integer> order = scope.fork(() -> fetchOrder());
    scope.join().throwIfFailed(); // Joins both subtasks
    System.out.println(user.get() + " : " + order.get());
}

// Spawning a lightweight Virtual Thread (Java 21+):
Thread.startVirtualThread(() -> {
    System.out.println("Virtual thread executing: " + Thread.currentThread());
});

// Virtual thread executor:
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(() -> System.out.println("Task in virtual thread"));
}

ForkJoinPool pool = new ForkJoinPool();
// Parallel streams rely on common ForkJoinPool:
List<Integer> list = List.of(1, 2, 3, 4, 5);
list.parallelStream().forEach(n -> System.out.println(n + " : " + Thread.currentThread().getName()));

CompletableFuture.supplyAsync(() -> "Order #101")
    .thenApply(order -> order + " [Processed]")
    .thenAccept(System.out::println)
    .exceptionally(ex -> { System.err.println(ex.getMessage()); return null; });

LongAdder adder = new LongAdder();
// High throughput parallel updates from multiple threads:
adder.increment();
adder.add(50);
long total = adder.sum(); // Aggregates cells when total count is required

AtomicInteger counter = new AtomicInteger(0);
// Atomically increments and returns updated value via CAS:
int newVal = counter.incrementAndGet();
// Explicit CAS update:
counter.compareAndSet(1, 10); // Updates to 10 if current value is 1

public class UserContext {
    private static final ThreadLocal<String> userHolder = new ThreadLocal<>();
    public static void set(String user) { userHolder.set(user); }
    public static String get() { return userHolder.get(); }
    public static void clear() { userHolder.remove(); } // Crucial cleanup!
}

Semaphore semaphore = new Semaphore(3); // Allows max 3 concurrent threads
Runnable task = () -> {
    try {
        semaphore.acquire();
        System.out.println("Accessing resource: " + Thread.currentThread().getName());
    } finally {
        semaphore.release();
    }
};