1. import java.util.concurrent.*;
  2. import java.util.List;
  3. import java.util.ArrayList;
  4. import java.util.Map;
  5. import java.util.HashMap;
  7. class TaskQueueAggregator {
  9. private final List<TaskQueue> taskQueues;
  10. private final int retryIntervalMillis;
  11. private final int maxRetries;
  13. public TaskQueueAggregator(List<TaskQueue> taskQueues, int retryIntervalMillis, int maxRetries) {
  14. this.taskQueues = taskQueues;
  15. this.retryIntervalMillis = retryIntervalMillis;
  16. this.maxRetries = maxRetries;
  17. }
  20. public Map<String, Integer> aggregateTasks() throws InterruptedException {
  21. Map<String, Integer> aggregatedResults = new HashMap<>();
  22. ExecutorService executor = Executors.newFixedThreadPool(taskQueues.size());
  24. for (TaskQueue taskQueue : taskQueues) {
  25. executor.submit(() -> processQueue(taskQueue, aggregatedResults));
  26. }
  28. executor.shutdown();
  29. executor.awaitTermination(60, TimeUnit.SECONDS); // Adjust timeout as needed.
  30. return aggregatedResults;
  31. }
  34. private void processQueue(TaskQueue taskQueue, Map<String, Integer> aggregatedResults) {
  35. for (Task task : taskQueue.getTasks()) {
  36. int retries = 0;
  37. boolean success = false;
  39. while (retries <= maxRetries) {
  40. try {
  41. success = task.execute(); // Execute the task
  42. if (success) {
  43. aggregatedResults.put(task.getTaskId(), task.getValue()); // Update aggregated results.
  44. break; // Task completed successfully, exit retry loop.
  45. } else {
  46. retries++;
  47. if (retries <= maxRetries) {
  48. Thread.sleep(retryIntervalMillis); // Wait before retrying
  49. }
  50. }
  51. } catch (Exception e) {
  52. retries++;
  53. if (retries <= maxRetries) {
  54. Thread.sleep(retryIntervalMillis);
  55. }
  56. // Log the exception (optional)
  57. System.err.println("Task " + task.getTaskId() + " failed. Retry: " + retries + ", Error: " + e.getMessage());
  58. }
  59. }
  60. }
  61. }
  64. //Inner class representing a TaskQueue
  65. static class TaskQueue {
  66. private final List<Task> tasks;
  68. public TaskQueue(List<Task> tasks) {
  69. this.tasks = tasks;
  70. }
  72. public List<Task> getTasks() {
  73. return tasks;
  74. }
  75. }
  78. //Inner class representing a Task
  79. static class Task {
  80. private final String taskId;
  81. private final Runnable taskFunction;
  83. public Task(String taskId, Runnable taskFunction) {
  84. this.taskId = taskId;
  85. this.taskFunction = taskFunction;
  86. }
  88. public String getTaskId() {
  89. return taskId;
  90. }
  92. public int execute() {
  93. try {
  94. taskFunction.run();
  95. return 1; // Indicate success
  96. } catch (Exception e) {
  97. throw e; // Re-throw the exception for retry logic.
  98. }
  99. }
  101. public int getValue() {
  102. return 0; // Default value. Override in specific task if needed.
  103. }
  106. }
  108. public static void main(String[] args) throws InterruptedException {
  109. // Example Usage
  110. List<Task> task1 = new ArrayList<>();
  111. task1.add(new Task("task1", () -> { System.out.println("Task 1 executing");}));
  113. List<Task> task2 = new ArrayList<>();
  114. task2.add(new Task("task2", () -> { System.out.println("Task 2 executing");}));
  116. List<Task> task3 = new ArrayList<>();
  117. task3.add(new Task("task3", () -> { throw new RuntimeException("Simulated error");}));
  120. List<TaskQueue> taskQueues = new ArrayList<>();
  121. taskQueues.add(new TaskQueue(task1));
  122. taskQueues.add(new TaskQueue(task2));
  123. taskQueues.add(new TaskQueue(task3));
  126. TaskQueueAggregator aggregator = new TaskQueueAggregator(taskQueues, 500, 3); //Retry every 500ms, max 3 retries

Add your comment