From 67fd9ad1d23f78b8e88ff36a85391ff203059ae8 Mon Sep 17 00:00:00 2001
From: godotg <jaysunxiao@gmail.com>
Date: Tue, 7 Mar 2023 19:36:30 +0800
Subject: [PATCH] perf[event]: asynchronous events are processed in different
 thread pools

---
 .../java/com/zfoo/event/manager/EventBus.java | 51 +++++++++++--------
 1 file changed, 31 insertions(+), 20 deletions(-)

diff --git a/event/src/main/java/com/zfoo/event/manager/EventBus.java b/event/src/main/java/com/zfoo/event/manager/EventBus.java
index 1b0eeaa3..99f46d8d 100644
--- a/event/src/main/java/com/zfoo/event/manager/EventBus.java
+++ b/event/src/main/java/com/zfoo/event/manager/EventBus.java
@@ -98,8 +98,37 @@ public abstract class EventBus {
      * Publish the event
      */
     public static void post(IEvent event) {
-        doSubmit(event, receiverMapSync.get(event.getClass()));
-        doSubmit(event, receiverMapAsync.get(event.getClass()));
+        if (event == null) {
+            return;
+        }
+        var clazz = event.getClass();
+        var listSync = receiverMapSync.get(clazz);
+        if (CollectionUtils.isNotEmpty(listSync)) {
+            for (var receiver : listSync) {
+                try {
+                    receiver.invoke(event);
+                } catch (Exception e) {
+                    logger.error("eventBus sync event [{}] unknown exception", clazz.getSimpleName(), e);
+                } catch (Throwable t) {
+                    logger.error("eventBus sync event [{}] unknown error", clazz.getSimpleName(), t);
+                }
+            }
+        }
+
+        var listAsync = receiverMapAsync.get(clazz);
+        if (CollectionUtils.isNotEmpty(listAsync)) {
+            for (var receiver : listAsync) {
+                executors[Math.abs(event.threadId() % EXECUTORS_SIZE)].execute(() -> {
+                    try {
+                        receiver.invoke(event);
+                    } catch (Exception e) {
+                        logger.error("eventBus async event [{}] unknown exception", clazz.getSimpleName(), e);
+                    } catch (Throwable t) {
+                        logger.error("eventBus async event [{}] unknown error", clazz.getSimpleName(), t);
+                    }
+                });
+            }
+        }
     }
 
 
@@ -114,24 +143,6 @@ public abstract class EventBus {
         executors[Math.abs(hashcode % EXECUTORS_SIZE)].execute(SafeRunnable.valueOf(runnable));
     }
 
-    /**
-     * The observer executes the method call
-     */
-    private static void doSubmit(IEvent event, List<IEventReceiver> receiverList) {
-        if (CollectionUtils.isEmpty(receiverList)) {
-            return;
-        }
-        for (var receiver : receiverList) {
-            try {
-                receiver.invoke(event);
-            } catch (Exception e) {
-                logger.error("eventBus unknown exception", e);
-            } catch (Throwable t) {
-                logger.error("eventBus unknown error", t);
-            }
-        }
-    }
-
     /**
      * Register the event and its counterpart observer
      */