Lesson 11

lessons/build.gradle

@@ -12,6 +12,8 @@ repositories {
 dependencies {
     testImplementation platform('org.junit:junit-bom:5.10.0')
     testImplementation 'org.junit.jupiter:junit-jupiter'
+    compileOnly 'org.projectlombok:lombok:1.18.34'
+    annotationProcessor 'org.projectlombok:lombok:1.18.34'
 }
 
 test {

lessons/src/main/java/lessons/lesson11/publisher_subscriber/App.java

@@ -0,0 +1,17 @@
+package lessons.lesson11.publisher_subscriber;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class App {
+    public static void main(String[] args) {
+        MessageQueue queue = new MessageQueue();
+
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+
+        executor.submit(new Publisher(queue));
+        executor.submit(new Subscriber(queue));
+
+        executor.shutdown();
+    }
+}

lessons/src/main/java/lessons/lesson11/publisher_subscriber/MessageQueue.java

@@ -0,0 +1,11 @@
+package lessons.lesson11.publisher_subscriber;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import lombok.Getter;
+
+@Getter
+public class MessageQueue {
+    private final BlockingQueue<String> queue = new LinkedBlockingQueue<>();
+}

lessons/src/main/java/lessons/lesson11/publisher_subscriber/Publisher.java

@@ -0,0 +1,30 @@
+package lessons.lesson11.publisher_subscriber;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.Scanner;
+
+@RequiredArgsConstructor
+public class Publisher implements Runnable {
+    private final MessageQueue queue;
+    private volatile boolean running = true;
+
+    @Override
+    public void run() {
+        try (Scanner scanner = new Scanner(System.in)) {
+            System.out.println("Words to publish ('exit' to quit):");
+            while (running) {
+                String input = scanner.nextLine();
+                if ("exit".equalsIgnoreCase(input.trim())) {
+                    running = false;
+                    queue.getQueue().put("exit");
+                } else {
+                    queue.getQueue().put(input);
+                }
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            System.out.println("Publisher interrupted.");
+        }
+    }
+}

lessons/src/main/java/lessons/lesson11/publisher_subscriber/Subscriber.java

@@ -0,0 +1,24 @@
+package lessons.lesson11.publisher_subscriber;
+
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class Subscriber implements Runnable {
+    private final MessageQueue queue;
+
+    @Override
+    public void run() {
+        try {
+            while (true) {
+                String message = queue.getQueue().take();
+                if ("exit".equalsIgnoreCase(message)) {
+                    break;
+                }
+                System.out.println("Subscriber received: " + message);
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            System.out.println("Subscriber interrupted.");
+        }
+    }
+}

lessons/src/main/java/lessons/lesson11/simplethreadpool/App.java

@@ -0,0 +1,27 @@
+package lessons.lesson11.simplethreadpool;
+
+public class App {
+    public static void main(String[] args) {
+        SimpleThreadPool pool = new SimpleThreadPool(10);
+
+        for (int i = 1; i <= 10; i++) {
+            int taskId = i;
+            pool.submit(() -> {
+                System.out.println(Thread.currentThread().getName() + " executing task " + taskId);
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
+            });
+        }
+
+        try {
+            Thread.sleep(5000);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+
+        pool.shutdown();
+    }
+}

lessons/src/main/java/lessons/lesson11/simplethreadpool/SimpleThreadPool.java

@@ -0,0 +1,47 @@
+package lessons.lesson11.simplethreadpool;
+
+import lombok.Getter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+public class SimpleThreadPool {
+
+    @Getter
+    private final BlockingQueue<Runnable> taskQueue;
+    private final List<WorkerThread> workers;
+    private volatile boolean isRunning = true;
+
+    public SimpleThreadPool(int threadCount) {
+        this.taskQueue = new LinkedBlockingQueue<>();
+        this.workers = new ArrayList<>();
+
+        for (int i = 0; i < threadCount; i++) {
+            WorkerThread worker = new WorkerThread(taskQueue);
+            worker.setName("Worker-" + (i + 1));
+            worker.start();
+            workers.add(worker);
+        }
+
+        System.out.println("Thread pool initialized with " + threadCount + " threads.");
+    }
+
+    public void submit(Runnable task) {
+        if (!isRunning) {
+            throw new IllegalStateException("Thread pool is shutting down.");
+        }
+        taskQueue.offer(task);
+    }
+
+    public void shutdown() {
+        System.out.println("Shutting down thread pool...");
+        isRunning = false;
+
+        for (WorkerThread worker : workers) {
+            worker.stopWorker();
+        }
+        System.out.println("Thread pool shutdown complete.");
+    }
+}

lessons/src/main/java/lessons/lesson11/simplethreadpool/WorkerThread.java

@@ -0,0 +1,29 @@
+package lessons.lesson11.simplethreadpool;
+
+import lombok.RequiredArgsConstructor;
+
+import java.util.concurrent.BlockingQueue;
+
+@RequiredArgsConstructor
+public class WorkerThread extends Thread {
+    private final BlockingQueue<Runnable> taskQueue;
+    private volatile boolean running = true;
+
+    @Override
+    public void run() {
+        while (running) {
+            try {
+                Runnable task = taskQueue.take();
+                task.run();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                System.out.println(Thread.currentThread().getName() + " interrupted.");
+            }
+        }
+    }
+
+    public void stopWorker() {
+        running = false;
+        this.interrupt();
+    }
+}