Merge branch 'feature/task-executor' into develop

task-executor-kit/README.md

@@ -13,7 +13,8 @@
 2. 获取任务执行实例
 
     ```java
-        TaskExecutor<User, User, String> executor = TaskPoolFactory.createExecutor();
+        TaskPoolFactory taskPoolFactory = new TaskPoolFactory();
+        TaskExecutor<User, User, String> executor = taskPoolFactory.createExecutor();
     ```
    
 3. 初始化执行器

task-executor-kit/pom.xml

@@ -2,20 +2,22 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0"
          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>pubTools</artifactId>
-        <groupId>com.mushiny</groupId>
-        <version>1.0.0-RELEASE</version>
-    </parent>
     <modelVersion>4.0.0</modelVersion>
-
+    <groupId>com.mushiny</groupId>
     <artifactId>task-executor-kit</artifactId>
+    <version>1.0.0-RELEASE</version>
+    <packaging>pom</packaging>
+    <modules>
+        <module>task-executor-core</module>
+        <module>task-executor-spring-boot-starter</module>
+    </modules>
 
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
     </properties>
 
+    <dependencyManagement>
         <dependencies>
             <dependency>
                 <groupId>junit</groupId>
@@ -24,4 +26,6 @@
                 <scope>test</scope>
             </dependency>
         </dependencies>
+    </dependencyManagement>
+
 </project>
\ No newline at end of file

task-executor-kit/task-executor-core/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>task-executor-kit</artifactId>
+        <groupId>com.mushiny</groupId>
+        <version>1.0.0-RELEASE</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>task-executor-core</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file

task-executor-kit/src/main/java/com/mushiny/task/TaskDefinition.java → task-executor-kit/task-executor-core/src/main/java/com/mushiny/task/executor/core/TaskDefinition.java

-package com.mushiny.task;
+package com.mushiny.task.executor.core;
 
 import java.util.Collection;
 import java.util.function.BiFunction;

task-executor-kit/src/main/java/com/mushiny/task/TaskExecutor.java → task-executor-kit/task-executor-core/src/main/java/com/mushiny/task/executor/core/TaskExecutor.java

-package com.mushiny.task;
+package com.mushiny.task.executor.core;
 
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Objects;
-import java.util.concurrent.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -62,6 +64,7 @@ public class TaskExecutor<T, U, R> {
         }
         countDownLatch = new CountDownLatch(data.size());
         AtomicInteger failCount = new AtomicInteger();
+
         data.forEach(d -> pool.execute(() -> {
             try {
                 Object apply = t.getHandler().apply(d, t.getContext());

task-executor-kit/src/main/java/com/mushiny/task/TaskPoolFactory.java → task-executor-kit/task-executor-core/src/main/java/com/mushiny/task/executor/core/TaskPoolFactory.java

-package com.mushiny.task;
+package com.mushiny.task.executor.core;
 
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -8,22 +8,22 @@ import java.util.concurrent.TimeUnit;
  * @author lihao
  */
 public class TaskPoolFactory {
-    TaskPoolFactory() {
+    public TaskPoolFactory() {
         // ignore
     }
 
-    public static <T, U, R> TaskExecutor<T, U, R> createExecutor() {
+    public <T, U, R> TaskExecutor<T, U, R> createExecutor() {
         int core = Runtime.getRuntime().availableProcessors();
         int poolSize = Math.min(core / 2, 4);
         return createExecutor(new ThreadPoolExecutor(poolSize, poolSize + 1, 0L, TimeUnit.MILLISECONDS,
                 new LinkedBlockingQueue<>()), 15L, TimeUnit.SECONDS);
     }
 
-    public static <T, U, R> TaskExecutor<T, U, R> createExecutor(ThreadPoolExecutor pool) {
+    public <T, U, R> TaskExecutor<T, U, R> createExecutor(ThreadPoolExecutor pool) {
         return createExecutor(pool, 15L, TimeUnit.SECONDS);
     }
 
-    public static <T, U, R> TaskExecutor<T, U, R> createExecutor(ThreadPoolExecutor pool, Long timeOut, TimeUnit timeUnit) {
+    public <T, U, R> TaskExecutor<T, U, R> createExecutor(ThreadPoolExecutor pool, Long timeOut, TimeUnit timeUnit) {
         return new TaskExecutor<>(pool, timeOut, timeUnit);
     }
 }

task-executor-kit/src/test/java/com/mushiny/task/TaskTests.java → task-executor-kit/task-executor-core/src/test/java/com/mushiny/task/executor/core/TaskTests.java

-package com.mushiny.task;
+package com.mushiny.task.executor.core;
 
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -12,21 +13,22 @@ import java.util.concurrent.TimeUnit;
 
 public class TaskTests {
 
-    private int SLEEP_TIME = 1000;
+    private int SLEEP_TIME = 100;
 
     private List<User> list;
 
     @Before
     public void init() {
         list = new ArrayList<>();
-        for (int i = 0; i < 20; i++) {
+        for (int i = 0; i < 200; i++) {
             list.add(new User(i, "leehyoo" + i));
         }
     }
 
     @Test
     public void multiTaskLoopNoArgsTest() {
-        TaskExecutor<User, User, String> executor = TaskPoolFactory.createExecutor();
+        TaskPoolFactory taskPoolFactory = new TaskPoolFactory();
+        TaskExecutor<User, User, String> executor = taskPoolFactory.createExecutor();
         executor.init((data, context) -> {
             data.setName(data.getName() + "handle");
             try {
@@ -42,26 +44,12 @@ public class TaskTests {
 
     }
 
-    @Test
-    public void normalLoopTest() {
-        User context = new User(30, "leehyoo");
-        list.forEach(data -> {
-            data.setName(data.getName() + "handle");
-            try {
-                Thread.sleep(SLEEP_TIME);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-        });
-
-
-    }
-
 
     @Test
     public void multiTaskLoopWithArgsTest() {
-        TaskExecutor<User, User, String> executor = TaskPoolFactory.createExecutor(new ThreadPoolExecutor(5, 6,
-                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()));
+        TaskPoolFactory taskPoolFactory = new TaskPoolFactory();
+        TaskExecutor<User, User, String> executor = taskPoolFactory.createExecutor(new ThreadPoolExecutor(8, 9,
+                0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>()), 60L, TimeUnit.SECONDS);
 
         executor.init((data, context) -> {
             data.setName(data.getName() + "handle");
@@ -82,7 +70,8 @@ public class TaskTests {
 
     @Test
     public void failCaseTest() {
-        TaskExecutor<User, User, String> executor = TaskPoolFactory.createExecutor();
+        TaskPoolFactory taskPoolFactory = new TaskPoolFactory();
+        TaskExecutor<User, User, String> executor = taskPoolFactory.createExecutor();
 
         executor.init((data, context) -> {
             data.setName(data.getName() + "handle");
@@ -102,8 +91,9 @@ public class TaskTests {
 
     @Test
     public void errorRecordCallbackTest() {
-        TaskExecutor<User, User, String> executor = TaskPoolFactory.createExecutor();
-        List<User> errors = new ArrayList<>();
+        TaskPoolFactory taskPoolFactory = new TaskPoolFactory();
+        TaskExecutor<User, User, String> executor = taskPoolFactory.createExecutor();
+        List<User> errors = Collections.synchronizedList(new ArrayList<>());
         executor.init((data, context) -> {
             data.setName(data.getName() + "handle");
             try {
@@ -122,4 +112,18 @@ public class TaskTests {
         Assert.assertFalse(result);
 
     }
+
+    @Test
+    public void normalLoopTest() {
+        User context = new User(30, "leehyoo");
+        list.forEach(data -> {
+            data.setName(data.getName() + "handle");
+            try {
+                Thread.sleep(SLEEP_TIME);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        });
+
+    }
 }

task-executor-kit/src/test/java/com/mushiny/task/User.java → task-executor-kit/task-executor-core/src/test/java/com/mushiny/task/executor/core/User.java

-package com.mushiny.task;
+package com.mushiny.task.executor.core;
 
 public class User {
 

task-executor-kit/task-executor-spring-boot-starter/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>task-executor-kit</artifactId>
+        <groupId>com.mushiny</groupId>
+        <version>1.0.0-RELEASE</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>task-executor-spring-boot-starter</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <spring-boot.version>2.3.12.RELEASE</spring-boot.version>
+        <spring-cloud-alibaba.version>2.2.7.RELEASE</spring-cloud-alibaba.version>
+
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+            <version>${spring-boot.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- 自动依赖配置 -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <version>${spring-boot.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- 将被@ConfigurationProperties注解的类的属性注入到元属性 -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <version>${spring-boot.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.cloud</groupId>
+            <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
+            <version>${spring-cloud-alibaba.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.mushiny</groupId>
+            <artifactId>task-executor-core</artifactId>
+            <version>1.0.0-RELEASE</version>
+        </dependency>
+    </dependencies>
+</project>
\ No newline at end of file

task-executor-kit/task-executor-spring-boot-starter/src/main/java/com/mushiny/task/executor/autoconfig/TaskExecutorAutoConfiguration.java

+package com.mushiny.task.executor.autoconfig;
+
+import com.mushiny.task.executor.core.TaskPoolFactory;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+@Configuration(proxyBeanMethods = false)
+@EnableConfigurationProperties(value = TaskExecutorProperties.class)
+public class TaskExecutorAutoConfiguration {
+
+    @Bean("listLoopThreadPool")
+    @ConditionalOnMissingBean(name = { "listLoopThreadPool" } )
+    public ThreadPoolExecutor listLoopThreadPool(TaskExecutorProperties taskExecutorProperties) {
+        TaskPoolConfig pool = taskExecutorProperties.getPool();
+        return new ThreadPoolExecutor(pool.getCorePoolSize(), pool.getMaximumPoolSize(), pool.getKeepAliveTime(), pool.getTimeUnit(), new LinkedBlockingQueue<>());
+    }
+
+    @Bean
+    @ConditionalOnBean(name = { "listLoopThreadPool" })
+    public TaskPoolFactory taskPoolFactory () {
+        return new TaskPoolFactory();
+    }
+
+
+}

task-executor-kit/task-executor-spring-boot-starter/src/main/java/com/mushiny/task/executor/autoconfig/TaskExecutorProperties.java

+package com.mushiny.task.executor.autoconfig;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+
+
+@ConfigurationProperties(prefix = "com.mushiny.tools.task.executor")
+public class TaskExecutorProperties {
+
+    private Boolean enabled;
+
+    @NestedConfigurationProperty
+    private TaskPoolConfig pool;
+
+    public Boolean getEnabled() {
+        return enabled;
+    }
+
+    public void setEnabled(Boolean enabled) {
+        this.enabled = enabled;
+    }
+
+    public TaskPoolConfig getPool() {
+        return pool;
+    }
+
+    public void setPool(TaskPoolConfig pool) {
+        this.pool = pool;
+    }
+}

task-executor-kit/task-executor-spring-boot-starter/src/main/java/com/mushiny/task/executor/autoconfig/TaskPoolConfig.java

+package com.mushiny.task.executor.autoconfig;
+
+import java.util.concurrent.TimeUnit;
+
+public class TaskPoolConfig {
+
+
+    private Integer corePoolSize;
+
+    private Integer maximumPoolSize;
+
+    private Long keepAliveTime;
+
+    private TimeUnit timeUnit;
+
+    public Integer getCorePoolSize() {
+        return corePoolSize;
+    }
+
+    public void setCorePoolSize(Integer corePoolSize) {
+        this.corePoolSize = corePoolSize;
+    }
+
+    public Integer getMaximumPoolSize() {
+        return maximumPoolSize;
+    }
+
+    public void setMaximumPoolSize(Integer maximumPoolSize) {
+        this.maximumPoolSize = maximumPoolSize;
+    }
+
+    public Long getKeepAliveTime() {
+        return keepAliveTime;
+    }
+
+    public void setKeepAliveTime(Long keepAliveTime) {
+        this.keepAliveTime = keepAliveTime;
+    }
+
+    public TimeUnit getTimeUnit() {
+        return timeUnit;
+    }
+
+    public void setTimeUnit(TimeUnit timeUnit) {
+        this.timeUnit = timeUnit;
+    }
+}

task-executor-kit/task-executor-spring-boot-starter/src/main/resources/META-INF/spring.factories

+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+  com.mushiny.task.executor.autoconfig.TaskExecutorAutoConfiguration
\ No newline at end of file