1. import java.util.*;
  2. import java.util.concurrent.*;
  3. import java.util.concurrent.atomic.AtomicInteger;
  5. public class TaskQueueAnomalyDetector {
  7. private final Queue<String> taskQueue = new LinkedList<>(); //Simulate a task queue
  8. private final BlockingQueue<String> realTaskQueue = new LinkedBlockingQueue<>(); //Actual task queue
  9. private final int anomalyThreshold = 5; // Define the anomaly threshold
  10. private final AtomicInteger anomalyCount = new AtomicInteger(0); // Track anomaly count
  11. private final List<String> anomalyLog = new ArrayList<>(); // Log anomalies
  13. public TaskQueueAnomalyDetector() {
  14. // Initialize the real task queue
  15. realTaskQueue.add("Task1");
  16. realTaskQueue.add("Task2");
  17. realTaskQueue.add("Task3");
  18. realTaskQueue.add("Task4");
  19. realTaskQueue.add("Task5");
  20. realTaskQueue.add("Task6");
  21. realTaskQueue.add("Task7");
  22. }
  24. public void enqueueTask(String task) {
  25. taskQueue.offer(task);
  26. realTaskQueue.add(task);
  27. }
  28. public List<String> getTaskQueue(){
  29. return taskQueue;
  30. }
  32. public void checkQueueForAnomalies() {
  33. try {
  34. // Simulate processing tasks
  35. for (int i = 0; i < 10; i++) {
  36. String task = realTaskQueue.poll();
  37. if (task != null) {
  38. processTask(task);
  39. }
  40. }
  41. } catch (InterruptedException e) {
  42. System.err.println("Queue processing interrupted: " + e.getMessage());
  43. //Handle the interruption gracefully
  44. }
  48. //Anomaly detection logic
  49. if (taskQueue.size() > anomalyThreshold) {
  50. anomalyCount.incrementAndGet();
  51. anomalyLog.add("High queue size: " + taskQueue.size());
  52. System.err.println("Anomaly detected: High queue size - " + taskQueue.size());
  54. } else if (taskQueue.isEmpty() && !realTaskQueue.isEmpty()){
  55. anomalyCount.incrementAndGet();
  56. anomalyLog.add("Empty queue, but real queue is not");
  57. System.err.println("Anomaly detected: Empty task queue but real queue is not empty.");
  59. }
  60. }
  62. private void processTask(String task) throws InterruptedException {
  63. // Simulate task processing with potential failures
  64. if (task.equals("Task5")) { //Simulate a task that may fail
  65. throw new RuntimeException("Simulated task failure for Task5");
  66. }
  67. System.out.println("Processing task: " + task);
  68. Thread.sleep(100); // Simulate processing time
  69. }
  71. public int getAnomalyCount() {
  72. return anomalyCount.get();
  73. }
  75. public List<String> getAnomalyLog() {
  76. return new ArrayList<>(anomalyLog); // Return a copy to prevent external modification
  77. }
  79. public static void main(String[] args) throws InterruptedException {
  80. TaskQueueAnomalyDetector detector = new TaskQueueAnomalyDetector();
  82. // Simulate adding tasks
  83. for (int i = 0; i < 10; i++) {
  84. detector.enqueueTask("Task" + (i + 1));
  85. }
  87. // Check for anomalies periodically
  88. for (int i = 0; i < 20; i++) {
  89. detector.checkQueueForAnomalies();
  90. Thread.sleep(500); // Check every 500ms
  91. }
  93. System.out.println("Total anomalies: " + detector.getAnomalyCount());
  94. System.out.println("Anomaly Log: " + detector.getAnomalyLog());
  95. }
  96. }

Add your comment