1. import java.util.ArrayList;
  2. import java.util.List;
  3. import java.util.Queue;
  4. import java.util.LinkedList;
  5. import java.util.concurrent.ExecutorService;
  6. import java.util.concurrent.Executors;
  7. import java.util.concurrent.TimeUnit;
  9. public class LogBatcher {
  11. private final int batchSize;
  12. private final List<LogEntry> logEntries = new ArrayList<>();
  13. private final Queue<List<LogEntry>> batchQueue = new LinkedList<>();
  14. private final int numThreads;
  15. private final ExecutorService executorService;
  17. public LogBatcher(int batchSize, int numThreads) {
  18. this.batchSize = batchSize;
  19. this.numThreads = numThreads;
  20. this.executorService = Executors.newFixedThreadPool(numThreads);
  21. }
  23. public void addLogEntry(LogEntry entry) {
  24. logEntries.add(entry);
  26. if (logEntries.size() >= batchSize) {
  27. List<LogEntry> batch = new ArrayList<>(logEntries); // Create a copy to avoid modification issues
  28. batchQueue.offer(batch);
  29. logEntries.clear(); // Clear the logEntries list
  30. }
  31. }
  33. public void startBatching() {
  34. for (int i = 0; i < numThreads; i++) {
  35. executorService.submit(processBatch);
  36. }
  37. }
  39. private void processBatch() {
  40. while (true) {
  41. List<LogEntry> batch = batchQueue.poll(); // Get a batch from the queue
  42. if (batch == null) {
  43. break; // No more batches to process
  44. }
  46. try {
  47. processBatchInternally(batch);
  48. } catch (Exception e) {
  49. // Handle errors during batch processing (fallback logic)
  50. System.err.println("Error processing batch: " + e.getMessage());
  51. //Consider logging the error or implementing more sophisticated retry logic
  52. }
  53. }
  54. executorService.shutdown(); // Shutdown the thread after processing all batches
  55. }
  57. private void processBatchInternally(List<LogEntry> batch) {
  58. try {
  59. // Simulate some operation that processes the log entries
  60. // Replace this with your actual log processing logic
  61. System.out.println("Processing batch of size: " + batch.size());
  62. for (LogEntry entry : batch) {
  63. System.out.println(" - " + entry.getMessage());
  64. }
  65. // Simulate some delay
  66. TimeUnit.MILLISECONDS.sleep(100);
  67. } catch (InterruptedException e) {
  68. Thread.currentThread().interrupt(); // Restore interrupted status
  69. System.err.println("Batch processing interrupted: " + e.getMessage());
  70. //Fallback Logic: Log the batch to a file if processing fails.
  71. System.out.println("Logging batch to file due to interruption.");
  72. }
  73. }
  77. public void shutdown() {
  78. executorService.shutdownNow();
  79. }
  81. public static void main(String[] args) throws InterruptedException {
  82. int batchSize = 5;
  83. int numThreads = 3;
  84. LogBatcher batcher = new LogBatcher(batchSize, numThreads);
  86. // Simulate log entries being added
  87. for (int i = 0; i < 20; i++) {
  88. batcher.addLogEntry(new LogEntry("Log message " + i));
  89. Thread.sleep(50);
  90. }
  92. batcher.startBatching();
  94. //Wait for the threads to complete. Important to avoid premature exit.
  95. batcher.executorService.awaitTermination(5, TimeUnit.SECONDS);
  97. batcher.shutdown();
  98. }
  100. // Simple LogEntry class for demonstration
  101. public static class LogEntry {
  102. private final String message;
  104. public LogEntry(String message) {
  105. this.message = message;
  106. }
  108. public String getMessage() {
  109. return message;
  110. }
  111. }
  112. }

Add your comment