-
Notifications
You must be signed in to change notification settings - Fork 766
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
[ISSUE #283] add Priority executor #386
Merged
Merged
Changes from all commits
Commits
Show all changes
12 commits
Select commit
Hold shift + click to select a range
e6b9a90
feat(priority executor): init
KamToHung b015b4f
feat(priority executor): init
KamToHung c286d7d
feat(priority executor): add Callable
KamToHung 1762754
feat(priority executor): change comparable to Object #283
KamToHung a267222
feat(priority executor): add to spring #283
KamToHung 86b5192
feat(priority executor): fix priority compare error #283
KamToHung d3bbe57
feat(priority executor): fix priority compare error #283
KamToHung 36dd332
feat(priority executor): fix priority compare error #283
KamToHung e9af9c6
feat(priority executor): fix priority compare error #283
KamToHung a51ae40
feat(priority executor): add origin command in dtpRunnable #283
KamToHung 76034e9
feat(priority executor): add license header #283
KamToHung 683b647
feat(priority executor): add ThreadPoolBuilder #283
KamToHung File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
28 changes: 28 additions & 0 deletions
28
core/src/main/java/org/dromara/dynamictp/core/executor/priority/Priority.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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(); | ||
|
||
} |
49 changes: 49 additions & 0 deletions
49
core/src/main/java/org/dromara/dynamictp/core/executor/priority/PriorityCallable.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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(); | ||
} | ||
|
||
} |
182 changes: 182 additions & 0 deletions
182
core/src/main/java/org/dromara/dynamictp/core/executor/priority/PriorityDtpExecutor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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; | ||
}; | ||
} | ||
|
||
} |
54 changes: 54 additions & 0 deletions
54
core/src/main/java/org/dromara/dynamictp/core/executor/priority/PriorityFutureTask.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -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; | ||
} | ||
|
||
} |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
这块应该用降序吧
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
类似@order,感觉用升序好点?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
也行,加点注释吧,不然容易理解不一致
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
好,已添加注释