Die begrenzte Queue mit Blocking wird nicht mehr benötigt

diff --git a/src/main/java/de/juplo/kafka/ExampleProducer.java b/src/main/java/de/juplo/kafka/ExampleProducer.java
index 06d5b8135d1faf3718ef3eea5f5b9469fd559509..ca5ba67858f7578ff8101ca4d722c997bc742378 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleProducer.java
+++ b/src/main/java/de/juplo/kafka/ExampleProducer.java
@@ -20,14 +20,7 @@ public class ExampleProducer
   private final String topic;
   private final Producer<String, String> producer;
 
-  private final ExecutorService executor = new ThreadPoolExecutor(
-    1,
-    1,
-    60L,
-    TimeUnit.MILLISECONDS,
-    new ArrayBlockingQueue<>(MAX_PENDING_MESSAGES),
-    new BlockingRejectedExecutionHandler()
-  );
+  private final ExecutorService executor = Executors.newSingleThreadExecutor();
   private final Semaphore semaphore = new Semaphore(MAX_PENDING_MESSAGES);
   private volatile boolean running = true;
   private volatile boolean done = false;
@@ -189,22 +182,4 @@ public class ExampleProducer
 
     instance.run();
   }
-
-
-  static class BlockingRejectedExecutionHandler implements RejectedExecutionHandler
-  {
-
-    @Override
-    public void rejectedExecution(Runnable runnable, ThreadPoolExecutor executor)
-    {
-      try
-      {
-        executor.getQueue().put(runnable);   // blockiert bis Platz frei ist
-      }
-      catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-        throw new RejectedExecutionException(e);
-      }
-    }
-  }
 }