Remove busy waiting in ServiceExecutor and improve message processing

Author: Kaaldut8

commander/src/main/java/com/iluwatar/commander/Retry.java

@@ -28,6 +28,9 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 
@@ -53,6 +56,7 @@ public interface HandleErrorIssue<T> {
   }
 
   private static final SecureRandom RANDOM = new SecureRandom();
+  private static final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
 
   private final Operation op;
   private final HandleErrorIssue<T> handleError;
@@ -98,9 +102,23 @@ public void perform(List<Exception> list, T obj) {
           long testDelay =
               (long) Math.pow(2, this.attempts.intValue()) * 1000 + RANDOM.nextInt(1000);
           long delay = Math.min(testDelay, this.maxDelay);
-          Thread.sleep(delay);
-        } catch (InterruptedException f) {
-          // ignore
+          scheduler.schedule(() -> perform(list, obj), delay, TimeUnit.MILLISECONDS);
+          return;
+        } catch (Exception j) {
+          this.errors.add(j);
+
+          if (this.attempts.incrementAndGet() >= this.maxAttempts || !this.test.test(j)) {
+            this.handleError.handleIssue(obj, j);
+            return;
+          }
+
+          long testDelay =
+              (long) Math.pow(2, this.attempts.intValue()) * 1000 + RANDOM.nextInt(1000);
+
+          long delay = Math.min(testDelay, this.maxDelay);
+
+          scheduler.schedule(() -> perform(list, obj), delay, TimeUnit.MILLISECONDS);
+          return;
         }
       }
     } while (true);

microservices-log-aggregation/src/main/java/com/iluwatar/logaggregation/LogAggregator.java

@@ -25,7 +25,7 @@
 package com.iluwatar.logaggregation;
 
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -44,7 +44,7 @@ public class LogAggregator {
   private final CentralLogStore centralLogStore;
   private final ConcurrentLinkedQueue<LogEntry> buffer = new ConcurrentLinkedQueue<>();
   private final LogLevel minLogLevel;
-  private final ExecutorService executorService = Executors.newSingleThreadExecutor();
+  private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
   private final AtomicInteger logCount = new AtomicInteger(0);
 
   /**
@@ -104,16 +104,11 @@ private void flushBuffer() {
   }
 
   private void startBufferFlusher() {
-    executorService.execute(
-        () -> {
-          while (!Thread.currentThread().isInterrupted()) {
-            try {
-              Thread.sleep(5000); // Flush every 5 seconds.
-              flushBuffer();
-            } catch (InterruptedException e) {
-              Thread.currentThread().interrupt();
-            }
-          }
-        });
+    executorService.scheduleAtFixedRate(
+        this::flushBuffer,
+        5,
+        5,
+        TimeUnit.SECONDS
+    );
   }
 }

microservices-self-registration/eurekaserver/pom.xml

@@ -49,6 +49,18 @@
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
             </plugin>
+            <plugin>
+                <groupId>com.diffplug.spotless</groupId>
+                <artifactId>spotless-maven-plugin</artifactId>
+                <version>2.44.4</version>
+                <configuration>
+                    <java>
+                        <googleJavaFormat>
+                            <version>1.17.0</version>
+                        </googleJavaFormat>
+                    </java>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 

microservices-self-registration/eurekaserver/src/main/java/com/learning/eurekaserver/EurekaserverApplication.java

@@ -8,8 +8,7 @@
 @EnableEurekaServer
 public class EurekaserverApplication {
 
-	public static void main(String[] args) {
-		SpringApplication.run(EurekaserverApplication.class, args);
-	}
-
+  public static void main(String[] args) {
+    SpringApplication.run(EurekaserverApplication.class, args);
+  }
 }

microservices-self-registration/eurekaserver/src/test/java/com/learning/eurekaserver/EurekaserverApplicationTests.java

@@ -6,12 +6,13 @@
 @SpringBootTest
 class EurekaserverApplicationTests {
 
-	@Test
-	void contextLoads() {
-    // This is a basic integration test that checks if the Spring Application Context loads successfully.
+  @Test
+  void contextLoads() {
+    // This is a basic integration test that checks if the Spring Application Context loads
+    // successfully.
     // If the context loads without any exceptions, the test is considered passing.
     // It is often left empty as the act of loading the context is the primary verification.
-    // You can add specific assertions here if you want to verify the presence or state of certain beans.
-	}
-
+    // You can add specific assertions here if you want to verify the presence or state of certain
+    // beans.
+  }
 }