javadoc plugin pom.xml

pom.xml

@@ -155,6 +155,22 @@
       <id>release</id>
       <build>
         <plugins>
+          <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-javadoc-plugin</artifactId>
+            <version>3.4.1</version>
+            <executions>
+              <execution>
+                <id>attach-javadocs</id>
+                <goals>
+                  <goal>jar</goal>
+                </goals>
+              </execution>
+            </executions>
+            <configuration>
+              <doclint>none</doclint>
+            </configuration>
+          </plugin>
           <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-gpg-plugin</artifactId>
@@ -762,22 +778,6 @@
           </execution>
         </executions>
       </plugin>
-      <plugin>
-        <groupId>org.apache.maven.plugins</groupId>
-        <artifactId>maven-javadoc-plugin</artifactId>
-        <version>3.4.1</version>
-        <executions>
-          <execution>
-            <id>attach-javadocs</id>
-            <goals>
-              <goal>jar</goal>
-            </goals>
-          </execution>
-        </executions>
-        <configuration>
-          <doclint>none</doclint>
-        </configuration>
-      </plugin>
 
       <!--
         http://wvengen.github.io/proguard-maven-plugin

src/main/java/cn/ponfee/commons/collect/Collects.java

@@ -470,10 +470,10 @@ public static <T> List<T> concat(List<T> list, T... array) {
         return result;
     }
 
-    public static <T> T[] newArray(Class<? extends T[]> newType, int length) {
-        return ((Object) newType == (Object) Object[].class)
+    public static <T> T[] newArray(Class<? extends T[]> arrayType, int length) {
+        return arrayType.equals(Object[].class)
             ? (T[]) new Object[length]
-            : (T[]) Array.newInstance(newType.getComponentType(), length);
+            : (T[]) Array.newInstance(arrayType.getComponentType(), length);
     }
 
     public static <E> Stream<E> stream(Collection<E> collection) {

src/main/java/cn/ponfee/commons/concurrent/MultithreadExecutors.java

@@ -8,22 +8,14 @@
 
 package cn.ponfee.commons.concurrent;
 
-import com.google.common.base.Stopwatch;
-import org.apache.commons.collections4.CollectionUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import java.util.stream.IntStream;
 
 /**
  * Multi Thread executor
@@ -34,99 +26,35 @@
  */
 public class MultithreadExecutors {
 
-    private static final Logger LOG = LoggerFactory.getLogger(MultithreadExecutors.class);
-
-    /**
-     * Exec async, usual use in test case
-     *
-     * @param parallelism the parallelism
-     * @param command     the command
-     * @param execSeconds the execSeconds
-     * @param executor    the executor
-     */
-    public static void execute(int parallelism, Runnable command,
-                               int execSeconds, Executor executor) {
-        Stopwatch watch = Stopwatch.createStarted();
-        AtomicBoolean flag = new AtomicBoolean(true);
-
-        // CALLER_RUNS: caller run will be dead loop
-        // caller thread will be loop exec command, can't to run the after code{flag.set(false)}
-        // threadNumber > 32
-        CompletableFuture<?>[] futures = IntStream
-            .range(0, parallelism)
-            .mapToObj(i -> (Runnable) () -> {
-                while (flag.get() && !Thread.currentThread().isInterrupted()) {
-                    command.run();
-                }
-            })
-            .map(runnable -> CompletableFuture.runAsync(runnable, executor))
-            .toArray(CompletableFuture[]::new);
-
-        try {
-            // parent thread sleep
-            Thread.sleep(execSeconds * 1000L);
-            flag.set(false);
-            CompletableFuture.allOf(futures).join();
-        } catch (InterruptedException e) {
-            flag.set(false);
-            Thread.currentThread().interrupt();
-            throw new RuntimeException(e);
-        } finally {
-            LOG.info("multi thread exec async duration: {}", watch.stop());
-        }
-    }
-
-    // -----------------------------------------------------------------execAsync
-
-    /**
-     * Exec async
-     *
-     * @param command     the command
-     * @param parallelism the parallelism
-     * @param executor    thread executor service
-     */
-    public static void execute(Runnable command, int parallelism, Executor executor) {
-        Stopwatch watch = Stopwatch.createStarted();
-        CompletableFuture[] futures = IntStream.range(0, parallelism)
-                                               .mapToObj(i -> CompletableFuture.runAsync(command, executor))
-                                               .toArray(CompletableFuture[]::new);
-
-        CompletableFuture.allOf(futures).join();
-        LOG.info("multi thread run async duration: {}", watch.stop());
-    }
 
-    // -----------------------------------------------------------------callAsync
-    public static <U> List<U> execute(Supplier<U> supplier, int parallelism) {
-        Stopwatch watch = Stopwatch.createStarted();
-        List<U> result = IntStream.range(0, parallelism)
-                                  .mapToObj(i -> CompletableFuture.supplyAsync(supplier))
-                                  .collect(Collectors.toList())
-                                  .stream()
-                                  .map(CompletableFuture::join)
-                                  .collect(Collectors.toList());
-        LOG.info("multi thread call async duration: {}", watch.stop());
-        return result;
+    public static <T> void run(Collection<T> coll, Consumer<T> action, Executor executor) {
+        run(coll, action, executor, 2);
     }
 
-    // -----------------------------------------------------------------runAsync
-
     /**
      * Run async, action the T collection
      *
      * @param coll     the T collection
      * @param action   the T action
      * @param executor thread executor service
      */
-    public static <T> void execute(Collection<T> coll, Consumer<T> action, Executor executor) {
-        Stopwatch watch = Stopwatch.createStarted();
+    public static <T> void run(Collection<T> coll, Consumer<T> action, Executor executor, int dataSizeThreshold) {
+        if (coll == null || coll.isEmpty()) {
+            return;
+        }
+        if (dataSizeThreshold <= 0 || coll.size() < dataSizeThreshold) {
+            coll.forEach(action);
+            return;
+        }
         coll.stream()
             .map(e -> CompletableFuture.runAsync(() -> action.accept(e), executor))
             .collect(Collectors.toList())
             .forEach(CompletableFuture::join);
-        LOG.info("multi thread run async duration: {}", watch.stop());
     }
 
-    // -----------------------------------------------------------------callAsync
+    public static <T, U> List<U> call(Collection<T> coll, Function<T, U> mapper, Executor executor) {
+        return call(coll, mapper, executor, 2);
+    }
 
     /**
      * Call async, mapped T to U
@@ -136,91 +64,22 @@ public static <T> void execute(Collection<T> coll, Consumer<T> action, Executor
      * @param executor thread executor service
      * @return the U collection
      */
-    public static <T, U> List<U> execute(Collection<T> coll, Function<T, U> mapper, Executor executor) {
-        Stopwatch watch = Stopwatch.createStarted();
-        List<U> result = coll.stream()
-                             .map(e -> CompletableFuture.supplyAsync(() -> mapper.apply(e), executor))
-                             .collect(Collectors.toList())
-                             .stream()
-                             .map(CompletableFuture::join)
-                             .collect(Collectors.toList());
-        LOG.info("multi thread call async duration: {}", watch.stop());
-        return result;
-    }
-
-    /**
-     * 根据数据（任务）数量来判断是否主线程执行还是提交到线程池执行
-     *
-     * @param data              the data
-     * @param action            the action
-     * @param dataSizeThreshold the dataSizeThreshold
-     * @param executor          the executor
-     * @param <T>               data element type
-     * @param <R>               result element type
-     * @return list for action result
-     */
-    public static <T, R> List<R> execute(Collection<T> data, Function<T, R> action,
-                                         int dataSizeThreshold, Executor executor) {
-        if (CollectionUtils.isEmpty(data)) {
-            return Collections.emptyList();
-        }
-        if (dataSizeThreshold < 1 || data.size() < dataSizeThreshold) {
-            return data.stream().map(action).collect(Collectors.toList());
+    public static <T, U> List<U> call(Collection<T> coll, Function<T, U> mapper, Executor executor, int dataSizeThreshold) {
+        if (coll == null) {
+            return null;
         }
-
-        CompletionService<R> service = new ExecutorCompletionService<>(executor);
-        data.forEach(e -> service.submit(() -> action.apply(e)));
-        return join(service, data.size());
-    }
-
-    /**
-     * 根据数据（任务）数量来判断是否主线程执行还是提交到线程池执行
-     *
-     * @param data              the data
-     * @param action            the action
-     * @param dataSizeThreshold the dataSizeThreshold
-     * @param executor          the executor
-     * @param <T>               data element type
-     */
-    public static <T> void execute(Collection<T> data, Consumer<T> action,
-                                   int dataSizeThreshold, Executor executor) {
-        if (CollectionUtils.isEmpty(data)) {
-            return;
+        if (coll.isEmpty()) {
+            return Collections.emptyList();
         }
-        if (dataSizeThreshold < 1 || data.size() < dataSizeThreshold) {
-            data.forEach(action);
-            return;
-        }
-
-        CompletionService<Void> service = new ExecutorCompletionService<>(executor);
-        data.forEach(e -> service.submit(() -> action.accept(e), null));
-        joinDiscard(service, data.size());
-    }
-
-    // -----------------------------------------------------------------join
-    public static <T> List<T> join(CompletionService<T> service, int count) {
-        List<T> result = new ArrayList<>(count);
-        join(service, count, result::add);
-        return result;
-    }
-
-    public static <T> void joinDiscard(CompletionService<T> service, int count) {
-        join(service, count, t -> { });
-    }
-
-    public static <T> void join(CompletionService<T> service, int count, Consumer<T> accept) {
-        try {
-            while (count-- > 0) {
-                // block until a task done
-                Future<T> future = service.take();
-                accept.accept(future.get());
-            }
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new RuntimeException(e);
-        } catch (ExecutionException e) {
-            throw new RuntimeException(e);
+        if (dataSizeThreshold <= 0 || coll.size() < dataSizeThreshold) {
+            return coll.stream().map(mapper).collect(Collectors.toList());
         }
+        return coll.stream()
+            .map(e -> CompletableFuture.supplyAsync(() -> mapper.apply(e), executor))
+            .collect(Collectors.toList())
+            .stream()
+            .map(CompletableFuture::join)
+            .collect(Collectors.toList());
     }
 
 }

src/main/java/cn/ponfee/commons/concurrent/SingleThreadShutdownHook.java

@@ -40,7 +40,7 @@ private static class HookThread extends Thread {
         @Override
         public void run() {
             synchronized (LOCK) {
-                hooks.forEach(e -> ThrowingRunnable.caught(e::run));
+                hooks.forEach(e -> ThrowingRunnable.doCaught(e::run));
             }
         }
     }

src/main/java/cn/ponfee/commons/data/MultipleDataSourceAdvisor.java

@@ -81,7 +81,7 @@ public Object doAround(ProceedingJoinPoint pjp) throws Throwable {
         return around(
             ((MethodSignature) pjp.getSignature()).getMethod(),
             pjp.getArgs(),
-            ThrowingCallable.checked(pjp::proceed)
+            ThrowingCallable.toChecked(pjp::proceed)
         );
     }
 

src/main/java/cn/ponfee/commons/data/MultipleDataSourceAspect.java

@@ -58,7 +58,7 @@ public Object doAround(ProceedingJoinPoint pjp, DataSourceNaming dsn) throws Thr
         return MultipleDataSourceAdvisor.around(
             ((MethodSignature) pjp.getSignature()).getMethod(),
             pjp.getArgs(), dsn,
-            ThrowingCallable.checked(pjp::proceed)
+            ThrowingCallable.toChecked(pjp::proceed)
         );
     }