1. import java.io.File;
  2. import java.util.concurrent.*;
  3. import java.util.concurrent.atomic.AtomicInteger;
  5. public class DirectoryScheduler {
  7. private final int maxConcurrentDirectories;
  8. private final int rateLimitSeconds;
  9. private final BlockingQueue<File> directoryQueue;
  10. private final Semaphore semaphore;
  11. private final AtomicInteger executionCount = new AtomicInteger(0);
  12. private final ExecutorService executorService;
  14. public DirectoryScheduler(int maxConcurrentDirectories, int rateLimitSeconds) {
  15. this(maxConcurrentDirectories, rateLimitSeconds, Executors.newFixedThreadPool(maxConcurrentDirectories));
  16. }
  18. public DirectoryScheduler(int maxConcurrentDirectories, int rateLimitSeconds, ExecutorService executorService) {
  19. this.maxConcurrentDirectories = maxConcurrentDirectories;
  20. this.rateLimitSeconds = rateLimitSeconds;
  21. this.directoryQueue = new LinkedBlockingQueue<>();
  22. this.semaphore = new Semaphore(maxConcurrentDirectories);
  23. this.executorService = executorService;
  24. }
  27. public synchronized void scheduleDirectory(File directory) {
  28. directoryQueue.offer(directory);
  29. try {
  30. scheduleNext();
  31. } catch (InterruptedException e) {
  32. Thread.currentThread().interrupt();
  33. System.err.println("Scheduler interrupted: " + e.getMessage());
  34. }
  35. }
  37. private void scheduleNext() throws InterruptedException {
  38. if (executionCount.get() >= maxConcurrentDirectories) {
  39. return;
  40. }
  42. long startTime = System.currentTimeMillis();
  43. semaphore.acquire(); // Wait for a slot to become available
  45. try {
  46. executorService.submit(() -> {
  47. try {
  48. processDirectory(directoryQueue.take()); // Take directory from queue
  49. System.out.println("Directory processed: " + directoryQueue.take().getAbsolutePath());
  50. } catch (Exception e) {
  51. System.err.println("Error processing directory: " + e.getMessage());
  52. } finally {
  53. semaphore.release(); // Release the semaphore after processing
  54. }
  55. });
  56. } catch (InterruptedException e) {
  57. Thread.currentThread().interrupt();
  58. System.err.println("Semaphore Interrupted: " + e.getMessage());
  59. }
  61. long endTime = System.currentTimeMillis();
  62. long duration = endTime - startTime;
  64. if (duration < rateLimitSeconds * 1000) {
  65. Thread.sleep(rateLimitSeconds * 1000 - duration); // Sleep to maintain rate limit
  66. }
  67. }
  69. private void processDirectory(File directory) throws Exception {
  70. // Simulate directory processing
  71. try {
  72. Thread.sleep(2000); // Simulate work
  73. } catch (InterruptedException e) {
  74. Thread.currentThread().interrupt();
  75. throw new Exception("Interrupted during directory processing");
  76. }
  77. }
  79. public void shutdown() {
  80. executorService.shutdown();
  81. }
  83. public static void main(String[] args) throws InterruptedException {
  84. // Example usage
  85. int maxConcurrent = 3;
  86. int rateLimit = 1;
  88. DirectoryScheduler scheduler = new DirectoryScheduler(maxConcurrent, rateLimit);
  90. // Simulate adding directories to schedule
  91. File dir1 = new File("dir1");
  92. File dir2 = new File("dir2");
  93. File dir3 = new File("dir3");
  94. File dir4 = new File("dir4");
  95. File dir5 = new File("dir5");
  98. scheduler.scheduleDirectory(dir1);
  99. scheduler.scheduleDirectory(dir2);
  100. scheduler.scheduleDirectory(dir3);
  101. scheduler.scheduleDirectory(dir4);
  102. scheduler.scheduleDirectory(dir5);
  105. Thread.sleep(10000); // Let the scheduler run for a while
  106. scheduler.shutdown();
  107. }
  108. }

Add your comment