feat: run DisruptorAsyncScriptEngine on multiple threads

tomtom-international · Oct 18, 2024 · 4ffa973 · 4ffa973
1 parent cdc7c84
commit 4ffa973
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 2 deletions.
diff --git a/james-agent/src/main/java/com/tomtom/james/disruptor/JobEventHandler.java b/james-agent/src/main/java/com/tomtom/james/disruptor/JobEventHandler.java
@@ -1,9 +1,10 @@
 package com.tomtom.james.disruptor;
 
 import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.WorkHandler;
 import com.tomtom.james.common.log.Logger;
 
-public class JobEventHandler implements EventHandler<JobEvent> {
+public class JobEventHandler implements EventHandler<JobEvent>, WorkHandler<JobEvent> {
 
     private static final Logger LOG = Logger.getLogger(JobEventHandler.class);
 
@@ -16,7 +17,11 @@ public void onEvent(
         JobEvent event,
         long sequence,
         boolean endOfBatch) throws Exception {
+        onEvent(event);
+    }
 
+    @Override
+    public void onEvent(final JobEvent event) throws Exception {
         try{
 
             Runnable job = event.getJob();

diff --git a/james-agent/src/main/java/com/tomtom/james/script/DisruptorAsyncScriptEngine.java b/james-agent/src/main/java/com/tomtom/james/script/DisruptorAsyncScriptEngine.java
@@ -62,7 +62,11 @@ class DisruptorAsyncScriptEngine implements ScriptEngine, QueueBacked {
         disruptor =
                 new Disruptor<>(new JobEvent.Factory(), bufferSize, executor);
         // Start the Disruptor, starts all threads running
-        disruptor.handleEventsWith(new JobEventHandler());
+        final JobEventHandler[] workHandlers = new JobEventHandler[numberOfWorkers];
+        for (int i = 0; i < numberOfWorkers; i++) {
+            workHandlers[i] = new JobEventHandler();
+        }
+        disruptor.handleEventsWithWorkerPool(workHandlers);
         disruptor.start();
 
         this.delegate = Objects.requireNonNull(delegate);
@@ -105,6 +109,7 @@ public void invokeSuccessHandler(InformationPoint informationPoint,
                 callStack,
                 returnValue,
                 initialContextProvider)))) {
+            LOG.warn("Dropping success handler execution for " + informationPoint);
             droppedJobsCount.incrementAndGet();
         }
     }
@@ -137,6 +142,7 @@ public void invokeErrorHandler(InformationPoint informationPoint,
                         errorCause,
                         initialContextProvider)))) {
             droppedJobsCount.incrementAndGet();
+            LOG.warn("Dropping error handler execution for " + informationPoint);
         }
 
     }