import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; /** * MultithreadingDemo.java * Demonstrates Java concurrency concepts for MSc students. * Covers: Thread subclass, Runnable, synchronized methods, race conditions, * wait/notify (producer-consumer), ExecutorService with Callable/Future, * AtomicInteger, and CountDownLatch. * * Compile: javac MultithreadingDemo.java * Run: java MultithreadingDemo */ public class MultithreadingDemo { // ================================================================ // 1. Thread subclass // ================================================================ static class GreeterThread extends Thread { private String message; GreeterThread(String message) { this.message = message; } @Override public void run() { for (int i = 1; i <= 3; i++) { System.out.println("[Thread subclass] " + message + " #" + i + " (thread=" + Thread.currentThread().getName() + ")"); try { Thread.sleep(100); } catch (InterruptedException ignored) {} } } } // ================================================================ // 2. Runnable implementation // ================================================================ static class CounterTask implements Runnable { private String label; CounterTask(String label) { this.label = label; } @Override public void run() { for (int i = 1; i <= 3; i++) { System.out.println("[Runnable] " + label + " count=" + i + " (thread=" + Thread.currentThread().getName() + ")"); try { Thread.sleep(80); } catch (InterruptedException ignored) {} } } } // ================================================================ // 3. Race condition demo (unsynchronized) // ================================================================ static class UnsafeCounter { int count = 0; void increment() { count++; } // NOT thread-safe } // ================================================================ // 4. Synchronized counter (thread-safe) // ================================================================ static class SafeCounter { private int count = 0; synchronized void increment() { count++; } synchronized int getCount() { return count; } } // ================================================================ // 5. Producer-Consumer with wait/notify // ================================================================ static class SharedBuffer { private final int[] buffer; private int count = 0; private final int capacity; SharedBuffer(int capacity) { this.capacity = capacity; this.buffer = new int[capacity]; } synchronized void produce(int item) throws InterruptedException { // Wait while buffer is full while (count == capacity) { System.out.println(" [Producer] Buffer full, waiting..."); wait(); } buffer[count++] = item; System.out.println(" [Producer] Produced: " + item + " (buffer size=" + count + ")"); notifyAll(); // wake up consumers } synchronized int consume() throws InterruptedException { // Wait while buffer is empty while (count == 0) { System.out.println(" [Consumer] Buffer empty, waiting..."); wait(); } int item = buffer[--count]; System.out.println(" [Consumer] Consumed: " + item + " (buffer size=" + count + ")"); notifyAll(); // wake up producers return item; } } // ================================================================ // Main // ================================================================ public static void main(String[] args) throws Exception { // --- 1. Thread subclass --- System.out.println("=== Thread Subclass ==="); GreeterThread t1 = new GreeterThread("Hello"); t1.start(); t1.join(); // wait for t1 to finish System.out.println(); // --- 2. Runnable --- System.out.println("=== Runnable Implementation ==="); Thread t2 = new Thread(new CounterTask("TaskA")); Thread t3 = new Thread(new CounterTask("TaskB")); t2.start(); t3.start(); t2.join(); t3.join(); System.out.println(); // --- 3. Race condition --- System.out.println("=== Race Condition (Unsynchronized) ==="); UnsafeCounter unsafe = new UnsafeCounter(); Thread[] racers = new Thread[10]; for (int i = 0; i < racers.length; i++) { racers[i] = new Thread(() -> { for (int j = 0; j < 1000; j++) unsafe.increment(); }); racers[i].start(); } for (Thread t : racers) t.join(); System.out.println("Expected: 10000, Actual (unsafe): " + unsafe.count); System.out.println("(If actual != expected, a race condition occurred)\n"); // --- 4. Synchronized --- System.out.println("=== Synchronized Counter ==="); SafeCounter safe = new SafeCounter(); Thread[] safeThreads = new Thread[10]; for (int i = 0; i < safeThreads.length; i++) { safeThreads[i] = new Thread(() -> { for (int j = 0; j < 1000; j++) safe.increment(); }); safeThreads[i].start(); } for (Thread t : safeThreads) t.join(); System.out.println("Expected: 10000, Actual (safe): " + safe.getCount()); System.out.println(); // --- 5. Producer-Consumer --- System.out.println("=== Producer-Consumer (wait/notify) ==="); SharedBuffer buf = new SharedBuffer(3); Thread producer = new Thread(() -> { try { for (int i = 1; i <= 6; i++) { buf.produce(i); Thread.sleep(50); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); Thread consumer = new Thread(() -> { try { for (int i = 1; i <= 6; i++) { buf.consume(); Thread.sleep(120); // consumer is slower } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); producer.start(); consumer.start(); producer.join(); consumer.join(); System.out.println(); // --- 6. ExecutorService with Callable/Future --- System.out.println("=== ExecutorService + Callable/Future ==="); ExecutorService executor = Executors.newFixedThreadPool(3); List> futures = new java.util.ArrayList<>(); for (int i = 1; i <= 5; i++) { final int taskId = i; Callable task = () -> { Thread.sleep(200); return "Result from task " + taskId + " (thread=" + Thread.currentThread().getName() + ")"; }; futures.add(executor.submit(task)); } // Collect results for (Future f : futures) { System.out.println(" " + f.get()); // blocks until result ready } executor.shutdown(); System.out.println(); // --- 7. AtomicInteger --- System.out.println("=== AtomicInteger (lock-free thread safety) ==="); AtomicInteger atomicCount = new AtomicInteger(0); Thread[] atomicThreads = new Thread[10]; for (int i = 0; i < atomicThreads.length; i++) { atomicThreads[i] = new Thread(() -> { for (int j = 0; j < 1000; j++) { atomicCount.incrementAndGet(); } }); atomicThreads[i].start(); } for (Thread t : atomicThreads) t.join(); System.out.println("Expected: 10000, Actual (atomic): " + atomicCount.get()); System.out.println(); // --- 8. CountDownLatch --- System.out.println("=== CountDownLatch ==="); int workerCount = 4; CountDownLatch latch = new CountDownLatch(workerCount); ExecutorService pool = Executors.newFixedThreadPool(workerCount); for (int i = 1; i <= workerCount; i++) { final int id = i; pool.submit(() -> { try { int delay = id * 100; System.out.println(" Worker " + id + " starting (will take " + delay + " ms)"); Thread.sleep(delay); System.out.println(" Worker " + id + " done"); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { latch.countDown(); // signal completion } }); } System.out.println(" Main thread waiting for all workers..."); latch.await(); // blocks until count reaches 0 System.out.println(" All workers finished! Main thread continues."); pool.shutdown(); System.out.println("\n--- MultithreadingDemo complete ---"); } }