添加triggering异步逻辑

Author: saleson

spring-cloud-gray-event/src/main/java/cn/springcloud/gray/event/server/AbstractGrayEventTrigger.java

@@ -1,5 +1,6 @@
 package cn.springcloud.gray.event.server;
 
+import cn.springcloud.gray.concurrent.DefaultThreadFactory;
 import cn.springcloud.gray.event.GrayEvent;
 import cn.springcloud.gray.keeper.ListKeeper;
 import cn.springcloud.gray.keeper.SyncListKeeper;
@@ -8,6 +9,10 @@
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author saleson
@@ -19,13 +24,17 @@ public abstract class AbstractGrayEventTrigger implements GrayEventTrigger {
     private GrayEventSender grayEventSender;
     private GenericRetriever<EventConverter> genericRetriever;
     private ListKeeper<GrayEventObserver> grayEventObservers = new SyncListKeeper<>();
+    private ExecutorService executorService;
+    private long triggerDelayMills = 100;
 
     public AbstractGrayEventTrigger(GrayEventSender grayEventSender) {
-        this(grayEventSender, null);
+        this(grayEventSender, null,
+                new ThreadPoolExecutor(10, 50, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100), new DefaultThreadFactory("event-trigger")));
     }
 
-    public AbstractGrayEventTrigger(GrayEventSender grayEventSender, List<EventConverter> eventConverters) {
+    public AbstractGrayEventTrigger(GrayEventSender grayEventSender, List<EventConverter> eventConverters, ExecutorService executorService) {
         this.grayEventSender = grayEventSender;
+        this.executorService = executorService;
         if (!Objects.isNull(eventConverters)) {
             createEventConverterRetriever(eventConverters);
         }
@@ -34,22 +43,16 @@ public AbstractGrayEventTrigger(GrayEventSender grayEventSender, List<EventConve
 
     @Override
     public void triggering(Object eventSource, TriggerType triggerType) {
-        GrayEvent grayEvent = convertGrayEvent(eventSource, triggerType);
-        if (Objects.isNull(grayEvent)) {
-//            log.warn("转换失败, grayEvent is null, eventSource:{}, triggerType:{}", eventSource, triggerType);
-            return;
-        }
-
-        if (Objects.isNull(grayEvent.getTriggerType())) {
-            grayEvent.setTriggerType(triggerType);
-        }
-
-        noticeObservers(GrayEventObserveState.CREATED, grayEvent);
-
-//        noticeObservers(GrayEventObserveState.READY_FOR_SENDING, grayEvent);
-        grayEventSender.send(grayEvent);
-
-        noticeObservers(GrayEventObserveState.SENT, grayEvent);
+        executorService.execute(() -> {
+            long triggerDelay = getTriggerDelayMills();
+            if (triggerDelay > 0) {
+                try {
+                    Thread.sleep(triggerDelay);
+                } catch (InterruptedException e) {
+                }
+            }
+            innerTriggering(eventSource, triggerType);
+        });
     }
 
     protected abstract void logEventTrigger(Object eventSource, TriggerType triggerType, GrayEvent grayEvent);
@@ -72,6 +75,25 @@ public List<GrayEventObserver> getObservers() {
         return grayEventObservers.values();
     }
 
+    protected void innerTriggering(Object eventSource, TriggerType triggerType) {
+        GrayEvent grayEvent = convertGrayEvent(eventSource, triggerType);
+        if (Objects.isNull(grayEvent)) {
+//            log.warn("转换失败, grayEvent is null, eventSource:{}, triggerType:{}", eventSource, triggerType);
+            return;
+        }
+
+        if (Objects.isNull(grayEvent.getTriggerType())) {
+            grayEvent.setTriggerType(triggerType);
+        }
+
+        noticeObservers(GrayEventObserveState.CREATED, grayEvent);
+
+//        noticeObservers(GrayEventObserveState.READY_FOR_SENDING, grayEvent);
+        grayEventSender.send(grayEvent);
+
+        noticeObservers(GrayEventObserveState.SENT, grayEvent);
+    }
+
     protected void noticeObservers(GrayEventObserveState observeState, GrayEvent grayEvent) {
         getObservers().forEach(observer -> observer.observe(observeState, grayEvent));
     }
@@ -97,4 +119,11 @@ protected EventConverter getEventConverter(Class<?> sourceCls) {
     }
 
 
+    public void setTriggerDelayMills(long triggerDelayMills) {
+        this.triggerDelayMills = triggerDelayMills;
+    }
+
+    public long getTriggerDelayMills() {
+        return triggerDelayMills;
+    }
 }

spring-cloud-gray-event/src/main/java/cn/springcloud/gray/event/server/DefaultGrayEventTrigger.java

@@ -3,6 +3,7 @@
 import cn.springcloud.gray.event.GrayEvent;
 
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 
 /**
  * @author saleson
@@ -12,13 +13,19 @@ public class DefaultGrayEventTrigger extends AbstractGrayEventTrigger {
 
     private GrayEventLogger grayEventLogger;
 
-    public DefaultGrayEventTrigger(GrayEventSender grayEventSender, GrayEventLogger grayEventLogger) {
-        this(grayEventSender, null, grayEventLogger);
+    public DefaultGrayEventTrigger(
+            GrayEventSender grayEventSender,
+            GrayEventLogger grayEventLogger,
+            ExecutorService executorService) {
+        this(grayEventSender, null, grayEventLogger, executorService);
     }
 
     public DefaultGrayEventTrigger(
-            GrayEventSender grayEventSender, List<EventConverter> eventConverters, GrayEventLogger grayEventLogger) {
-        super(grayEventSender, eventConverters);
+            GrayEventSender grayEventSender,
+            List<EventConverter> eventConverters,
+            GrayEventLogger grayEventLogger,
+            ExecutorService executorService) {
+        super(grayEventSender, eventConverters, executorService);
         this.grayEventLogger = grayEventLogger;
     }
 

spring-cloud-gray-server/src/main/java/cn/springcloud/gray/server/configuration/GrayServerEventTriggeringAutoConfiguration.java

@@ -1,5 +1,6 @@
 package cn.springcloud.gray.server.configuration;
 
+import cn.springcloud.gray.concurrent.ExectorServiceUtils;
 import cn.springcloud.gray.event.codec.GrayEventCodec;
 import cn.springcloud.gray.event.codec.JsonGrayEventCodec;
 import cn.springcloud.gray.event.server.*;
@@ -13,13 +14,15 @@
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ExecutorService;
 
 /**
  * @author saleson
@@ -57,11 +60,21 @@ public GrayEventLogger grayEventLogger(GrayEventLogModule grayEventLogModule, Gr
     }
 
 
+    @Bean
+    @ConditionalOnMissingBean(name = "eventTriggerExecutorService")
+    public ExecutorService eventTriggerExecutorService() {
+        return ExectorServiceUtils.createExecutorService(grayServerEventProperties.getTriggerThreadPool());
+    }
+
     @Bean
     @ConditionalOnMissingBean
     public GrayEventTrigger grayEventTrigger(
-            GrayEventSender grayEventSender, GrayEventLogger grayEventLogger) {
-        return new DefaultGrayEventTrigger(grayEventSender, grayEventLogger);
+            GrayEventSender grayEventSender,
+            GrayEventLogger grayEventLogger,
+            @Autowired @Qualifier("eventTriggerExecutorService") ExecutorService eventTriggerExecutorService) {
+        AbstractGrayEventTrigger eventTrigger = new DefaultGrayEventTrigger(grayEventSender, grayEventLogger, eventTriggerExecutorService);
+        eventTrigger.setTriggerDelayMills(grayServerEventProperties.getTriggerDelayMills());
+        return eventTrigger;
     }
 
     @Bean

spring-cloud-gray-server/src/main/java/cn/springcloud/gray/server/configuration/properties/GrayServerEventProperties.java

@@ -1,5 +1,6 @@
 package cn.springcloud.gray.server.configuration.properties;
 
+import cn.springcloud.gray.concurrent.ConcurrnetProperties;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
@@ -23,6 +24,10 @@ public class GrayServerEventProperties {
 
     private EventTypeMapping eventTypeMapping = new EventTypeMapping();
 
+    private ConcurrnetProperties triggerThreadPool = new ConcurrnetProperties(5, 20, 30000, 100, "event-trigger");
+
+    private long triggerDelayMills = 100;
+
     @Data
     public static class EventTypeMapping {
         private List<String> clesses = new ArrayList<>();