Merge pull request #386 from KamToHung/priority-executor

[ISSUE #283] add Priority executor

core/src/main/java/org/dromara/dynamictp/core/aware/TaskEnhanceAware.java

@@ -35,19 +35,19 @@ public interface TaskEnhanceAware extends DtpAware {
     /**
      * Enhance task
      *
-     * @param command command
+     * @param command      command
      * @param taskWrappers task wrappers
      * @return enhanced task
      */
     default Runnable getEnhancedTask(Runnable command, List<TaskWrapper> taskWrappers) {
+        Runnable wrapRunnable = command;
         if (CollectionUtils.isNotEmpty(taskWrappers)) {
             for (TaskWrapper t : taskWrappers) {
-                command = t.wrap(command);
+                wrapRunnable = t.wrap(wrapRunnable);
             }
         }
-        String taskName = (command instanceof NamedRunnable) ? ((NamedRunnable) command).getName() : null;
-        command = new DtpRunnable(command, taskName);
-        return command;
+        String taskName = (wrapRunnable instanceof NamedRunnable) ? ((NamedRunnable) wrapRunnable).getName() : null;
+        return new DtpRunnable(command, wrapRunnable, taskName);
     }
 
     /**

core/src/main/java/org/dromara/dynamictp/core/executor/ExecutorType.java

@@ -19,6 +19,7 @@
 
 import org.dromara.dynamictp.core.executor.eager.EagerDtpExecutor;
 import lombok.Getter;
+import org.dromara.dynamictp.core.executor.priority.PriorityDtpExecutor;
 
 /**
  * ExecutorType related
@@ -30,12 +31,29 @@
 public enum ExecutorType {
 
     /**
-     * Executor type.
+     * Common executor type.
      */
     COMMON("common", DtpExecutor.class),
+
+    /**
+     * Eager executor type.
+     */
     EAGER("eager", EagerDtpExecutor.class),
+
+    /**
+     * Scheduled executor type.
+     */
     SCHEDULED("scheduled", ScheduledDtpExecutor.class),
-    ORDERED("ordered", OrderedDtpExecutor.class);
+
+    /**
+     * Ordered executor type.
+     */
+    ORDERED("ordered", OrderedDtpExecutor.class),
+
+    /**
+     * Priority executor type.
+     */
+    PRIORITY("priority", PriorityDtpExecutor.class);
 
     private final String name;
 
@@ -54,4 +72,5 @@ public static Class<?> getClass(String name) {
         }
         return COMMON.getClazz();
     }
+
 }

core/src/main/java/org/dromara/dynamictp/core/executor/priority/Priority.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.dromara.dynamictp.core.executor.priority;
+
+/**
+ * Priority related
+ *
+ * @author <a href = "mailto:[email protected]">KamTo Hung</a>
+ */
+public interface Priority {
+
+    int getPriority();
+
+}

core/src/main/java/org/dromara/dynamictp/core/executor/priority/PriorityCallable.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.dromara.dynamictp.core.executor.priority;
+
+import lombok.Getter;
+
+import java.util.concurrent.Callable;
+
+/**
+ * PriorityCallable related
+ *
+ * @author <a href = "mailto:[email protected]">KamTo Hung</a>
+ */
+public class PriorityCallable<V> implements Priority, Callable<V> {
+
+    private final Callable<V> callable;
+
+    @Getter
+    private final int priority;
+
+    private PriorityCallable(Callable<V> callable, int priority) {
+        this.callable = callable;
+        this.priority = priority;
+    }
+
+    public static <T> Callable<T> of(Callable<T> task, int i) {
+        return new PriorityCallable<>(task, i);
+    }
+
+    @Override
+    public V call() throws Exception {
+        return callable.call();
+    }
+
+}

core/src/main/java/org/dromara/dynamictp/core/executor/priority/PriorityDtpExecutor.java

@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.dromara.dynamictp.core.executor.priority;
+
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.dynamictp.core.executor.DtpExecutor;
+import org.dromara.dynamictp.core.support.task.runnable.DtpRunnable;
+
+import java.util.Comparator;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * PriorityDtpExecutor related, extending DtpExecutor, implements priority feature
+ *
+ * @author <a href = "mailto:[email protected]">KamTo Hung</a>
+ */
+@Slf4j
+public class PriorityDtpExecutor extends DtpExecutor {
+
+    /**
+     * The default priority.
+     */
+    private static final int DEFAULT_PRIORITY = 0;
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               int capacity) {
+        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new PriorityBlockingQueue<>(capacity), Executors.defaultThreadFactory(), new AbortPolicy());
+    }
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               int capacity,
+                               ThreadFactory threadFactory) {
+        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new PriorityBlockingQueue<>(capacity), threadFactory, new AbortPolicy());
+    }
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               int capacity,
+                               RejectedExecutionHandler handler) {
+        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new PriorityBlockingQueue<>(capacity), Executors.defaultThreadFactory(), handler);
+    }
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               int capacity,
+                               ThreadFactory threadFactory,
+                               RejectedExecutionHandler handler) {
+        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, new PriorityBlockingQueue<>(capacity), threadFactory, handler);
+    }
+
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               PriorityBlockingQueue<Runnable> workQueue) {
+        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), new AbortPolicy());
+    }
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               PriorityBlockingQueue<Runnable> workQueue,
+                               ThreadFactory threadFactory) {
+        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new AbortPolicy());
+    }
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               PriorityBlockingQueue<Runnable> workQueue,
+                               RejectedExecutionHandler handler) {
+        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler);
+    }
+
+    public PriorityDtpExecutor(int corePoolSize,
+                               int maximumPoolSize,
+                               long keepAliveTime,
+                               TimeUnit unit,
+                               PriorityBlockingQueue<Runnable> workQueue,
+                               ThreadFactory threadFactory,
+                               RejectedExecutionHandler handler) {
+        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
+    }
+
+
+    @Override
+    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
+        return new PriorityFutureTask<>(runnable, value);
+    }
+
+    @Override
+    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
+        return new PriorityFutureTask<>(callable);
+    }
+
+    public void execute(Runnable command, int priority) {
+        super.execute(PriorityRunnable.of(command, priority));
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+        return super.submit(PriorityRunnable.of(task, DEFAULT_PRIORITY));
+    }
+
+    public Future<?> submit(Runnable task, int priority) {
+        return super.submit(PriorityRunnable.of(task, priority));
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+        return super.submit(PriorityRunnable.of(task, DEFAULT_PRIORITY), result);
+    }
+
+    public <T> Future<T> submit(Runnable task, T result, int priority) {
+        return super.submit(PriorityRunnable.of(task, priority), result);
+    }
+
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+        return super.submit(PriorityCallable.of(task, DEFAULT_PRIORITY));
+    }
+
+    public <T> Future<T> submit(Callable<T> task, int priority) {
+        return super.submit(PriorityCallable.of(task, priority));
+    }
+
+
+    /**
+     * Priority Comparator
+     *
+     * @return Comparator
+     */
+    public static Comparator<Runnable> getRunnableComparator() {
+        return (o1, o2) -> {
+            if (!(o1 instanceof DtpRunnable) || !(o2 instanceof DtpRunnable)) {
+                return 0;
+            }
+            Runnable po1 = ((DtpRunnable) o1).getOriginRunnable();
+            Runnable po2 = ((DtpRunnable) o2).getOriginRunnable();
+            if (po1 instanceof Priority && po2 instanceof Priority) {
+                return Integer.compare(((Priority) po1).getPriority(), ((Priority) po2).getPriority());
+            }
+            return 0;
+        };
+    }
+
+}

core/src/main/java/org/dromara/dynamictp/core/executor/priority/PriorityFutureTask.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.dromara.dynamictp.core.executor.priority;
+
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.FutureTask;
+
+/**
+ * PriorityFutureTask related
+ *
+ * @author <a href = "mailto:[email protected]">KamTo Hung</a>
+ */
+public class PriorityFutureTask<V> extends FutureTask<V> implements Priority {
+
+    /**
+     * The runnable.
+     */
+    private final Priority obj;
+
+    private final int priority;
+
+    public PriorityFutureTask(Runnable runnable, V result) {
+        super(runnable, result);
+        this.obj = (PriorityRunnable) runnable;
+        this.priority = this.obj.getPriority();
+    }
+
+    public PriorityFutureTask(Callable<V> callable) {
+        super(callable);
+        this.obj = (PriorityCallable<V>) callable;
+        this.priority = this.obj.getPriority();
+    }
+
+    @Override
+    public int getPriority() {
+        return this.priority;
+    }
+
+}