Repository navigation
Expand file tree
/
Copy pathWorkerPool.java
More file actions
74 lines (61 loc) · 2.28 KB
/
Copy pathWorkerPool.java
File metadata and controls
74 lines (61 loc) · 2.28 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
package com.example.worker;
import com.example.queue.DeadLetterQueue;
import com.example.queue.PersistentTaskQueue;
import com.example.retry.ExponentialBackoff;
import com.example.retry.RetryStrategy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class WorkerPool {
private final int poolSize;
private final PersistentTaskQueue taskQueue;
private final DeadLetterQueue dlq;
private final Map<String, TaskHandler> handlers;
private final List<Worker> workers;
private final RetryStrategy retryStrategy;
private ExecutorService executor;
public WorkerPool(int poolSize, PersistentTaskQueue taskQueue, DeadLetterQueue dlq) {
this.poolSize = poolSize;
this.taskQueue = taskQueue;
this.dlq = dlq;
this.handlers = new HashMap<>();
this.workers = new ArrayList<>();
this.retryStrategy = new ExponentialBackoff();
}
public void registerHandler(TaskHandler handler) {
handlers.put(handler.getTaskType(), handler);
System.out.println("[POOL] Registered handler for: " + handler.getTaskType());
}
public DeadLetterQueue getDeadLetterQueue() {
return dlq;
}
public void start() {
System.out.println("[POOL] Starting " + poolSize + " workers...");
executor = Executors.newFixedThreadPool(poolSize);
for (int i = 1; i <= poolSize; i++) {
Worker worker = new Worker("Worker-" + i, taskQueue, dlq, handlers, retryStrategy);
workers.add(worker);
executor.submit(worker);
}
System.out.println("[POOL] All workers started!");
}
public void shutdown() {
System.out.println("[POOL] Shutting down...");
for (Worker worker : workers) {
worker.stop();
}
executor.shutdown();
try {
if (!executor.awaitTermination(10, TimeUnit.SECONDS)) {
executor.shutdownNow();
}
} catch (InterruptedException e) {
executor.shutdownNow();
}
System.out.println("[POOL] Shutdown complete");
}
}