Java Multithreading, Concurrency & Thread Synchronization

package com.accenture.ltt.multithreading;

public class SavingAccount {
    private double balance;

    public SavingAccount(double initialBalance) {
        this.balance = initialBalance;
    }

    // Synchronized method: Acquires the intrinsic lock of 'this' instance
    public synchronized void withdraw(double amount) {
        if (balance >= amount) {
            System.out.println(Thread.currentThread().getName() + " withdrawing: $" + amount);
            balance -= amount;
        } else {
            System.out.println("Insufficient balance for " + Thread.currentThread().getName());
        }
    }

    public synchronized double getBalance() {
        return balance;
    }
}

import java.util.concurrent.Executors;

public class VirtualThreadsDemo {
    public static void main(String[] args) {
        // Java 21: Auto-managed virtual thread executor
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 1; i <= 10_000; i++) {
                final int taskId = i;
                executor.submit(() -> {
                    // Simulates blocking I/O: Virtual thread unmounts from carrier thread during sleep!
                    Thread.sleep(1000);
                    System.out.println("Virtual task " + taskId + " completed on: " + Thread.currentThread());
                    return taskId;
                });
            }
        } // Auto-waits for all 10,000 tasks upon try-with-resources exit
    }
}

import java.util.*;
import java.util.stream.Collectors;

public class StreamPipelineCompleteDemo {
    record Employee(String name, String dept, double salary, List<String> skills) {}

    public static void main(String[] args) {
        List<Employee> employees = List.of(
            new Employee("Alice", "IT", 85000, List.of("Java", "Docker")),
            new Employee("Bob", "HR", 62000, List.of("Recruiting")),
            new Employee("Charlie", "IT", 92000, List.of("Spring", "Kubernetes"))
        );

        // 1. Filter, Map & Collect: IT high-earners
        List<String> itStars = employees.stream()
            .filter(e -> e.dept().equals("IT") && e.salary() > 80000)
            .map(Employee::name)
            .collect(Collectors.toList());
        System.out.println("IT Stars: " + itStars);

        // 2. FlatMap: Flatten multiple lists of skills into one unique stream
        Set<String> allSkills = employees.stream()
            .flatMap(e -> e.skills().stream())
            .collect(Collectors.toSet());
        System.out.println("Distinct Skills: " + allSkills);

        // 3. Collectors.groupingBy: Group employees by Department
        Map<String, List<Employee>> byDept = employees.stream()
            .collect(Collectors.groupingBy(Employee::dept));
        System.out.println("Grouped By Dept: " + byDept.keySet());
    }
}

import java.util.Optional;

public class OptionalCompleteDemo {
    public static Optional<String> findUserEmail(String userId) {
        if ("user_101".equals(userId)) {
            return Optional.of("john@accenture.com");
        }
        return Optional.empty(); // Returns clean Optional, NEVER null!
    }

    public static void main(String[] args) {
        // 1. orElse: Safe default fallback value
        String email = findUserEmail("user_999").orElse("no-reply@accenture.com");
        System.out.println("Email with fallback: " + email);

        // 2. map & ifPresent: Fluent chain without if-checks
        findUserEmail("user_101")
            .map(String::toUpperCase)
            .ifPresent(e -> System.out.println("Uppercase Email: " + e));

        // 3. orElseThrow: Standard Spring Boot pattern for 404 Not Found
        try {
            findUserEmail("unknown").orElseThrow(() -> new RuntimeException("User Not Found"));
        } catch (Exception e) {
            System.err.println("Caught Expected Exception: " + e.getMessage());
        }
    }
}

import java.util.concurrent.CompletableFuture;

public class CompletableFutureAsyncDemo {
    public static void main(String[] args) {
        // Async Pipeline: Fetch price, apply tax, and print without blocking main thread!
        CompletableFuture<Double> pipeline = CompletableFuture.supplyAsync(() -> {
            // Stage 1: Async REST / DB call
            System.out.println("Stage 1 fetching base price on: " + Thread.currentThread().getName());
            return 100.0;
        }).thenApply(basePrice -> {
            // Stage 2: Transform result (calculate 18% tax)
            System.out.println("Stage 2 applying tax on: " + Thread.currentThread().getName());
            return basePrice * 1.18;
        }).exceptionally(ex -> {
            // Error handler fallback
            System.err.println("Fallback error: " + ex.getMessage());
            return 0.0;
        });

        // join() blocks only at the end to retrieve computed result
        System.out.println("Final Total: $" + pipeline.join());
    }
}

import java.util.concurrent.locks.ReentrantLock;

public class SignalWaitNotifyDemo {
    private boolean ready = false;
    private final ReentrantLock lock = new ReentrantLock();

    // Coordination with wait() / notifyAll(): MUST use while loop to guard against spurious wakeups!
    public synchronized void waitForSignal() throws InterruptedException {
        while (!ready) {
            wait(); // Releases monitor lock and suspends thread!
        }
        System.out.println(Thread.currentThread().getName() + " woke up: Signal received!");
    }

    public synchronized void sendSignal() {
        ready = true;
        notifyAll(); // Wakes up all waiting threads!
    }
}

package com.accenture.ltt.threading.sample;

// Shared Data Monitor between Sender (Producer) and Receiver (Consumer)
class Data {
    private String packet;
    // true = receiver waits; false = sender waits
    private boolean transfer = true;

    public synchronized void send(String packet) {
        while (!transfer) {
            try {
                wait(); // Sender waits until receiver has processed previous packet
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        transfer = false;
        this.packet = packet;
        notifyAll(); // Wakes up receiver
    }

    public synchronized String receive() {
        while (transfer) {
            try {
                wait(); // Receiver waits until sender supplies packet
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        transfer = true;
        notifyAll(); // Wakes up sender
        return packet;
    }
}

public class SenderReceiverTest {
    public static void main(String[] args) {
        Data sharedData = new Data();
        new Thread(() -> {
            String[] packets = { "Packet 1", "Packet 2", "End" };
            for (String p : packets) sharedData.send(p);
        }).start();

        new Thread(() -> {
            String p = sharedData.receive();
            while (!"End".equals(p)) {
                System.out.println("Received: " + p);
                p = sharedData.receive();
            }
        }).start();
    }
}

package com.accenture.ltt.threading.sample;

public class ThreadJoinDemo extends Thread {
    @Override
    public void run() {
        for (int i = 0; i < 3; i++) {
            System.out.println("Worker thread computing step: " + i);
            try { Thread.sleep(500); } catch (InterruptedException e) {}
        }
        System.out.println("Worker thread finished!");
    }

    public static void main(String[] args) throws InterruptedException {
        ThreadJoinDemo worker = new ThreadJoinDemo();
        worker.start();

        // join() forces the calling thread (main) to wait until 'worker' terminates!
        worker.join();

        System.out.println("Main thread resumes after worker completed!");
    }
}

import java.util.concurrent.*;

public class ExecutorServicePoolDemo {
    public static void main(String[] args) {
        // Reusable Fixed Pool of 3 threads processing 6 submitted tasks
        ExecutorService pool = Executors.newFixedThreadPool(3);
        try {
            for (int i = 1; i <= 6; i++) {
                final int taskId = i;
                pool.submit(() -> {
                    System.out.println("Task " + taskId + " running on worker: " + Thread.currentThread().getName());
                });
            }
        } finally {
            pool.shutdown(); // Always shutdown pools to prevent thread leaks!
        }
    }
}