1. import java.util.ArrayList;
  2. import java.util.List;
  3. import java.util.concurrent.BlockingQueue;
  4. import java.util.concurrent.ExecutorService;
  5. import java.util.concurrent.Executors;
  6. import java.util.concurrent.LinkedBlockingQueue;
  8. class TaskValidator {
  10. /**
  11. * Batches operations from a list of task queues for validation checks.
  12. *
  13. * @param queues A list of task queues. Each queue contains tasks to be validated.
  14. * @param validationFunction A function to validate a single task.
  15. * @param batchSize The size of the batch to process at a time.
  16. * @return A list of validation results.
  17. */
  18. public static List<String> batchValidate(List<BlockingQueue<Object>> queues, ValidationFunction validationFunction, int batchSize) {
  19. List<String> results = new ArrayList<>(); // Store validation results
  20. ExecutorService executor = Executors.newFixedThreadPool(queues.size()); // Create a thread pool
  22. for (BlockingQueue<Object> queue : queues) {
  23. executor.submit(() -> { // Submit tasks from each queue to the thread pool
  24. while (true) {
  25. try {
  26. Object task = queue.take(); // Retrieve a task from the queue
  27. if (task == null) {
  28. break; // Exit if a null task is encountered (queue is empty)
  29. }
  30. String validationResult = validationFunction.validate(task); // Validate the task
  31. results.add(validationResult); // Add the result to the list
  32. } catch (InterruptedException e) {
  33. Thread.currentThread().interrupt(); // Restore interrupted status
  34. break;
  35. }
  36. }
  37. });
  38. }
  40. executor.shutdown(); // Shutdown the thread pool
  41. try {
  42. executor.awaitTermination(Long.MAX_VALUE, java.util.concurrent.TimeUnit.NANOSECONDS); // Wait for all tasks to complete
  43. } catch (InterruptedException e) {
  44. Thread.currentThread().interrupt(); // Restore interrupted status
  45. }
  46. return results; // Return the list of validation results
  47. }
  49. /**
  50. * Functional interface for validation functions.
  51. */
  52. @FunctionalInterface
  53. public interface ValidationFunction {
  54. String validate(Object task); // Validates a single task and returns a string result
  55. }
  57. public static void main(String[] args) {
  58. // Example usage
  59. BlockingQueue<Object> queue1 = new LinkedBlockingQueue<>();
  60. BlockingQueue<Object> queue2 = new LinkedBlockingQueue<>();
  62. // Add some sample tasks to the queues
  63. queue1.add("task1");
  64. queue1.add("task2");
  65. queue2.add("task3");
  66. queue2.add("task4");
  68. List<BlockingQueue<Object>> queues = List.of(queue1, queue2);
  69. int batchSize = 1;
  71. List<String> validationResults = batchValidate(queues, task -> {
  72. // Simulate a validation check
  73. if (task.equals("task1")) {
  74. return "task1 is valid";
  75. } else if (task.equals("task2")) {
  76. return "task2 is invalid";
  77. } else if (task.equals("task3")) {
  78. return "task3 is valid";
  79. } else {
  80. return "task4 is invalid";
  81. }
  82. }, batchSize);
  84. // Print the validation results
  85. for (String result : validationResults) {
  86. System.out.println(result);
  87. }
  88. }
  89. }

Add your comment