增加任务提交时的切入方式

Author: lvyahui8

spring-boot-data-aggregator-autoconfigure/src/main/java/io/github/lvyahui8/spring/autoconfigure/BeanAggregateAutoConfiguration.java

@@ -136,6 +136,7 @@ public DataBeanAggregateQueryService dataBeanAggregateQueryService (
         service.setRuntimeSettings(runtimeSettings);
         service.setExecutorService(aggregateExecutorService());
         service.setInterceptorChain(aggregateQueryInterceptorChain());
+        service.setTaskClazz(properties.getAsyncTaskClass());
         service.setApplicationContext(applicationContext);
         return service;
     }

spring-boot-data-aggregator-autoconfigure/src/main/java/io/github/lvyahui8/spring/autoconfigure/BeanAggregateProperties.java

@@ -1,5 +1,7 @@
 package io.github.lvyahui8.spring.autoconfigure;
 
+import io.github.lvyahui8.spring.aggregate.service.AbstractAsyncQueryTask;
+import io.github.lvyahui8.spring.aggregate.service.AsyncQueryTaskAdapter;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
@@ -38,4 +40,8 @@ public class BeanAggregateProperties {
      * Ignore exception thrown by asynchronous execution, method returns null value
      */
     private boolean ignoreException = false;
+    /**
+     * Async task implement
+     */
+    private Class<? extends AbstractAsyncQueryTask> asyncTaskClass = AsyncQueryTaskAdapter.class;
 }

spring-boot-data-aggregator-core/src/main/java/io/github/lvyahui8/spring/aggregate/service/AbstractAsyncQueryTask.java

@@ -0,0 +1,26 @@
+package io.github.lvyahui8.spring.aggregate.service;
+
+import lombok.Setter;
+
+import java.util.concurrent.Callable;
+
+/**
+ * @author lvyahui ([email protected],[email protected])
+ * @since 2019/12/25 22:40
+ */
+public abstract class AbstractAsyncQueryTask<T> implements AsyncQueryTask<T> {
+    @Setter
+    private Callable<T> callable;
+    @Setter
+    private Thread      taskFromThread;
+
+    @Override
+    public T call() throws Exception {
+        try {
+            beforeExecute(taskFromThread);
+            return callable.call();
+        } finally {
+            afterExecute(taskFromThread);
+        }
+    }
+}

spring-boot-data-aggregator-core/src/main/java/io/github/lvyahui8/spring/aggregate/service/AsyncQueryTask.java

@@ -0,0 +1,27 @@
+package io.github.lvyahui8.spring.aggregate.service;
+
+import java.util.concurrent.Callable;
+
+/**
+ * @author lvyahui ([email protected],[email protected])
+ * @since 2019/12/24 23:07
+ */
+public interface AsyncQueryTask<T> extends Callable<T> {
+    /**
+     * 任务提交之前执行. 此方法在提交任务的那个线程中执行
+     */
+    void beforeSubmit();
+
+    /**
+     * 任务开始执行前执行. 此方法在异步线程中执行
+     * @param taskFrom 提交任务的那个线程
+     */
+    void beforeExecute(Thread taskFrom);
+
+    /**
+     * 任务执行结束后执行. 此方法在异步线程中执行
+     * 注意, 不管用户的方法抛出何种异常, 此方法都会执行.
+     * @param taskFrom 提交任务的那个线程
+     */
+    void afterExecute(Thread taskFrom);
+}

spring-boot-data-aggregator-core/src/main/java/io/github/lvyahui8/spring/aggregate/service/AsyncQueryTaskAdapter.java

@@ -0,0 +1,23 @@
+package io.github.lvyahui8.spring.aggregate.service;
+
+/**
+ * @author lvyahui ([email protected],[email protected])
+ * @since 2019/12/25 22:57
+ */
+public class AsyncQueryTaskAdapter<T> extends AbstractAsyncQueryTask<T> {
+
+    @Override
+    public void beforeSubmit() {
+
+    }
+
+    @Override
+    public void beforeExecute(Thread taskFrom) {
+
+    }
+
+    @Override
+    public void afterExecute(Thread taskFrom) {
+
+    }
+}

spring-boot-data-aggregator-core/src/main/java/io/github/lvyahui8/spring/aggregate/service/impl/DataBeanAggregateQueryServiceImpl.java

@@ -6,6 +6,7 @@
 import io.github.lvyahui8.spring.aggregate.interceptor.AggregateQueryInterceptorChain;
 import io.github.lvyahui8.spring.aggregate.model.*;
 import io.github.lvyahui8.spring.aggregate.repository.DataProviderRepository;
+import io.github.lvyahui8.spring.aggregate.service.AbstractAsyncQueryTask;
 import io.github.lvyahui8.spring.aggregate.service.DataBeanAggregateQueryService;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
@@ -40,6 +41,9 @@ public class DataBeanAggregateQueryServiceImpl implements DataBeanAggregateQuery
     @Setter
     private AggregateQueryInterceptorChain interceptorChain;
 
+    @Setter
+    private Class<? extends AbstractAsyncQueryTask> taskClazz ;
+
     private AggregationContext initQueryContext(DataProvideDefinition rootProvider, Map<InvokeSignature,Object> queryCache) {
         AggregationContext aggregationContext = new AggregationContext();
         aggregationContext.setRootThread(Thread.currentThread());
@@ -139,14 +143,22 @@ private Map<String, Object> getDependObjectMap(Map<String, Object> invokeParams,
         Map<String,DataConsumeDefinition> consumeDefinitionMap = new HashMap<>(consumeDefinitions.size());
         for (DataConsumeDefinition depend : consumeDefinitions) {
             consumeDefinitionMap.put(depend.getId(),depend);
-            Future<?> future = executorService.submit(() -> {
+            AbstractAsyncQueryTask queryTask = null;
+            try {
+                queryTask = taskClazz.newInstance();
+            } catch (InstantiationException ignored) {
+                //
+            }
+            queryTask.setCallable(() -> {
                 try {
                     Object o = innerGet(repository.get(depend.getId()),invokeParams, depend.getClazz(),context,depend);
                     return depend.getClazz().cast(o);
                 } finally {
                     stopDownLatch.countDown();
                 }
             });
+            queryTask.setTaskFromThread(Thread.currentThread());
+            Future<?> future = executorService.submit(queryTask);
             futureMap.put(depend.getId() + "_" + depend.getOriginalParameterName(),future);
         }
         stopDownLatch.await(timeout, TimeUnit.MILLISECONDS);

spring-boot-data-aggregator-example/src/main/resources/application.properties

@@ -1,4 +1,5 @@
 spring.main.banner-mode=off
 io.github.lvyahui8.spring.base-packages=io.github.lvyahui8.spring.example
 io.github.lvyahui8.spring.ignore-exception=false
+io.github.lvyahui8.spring.async-task-class=io.github.lvyahui8.spring.aggregate.service.AsyncQueryTaskAdapter
 example.logging=true