Merge pull request #10 from lvyahui8/develop

Develop merge into master

Author: lvyahui8

README.md

@@ -114,6 +114,8 @@ DataBeanAggregateQueryFacade dataBeanAggregateQueryFacade;
 
 #### 方式一: 函数式调用
 
+注意这里不能将函数式调用改为Lambda表达式, 两者的实际行为是不一致的.
+
 ```java
 User user = dataBeanAggregateQueryFacade.get(
      Collections.singletonMap("userId", 1L), 

pom.xml

@@ -20,7 +20,7 @@ limitations under the License.
 
     <groupId>io.github.lvyahui8</groupId>
     <artifactId>spring-boot-data-aggregator</artifactId>
-    <version>1.1.0</version>
+    <version>1.1.1</version>
     <modules>
         <module>spring-boot-data-aggregator-core</module>
         <module>spring-boot-data-aggregator-autoconfigure</module>

spring-boot-data-aggregator-autoconfigure/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>spring-boot-data-aggregator</artifactId>
         <groupId>io.github.lvyahui8</groupId>
-        <version>1.1.0</version>
+        <version>1.1.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

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

@@ -6,6 +6,7 @@
 import io.github.lvyahui8.spring.aggregate.interceptor.AggregateQueryInterceptor;
 import io.github.lvyahui8.spring.aggregate.interceptor.AggregateQueryInterceptorChain;
 import io.github.lvyahui8.spring.aggregate.interceptor.impl.AggregateQueryInterceptorChainImpl;
+import io.github.lvyahui8.spring.aggregate.model.DataConsumeDefinition;
 import io.github.lvyahui8.spring.aggregate.model.DataProvideDefinition;
 import io.github.lvyahui8.spring.aggregate.repository.DataProviderRepository;
 import io.github.lvyahui8.spring.aggregate.repository.impl.DataProviderRepositoryImpl;
@@ -14,6 +15,7 @@
 import io.github.lvyahui8.spring.aggregate.util.DefinitionUtils;
 import io.github.lvyahui8.spring.annotation.DataProvider;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.reflections.Reflections;
 import org.reflections.scanners.MethodAnnotationsScanner;
 import org.springframework.beans.BeansException;
@@ -30,7 +32,6 @@
 import org.springframework.core.annotation.Order;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
 
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
@@ -39,6 +40,7 @@
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /**
  * @author lvyahui (lvyahui8@gmail.com,lvyahui8@126.com)
@@ -65,11 +67,43 @@ public DataBeanAggregateQueryFacade dataBeanAggregateQueryFacade(
         return new DataBeanAggregateQueryFacadeImpl(dataBeanAggregateQueryService(dataProviderRepository));
     }
 
+    private void checkCycle(Map<String,Set<String>> graphAdjMap) {
+        Map<String,Integer> visitStatusMap = new HashMap<>(graphAdjMap.size() * 2);
+        for (Map.Entry<String, Set<String>> item : graphAdjMap.entrySet()) {
+            if (visitStatusMap.containsKey(item.getKey())) {
+                continue;
+            }
+            dfs(graphAdjMap,visitStatusMap,item.getKey());
+        }
+    }
+
+    private void dfs(Map<String,Set<String>> graphAdjMap,Map<String,Integer> visitStatusMap, String node) {
+        if (visitStatusMap.containsKey(node)) {
+            if(visitStatusMap.get(node) == 1) {
+                List<String> relatedNodes = new ArrayList<>();
+                for (Map.Entry<String,Integer> item : visitStatusMap.entrySet()) {
+                    if (item.getValue() == 1) {
+                        relatedNodes.add(item.getKey());
+                    }
+                }
+                throw new IllegalStateException("There are loops in the dependency graph. Related nodes:" + StringUtils.join(relatedNodes));
+            }
+            return ;
+        }
+        visitStatusMap.put(node,1);
+        log.info("visited:{}", node);
+        for (String relateNode : graphAdjMap.get(node)) {
+            dfs(graphAdjMap,visitStatusMap,relateNode);
+        }
+        visitStatusMap.put(node,2);
+    }
+
     @Bean
     @ConditionalOnMissingBean
     public DataBeanAggregateQueryService dataBeanAggregateQueryService (
             @Qualifier("dataProviderRepository") DataProviderRepository dataProviderRepository) {
         if(properties.getBasePackages() != null) {
+            Map<String,Set<String>> provideDependMap = new HashMap<>();
             for (String basePackage : properties.getBasePackages()) {
                 Reflections reflections = new Reflections(basePackage, new MethodAnnotationsScanner());
                 Set<Method> providerMethods = reflections.getMethodsAnnotatedWith(DataProvider.class);
@@ -80,13 +114,18 @@ public DataBeanAggregateQueryService dataBeanAggregateQueryService (
                     Assert.isTrue(Modifier.isPublic(method.getModifiers()),"data provider method must be public");
                     Assert.isTrue(! StringUtils.isEmpty(dataId),"data id must be not null!");
                     DataProvideDefinition provider = DefinitionUtils.getProvideDefinition(method);
+
                     provider.setId(dataId);
                     provider.setIdempotent(beanProvider.idempotent());
                     provider.setTimeout(beanProvider.timeout() > 0 ? beanProvider.timeout() : properties.getDefaultTimeout());
+                    Assert.isTrue(! dataProviderRepository.contains(dataId), "Data providers with the same name are not allowed. dataId: " + dataId);
+                    provideDependMap.put(dataId,provider.getDepends().stream().map(DataConsumeDefinition::getId).collect(Collectors.toSet()));
                     dataProviderRepository.put(provider);
                 }
             }
+            checkCycle(provideDependMap);
         }
+
         DataBeanAggregateQueryServiceImpl service = new DataBeanAggregateQueryServiceImpl();
         RuntimeSettings runtimeSettings = new RuntimeSettings();
         runtimeSettings.setEnableLogging(properties.getEnableLogging() != null
@@ -97,6 +136,7 @@ public DataBeanAggregateQueryService dataBeanAggregateQueryService (
         service.setRuntimeSettings(runtimeSettings);
         service.setExecutorService(aggregateExecutorService());
         service.setInterceptorChain(aggregateQueryInterceptorChain());
+        service.setTaskWrapperClazz(properties.getTaskWrapperClass());
         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.AsyncQueryTaskWrapper;
+import io.github.lvyahui8.spring.aggregate.service.AsyncQueryTaskWrapperAdapter;
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
@@ -13,29 +15,33 @@ public class BeanAggregateProperties {
     /**
      * Packages that need to scan for aggregated annotations
      */
-    private String[] basePackages;
+    private String[]                               basePackages;
     /**
      * Thread name prefix for asynchronous threads
      */
-    private String   threadPrefix    = "aggregateTask-";
+    private String                          threadPrefix    = "aggregateTask-";
     /**
      * Thread size of the asynchronous thread pool
      */
     private int      threadNumber    = Runtime.getRuntime().availableProcessors() * 3;
     /**
      * The size of the queue that holds the task to be executed
      */
-    private int      queueSize       = 1000;
+    private int                                    queueSize       = 1000;
     /**
      * Set a default timeout for the method of providing data
      */
-    private Long     defaultTimeout  = 3000L;
+    private Long                                   defaultTimeout   = 3000L;
     /**
      * Allow output log
      */
-    private Boolean  enableLogging   = true;
+    private Boolean                                enableLogging    = true;
     /**
      * Ignore exception thrown by asynchronous execution, method returns null value
      */
-    private boolean ignoreException = false;
+    private boolean                                ignoreException  = false;
+    /**
+     * Async task implement
+     */
+    private Class<? extends AsyncQueryTaskWrapper> taskWrapperClass = AsyncQueryTaskWrapperAdapter.class;
 }

spring-boot-data-aggregator-core/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>spring-boot-data-aggregator</artifactId>
         <groupId>io.github.lvyahui8</groupId>
-        <version>1.1.0</version>
+        <version>1.1.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

spring-boot-data-aggregator-core/src/main/java/io/github/lvyahui8/spring/aggregate/model/DataConsumeDefinition.java

@@ -2,13 +2,17 @@
 
 import lombok.Data;
 
+import java.util.Map;
+
 /**
  * @author lvyahui (lvyahui8@gmail.com,lvyahui8@126.com)
  * @since 2019/6/2 22:13
  */
 @Data
 public class DataConsumeDefinition {
-    private String id;
-    private Class<?> clazz;
-    private Boolean ignoreException;
+    private String             id;
+    private Class<?>           clazz;
+    private Boolean            ignoreException;
+    private Map<String,String> dynamicParameterKeyMap;
+    private String             originalParameterName;
 }

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

@@ -0,0 +1,38 @@
+package io.github.lvyahui8.spring.aggregate.service;
+
+import java.util.concurrent.Callable;
+
+/**
+ * @author lvyahui (lvyahui8@gmail.com,lvyahui8@126.com)
+ * @since 2019/12/25 22:40
+ */
+public abstract class AsyncQueryTask<T> implements Callable<T> {
+    Thread      taskFromThread;
+    AsyncQueryTaskWrapper asyncQueryTaskWrapper;
+
+    public AsyncQueryTask(Thread taskFromThread, AsyncQueryTaskWrapper asyncQueryTaskWrapper) {
+        this.taskFromThread = taskFromThread;
+        this.asyncQueryTaskWrapper = asyncQueryTaskWrapper;
+    }
+
+    @Override
+    public T call() throws Exception {
+        try {
+            if(asyncQueryTaskWrapper != null) {
+                asyncQueryTaskWrapper.beforeExecute(taskFromThread);
+            }
+            return execute();
+        } finally {
+            if (asyncQueryTaskWrapper != null) {
+                asyncQueryTaskWrapper.afterExecute(taskFromThread);
+            }
+        }
+    }
+
+    /**
+     *
+     * @return
+     * @throws Exception
+     */
+    public abstract T  execute() throws Exception;
+}

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

@@ -0,0 +1,25 @@
+package io.github.lvyahui8.spring.aggregate.service;
+
+/**
+ * @author lvyahui (lvyahui8@gmail.com,lvyahui8@126.com)
+ * @since 2019/12/24 23:07
+ */
+public interface AsyncQueryTaskWrapper {
+    /**
+     * 任务提交之前执行. 此方法在提交任务的那个线程中执行
+     */
+    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/AsyncQueryTaskWrapperAdapter.java

@@ -0,0 +1,23 @@
+package io.github.lvyahui8.spring.aggregate.service;
+
+/**
+ * @author lvyahui (lvyahui8@gmail.com,lvyahui8@126.com)
+ * @since 2019/12/25 22:57
+ */
+public class AsyncQueryTaskWrapperAdapter implements AsyncQueryTaskWrapper{
+
+    @Override
+    public void beforeSubmit() {
+
+    }
+
+    @Override
+    public void beforeExecute(Thread taskFrom) {
+
+    }
+
+    @Override
+    public void afterExecute(Thread taskFrom) {
+
+    }
+}