lastExecutorService = new AtomicReference<>(null);
/**
* 该方法将会返回{@link #COMMON_POOL}，如果还未初始化则会懒加载初始化后再返回。
@@ -151,9 +186,11 @@ public class Async {
new ThreadFactory() {
private final AtomicLong threadCount = new AtomicLong(0);
+ @SuppressWarnings("NullableProblems")
@Override
public Thread newThread(Runnable r) {
- Thread t = new Thread(r, "asyncTool-commonPool-thread-" + threadCount.getAndIncrement());
+ Thread t = new Thread(r,
+ "asyncTool-commonPool-thread-" + threadCount.getAndIncrement());
t.setDaemon(true);
return t;
}
@@ -183,11 +220,12 @@ public class Async {
/**
* @param now 是否立即关闭
- * @throws IllegalStateException 如果尚未调用过{@link #getCommonPool()}，即没有使用过“使用默认线程池”的方法，该方法会抛出空指针异常。
+ * @return 如果尚未调用过{@link #getCommonPool()}，即没有初始化默认线程池，返回false。否则返回true。
*/
- public static synchronized void shutDownCommonPool(boolean now) {
+ @SuppressWarnings("unused")
+ public static synchronized boolean shutDownCommonPool(boolean now) {
if (COMMON_POOL == null) {
- throw new IllegalStateException("COMMON_POOL Not initialized yet");
+ return false;
}
if (!COMMON_POOL.isShutdown()) {
if (now) {
@@ -196,6 +234,52 @@ public class Async {
COMMON_POOL.shutdown();
}
}
+ return true;
+ }
+
+ // ========================= deprecated =========================
+
+ /**
+ * 同步执行一次任务。
+ *
+ * @return 只要执行未超时，就返回true。
+ * @deprecated 已经被 {@link #work(long, ExecutorService, Collection, String)}方法取代。
+ */
+ @Deprecated
+ public static boolean beginWork(long timeout,
+ ExecutorService executorService,
+ Collection> workerWrappers)
+ throws InterruptedException {
+ final OnceWork work = work(timeout, executorService, workerWrappers);
+ work.awaitFinish();
+ return work.hasTimeout();
+ }
+
+ /**
+ * 同步执行一次任务。
+ * 如果想自定义线程池，请传pool。不自定义的话，就走默认的COMMON_POOL
+ *
+ * @deprecated 已经被 {@link #work(long, ExecutorService, Collection, String)}方法取代。
+ */
+ @Deprecated
+ public static boolean beginWork(long timeout, ExecutorService executorService, WorkerWrapper... workerWrapper)
+ throws ExecutionException, InterruptedException {
+ if (workerWrapper == null || workerWrapper.length == 0) {
+ return false;
+ }
+ Set workerWrappers = Arrays.stream(workerWrapper).collect(Collectors.toSet());
+ //noinspection unchecked
+ return beginWork(timeout, executorService, workerWrappers);
+ }
+
+ /**
+ * 同步阻塞,直到所有都完成,或失败
+ *
+ * @deprecated 已经被 {@link #work(long, ExecutorService, Collection, String)}方法取代。
+ */
+ @Deprecated
+ public static boolean beginWork(long timeout, WorkerWrapper... workerWrapper) throws ExecutionException, InterruptedException {
+ return beginWork(timeout, getCommonPool(), workerWrapper);
}
/**
@@ -209,8 +293,9 @@ public class Async {
*/
@Deprecated
public static void shutDown() {
- if (lastExecutorService != COMMON_POOL) {
- shutDown(lastExecutorService);
+ final ExecutorService last = lastExecutorService.get();
+ if (last != COMMON_POOL) {
+ shutDown(last);
}
}
@@ -221,7 +306,7 @@ public class Async {
* @deprecated 没啥用的方法，要关闭线程池还不如直接调用线程池的关闭方法，避免歧义。
*/
@Deprecated
- public static void shutDown(ExecutorService executorService) {
+ public static void shutDown(@Nullable ExecutorService executorService) {
if (executorService != null) {
executorService.shutdown();
}
diff --git a/src/main/java/com/jd/platform/async/executor/PollingCenter.java b/asyncTool-core/src/main/java/com/jd/platform/async/executor/PollingCenter.java
similarity index 85%
rename from src/main/java/com/jd/platform/async/executor/PollingCenter.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/executor/PollingCenter.java
index 66d4cabdad615566616bf85c10b0b3229b5d9228..2dc1829fb05b28c634efd7144c44fb9c691d6c17 100644
--- a/src/main/java/com/jd/platform/async/executor/PollingCenter.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/executor/PollingCenter.java
@@ -1,9 +1,9 @@
package com.jd.platform.async.executor;
-import com.jd.platform.async.util.timer.Timeout;
-import com.jd.platform.async.util.timer.TimerTask;
-import com.jd.platform.async.util.timer.HashedWheelTimer;
-import com.jd.platform.async.util.timer.Timer;
+import com.jd.platform.async.openutil.timer.HashedWheelTimer;
+import com.jd.platform.async.openutil.timer.Timeout;
+import com.jd.platform.async.openutil.timer.Timer;
+import com.jd.platform.async.openutil.timer.TimerTask;
import com.jd.platform.async.wrapper.WorkerWrapperGroup;
import java.util.Set;
@@ -12,7 +12,7 @@ import java.util.concurrent.TimeUnit;
/**
* 检查{@link WorkerWrapperGroup}是否调用完成的轮询中心。
- *
+ * 内部使用时间轮进行轮询。
*
* ===========================================================================================
*
@@ -27,11 +27,11 @@ import java.util.concurrent.TimeUnit;
* 这是旧版本(v1.4及以前)中可能会引发线程耗尽bug的情况，在test/v15.wrappertest中示例testThreadPolling_V14Bug说明了这个bug
* 线程数：2
* A(5ms)--B1(10ms) ---|--> C1(5ms)
- * . \ | (B1、B2全部完成可执行C1、C2)
+ * . \ | (B1、B2任一完成可执行C1、C2)
* . ---> B2(20ms) --|--> C2(5ms)
*
* }
- * 线程1执行了A，然后在{@link java.util.concurrent.CompletableFuture#allOf(CompletableFuture[])}等待B1与B2执行完成。
+ * 线程1执行了A，然后在{@link CompletableFuture#allOf(CompletableFuture[])}等待B1与B2执行完成。
* 线程2执行了B1或B2中的一个，也在allOf方法等待C1、C2完成。
* 结果没有线程执行C和B2了，导致超时而死，并且这个线程池线程有可能被耗尽。
* >
@@ -65,7 +65,7 @@ public class PollingCenter {
thread.setDaemon(true);
return thread;
},
- 4,
+ 1,
TimeUnit.MILLISECONDS,
1024);
diff --git a/src/main/java/com/jd/platform/async/executor/timer/SystemClock.java b/asyncTool-core/src/main/java/com/jd/platform/async/executor/timer/SystemClock.java
similarity index 97%
rename from src/main/java/com/jd/platform/async/executor/timer/SystemClock.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/executor/timer/SystemClock.java
index e65dd8578d3318f2375747d738eddf77e7a7bf8e..6cba50a28e46c86301cb0b7e1136db091bccda8d 100644
--- a/src/main/java/com/jd/platform/async/executor/timer/SystemClock.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/executor/timer/SystemClock.java
@@ -2,7 +2,6 @@ package com.jd.platform.async.executor.timer;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
diff --git a/asyncTool-core/src/main/java/com/jd/platform/async/worker/OnceWork.java b/asyncTool-core/src/main/java/com/jd/platform/async/worker/OnceWork.java
new file mode 100644
index 0000000000000000000000000000000000000000..611638be5d03cf51dcd6b7d96569707b1878fb74
--- /dev/null
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/worker/OnceWork.java
@@ -0,0 +1,375 @@
+package com.jd.platform.async.worker;
+
+import com.jd.platform.async.executor.timer.SystemClock;
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapperGroup;
+
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * 一次工作结果的总接口。
+ *
+ * @author tcsnzh[zh.jobs@foxmail.com] create this in 2021/5/25-下午3:22
+ */
+public interface OnceWork {
+ /**
+ * 返回唯一的workId
+ */
+ String workId();
+
+ /**
+ * 判断是否结束。因超时而结束也算结束。
+ */
+ boolean isFinish();
+
+ /**
+ * 同步等待到结束。
+ */
+ void awaitFinish() throws InterruptedException;
+
+ /**
+ * 判断是否超时
+ *
+ * @return 如果尚未结束或已结束但未超时，返回false。已结束且已经超时返回true。
+ */
+ boolean hasTimeout();
+
+ /**
+ * 判断是否全部wrapper都处于 执行成功 或 跳过。
+ *
+ * @return 如果已经结束，所有wrapper都成功或跳过返回true，否则返回false。如果尚未结束，返回false。
+ */
+ default boolean allSuccess() {
+ if (!isFinish()) {
+ return false;
+ }
+ return getWrappers().values().stream().allMatch(wrapper -> {
+ final ResultState state = wrapper.getWorkResult().getResultState();
+ return state == ResultState.SUCCESS || state == ResultState.DEFAULT;
+ });
+ }
+
+ /**
+ * 获取全部参与到工作中的wrapper。
+ */
+ Map> getWrappers();
+
+ /**
+ * 获取{@link WorkResult#getResultState()}为{@link ResultState#SUCCESS}的wrapper。
+ */
+ default Map> getSuccessWrappers() {
+ return getWrappersOfState(ResultState.SUCCESS);
+ }
+
+ /**
+ * 获取状态于这些state中的wrapper。
+ *
+ * @param ofState 状态列表
+ * @return 返回Map
+ */
+ default Map> getWrappersOfState(ResultState... ofState) {
+ final HashSet states = new HashSet<>(Arrays.asList(ofState));
+ return getWrappers().entrySet().stream()
+ .filter(entry -> states.contains(entry.getValue().getWorkResult().getResultState()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ }
+
+ /**
+ * 获取启动时间
+ */
+ long getStartTime();
+
+ /**
+ * 获取结束时间
+ *
+ * @return 如果超时，返回超时的时刻。如果尚未结束，则抛出异常。
+ * @throws IllegalStateException 尚未结束，抛出异常。
+ */
+ long getFinishTime();
+
+ /**
+ * @return 已经取消完成
+ */
+ boolean isCancelled();
+
+ /**
+ * @return 是否正在取消中
+ */
+ boolean isWaitingCancel();
+
+ /**
+ * 请求异步取消。
+ */
+ void pleaseCancel();
+
+ /**
+ * 同步等待取消完成。
+ */
+ default void pleaseCancelAndAwaitFinish() throws InterruptedException {
+ if (!isCancelled() && !isWaitingCancel()) {
+ pleaseCancel();
+ }
+ awaitFinish();
+ }
+
+ /**
+ * @return 返回 {@link AsFuture}封装对象。
+ */
+ default AsFuture asFuture() {
+ return new AsFuture(this, limitTime -> limitTime / 16);
+ }
+
+ /**
+ * @param sleepCheckInterval 为防止线程爆炸，在{@link Future#get(long, TimeUnit)}方法时使用隔一段时间检查一次。
+ * 该Function的参数为总超时毫秒值，返回值为检查时间间隔。
+ * @return 返回 {@link AsFuture}封装对象。
+ */
+ default AsFuture asFuture(Function sleepCheckInterval) {
+ return new AsFuture(this, sleepCheckInterval);
+ }
+
+ // static
+
+ /**
+ * 空任务
+ */
+ static OnceWork emptyWork(String workId) {
+ return new EmptyWork(workId);
+ }
+
+ // class
+
+ class AsFuture implements Future>> {
+ private final OnceWork onceWork;
+ private final Function sleepCheckInterval;
+
+ private AsFuture(OnceWork onceWork, Function sleepCheckInterval) {
+ this.onceWork = onceWork;
+ this.sleepCheckInterval = sleepCheckInterval;
+ }
+
+ /**
+ * 同步等待取消
+ *
+ * @param ignore__mayInterruptIfRunning 该参数将被无视。因为暂未实现“修改允许打断属性”功能。 // todo 等待实现
+ */
+ @Override
+ public boolean cancel(boolean ignore__mayInterruptIfRunning) {
+ try {
+ onceWork.pleaseCancelAndAwaitFinish();
+ } catch (InterruptedException e) {
+ throw new RuntimeException("", e);
+ }
+ return true;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return onceWork.isCancelled();
+ }
+
+ @Override
+ public boolean isDone() {
+ return onceWork.isFinish();
+ }
+
+ @Override
+ public Map> get() throws InterruptedException, ExecutionException {
+ if (!onceWork.isFinish()) {
+ onceWork.awaitFinish();
+ }
+ return onceWork.getWrappers();
+ }
+
+ /**
+ * 避免线程爆炸，该方法不予单独开线程，而是单线程{@link Thread#sleep(long)}每睡一段时间检查一次。
+ */
+ @Override
+ public Map> get(long timeout,
+ @SuppressWarnings("NullableProblems") TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ final long millis = Objects.requireNonNull(unit).toMillis(timeout);
+ final long interval = Math.max(1, Math.min(millis, sleepCheckInterval.apply(millis)));
+ for (int i = 0; interval * i < millis; i++) {
+ if (onceWork.isFinish()) {
+ return onceWork.getWrappers();
+ }
+ Thread.sleep(interval);
+ }
+ throw new TimeoutException(
+ "onceWork.asFuture.get(long,TimeUnit) out of time limit(" +
+ timeout + "," + unit + ") , this is " + this);
+ }
+
+ @Override
+ public String toString() {
+ return "(asFuture from " + this + ")";
+ }
+ }
+
+ abstract class AbstractOnceWork implements OnceWork {
+ protected final String workId;
+
+ public AbstractOnceWork(String workId) {
+ this.workId = workId;
+ }
+
+ @Override
+ public String workId() {
+ return workId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof OnceWork)) {
+ return false;
+ }
+ OnceWork _o = (OnceWork) o;
+ return Objects.equals(_o.workId(), this.workId());
+ }
+
+ @Override
+ public int hashCode() {
+ return workId().hashCode();
+ }
+
+ @Override
+ public String toString() {
+ final boolean finish;
+ final StringBuilder sb = new StringBuilder(48)
+ .append(this.getClass().getSimpleName())
+ .append("{isFinish=").append(finish = isFinish())
+ .append(", hasTimeout=").append(hasTimeout())
+ .append(", allSuccess=").append(allSuccess())
+ .append(", getStartTime=").append(getStartTime())
+ .append(", isCancelled=").append(isCancelled())
+ .append(", isWaitingCancel=").append(isWaitingCancel());
+ if (finish) {
+ sb.append(", getFinishTime=").append(getFinishTime());
+ }
+ return sb
+ .append(", wrappers::getId=").append(getWrappers().keySet())
+ .append('}').toString();
+ }
+ }
+
+ class Impl extends AbstractOnceWork {
+ protected final WorkerWrapperGroup group;
+
+ public Impl(WorkerWrapperGroup group, String workId) {
+ super(workId);
+ this.group = group;
+ }
+
+ @Override
+ public boolean isFinish() {
+ return group.getEndCDL().getCount() > 0;
+ }
+
+ @Override
+ public void awaitFinish() throws InterruptedException {
+ group.getEndCDL().await();
+ }
+
+ @Override
+ public boolean hasTimeout() {
+ return group.getAnyTimeout().get();
+ }
+
+ @Override
+ public Map> getWrappers() {
+ return group.getForParamUseWrappers();
+ }
+
+ @Override
+ public long getStartTime() {
+ return group.getGroupStartTime();
+ }
+
+ @Override
+ public long getFinishTime() {
+ if (isFinish()) {
+ throw new IllegalStateException("work not finish.");
+ }
+ return group.getFinishTime();
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return group.isCancelled();
+ }
+
+ @Override
+ public boolean isWaitingCancel() {
+ return group.isWaitingCancel();
+ }
+
+ @Override
+ public void pleaseCancel() {
+ group.pleaseCancel();
+ }
+ }
+
+ class EmptyWork extends AbstractOnceWork {
+ private final long initTime = SystemClock.now();
+
+ public EmptyWork(String workId) {
+ super(workId);
+ }
+
+ @Override
+ public boolean isFinish() {
+ return true;
+ }
+
+ @Override
+ public void awaitFinish() {
+ // do nothing
+ }
+
+ @Override
+ public boolean hasTimeout() {
+ return false;
+ }
+
+ @Override
+ public Map> getWrappers() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public long getStartTime() {
+ return initTime;
+ }
+
+ @Override
+ public long getFinishTime() {
+ return initTime;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return false;
+ }
+
+ @Override
+ public boolean isWaitingCancel() {
+ return false;
+ }
+
+ @Override
+ public void pleaseCancel() {
+ }
+
+ @Override
+ public String toString() {
+ return "(it's empty work)";
+ }
+ }
+}
diff --git a/src/main/java/com/jd/platform/async/worker/ResultState.java b/asyncTool-core/src/main/java/com/jd/platform/async/worker/ResultState.java
similarity index 100%
rename from src/main/java/com/jd/platform/async/worker/ResultState.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/worker/ResultState.java
diff --git a/src/main/java/com/jd/platform/async/worker/WorkResult.java b/asyncTool-core/src/main/java/com/jd/platform/async/worker/WorkResult.java
similarity index 95%
rename from src/main/java/com/jd/platform/async/worker/WorkResult.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/worker/WorkResult.java
index 6d32659f5c495e2224bf3b0b5caedfa4ef3e7040..1cf0f04afd2082e622787631a52be0793a11afdb 100755
--- a/src/main/java/com/jd/platform/async/worker/WorkResult.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/worker/WorkResult.java
@@ -1,7 +1,5 @@
package com.jd.platform.async.worker;
-import java.util.concurrent.atomic.AtomicReference;
-
/**
* 执行结果
*/
diff --git a/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/QuickBuildWorkerWrapper.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/QuickBuildWorkerWrapper.java
new file mode 100644
index 0000000000000000000000000000000000000000..fd5aa1783de05c50219b3c073ad84ea9872e9401
--- /dev/null
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/QuickBuildWorkerWrapper.java
@@ -0,0 +1,100 @@
+package com.jd.platform.async.wrapper;
+
+import com.jd.platform.async.callback.ICallback;
+import com.jd.platform.async.callback.IWorker;
+import com.jd.platform.async.openutil.collection.DirectedGraph;
+import com.jd.platform.async.openutil.collection.Graph;
+import com.jd.platform.async.wrapper.strategy.WrapperStrategy;
+
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+/**
+ * 快速构造{@link WorkerWrapper}，少废话！
+ *
+ * 直接设置属性，不用麻烦Builder设置来设置去，
+ * 请 注 意：构造方法不会检查参数合法性，请程序员自己保证参数合法。
+ *
+ * 将关系存储于有向图中{@link DirectedGraph}以节省每个wrapper都要保存节点数据的开销。
+ *
+ *
+ * @author create by TcSnZh on 2021/5/13-上午11:54
+ */
+public class QuickBuildWorkerWrapper extends WorkerWrapper {
+ private final DirectedGraph, Object> graph;
+
+ private volatile Set> nextWrappersCache;
+ private volatile Set> dependWrappersCache;
+
+ /**
+ * 构造函数，传入所有属性
+ *
+ * @param id {@link WorkerWrapper#id}
+ * @param param {@link WorkerWrapper#param}
+ * @param worker {@link WorkerWrapper#worker}
+ * @param callback {@link WorkerWrapper#callback}
+ * @param allowInterrupt {@link WorkerWrapper#allowInterrupt}
+ * @param enableTimeout {@link WorkerWrapper#enableTimeout}
+ * @param timeoutLength {@link WorkerWrapper#timeoutLength}
+ * @param timeoutUnit {@link WorkerWrapper#timeoutLength}
+ * @param wrapperStrategy {@link WorkerWrapper#timeoutUnit}
+ * @param wrapperGraph 将节点信息保存在图中，而不是如{@link StableWorkerWrapper}在每个wrapper中都保存节点信息。
+ *
+ * {@link WorkerWrapper#getDependWrappers()}与{@link WorkerWrapper#getNextWrappers()}方法
+ * 将从本图中读取依赖顺序。除此之外，本类不会对本图进行任何修改操作。
+ * 因此，传入的此图应当保证读取时的线程安全。
+ *
+ */
+ public QuickBuildWorkerWrapper(String id,
+ T param,
+ IWorker worker,
+ ICallback callback,
+ boolean allowInterrupt,
+ boolean enableTimeout,
+ long timeoutLength,
+ TimeUnit timeoutUnit,
+ WrapperStrategy wrapperStrategy,
+ DirectedGraph, Object> wrapperGraph) {
+ super(id, worker, callback, allowInterrupt, enableTimeout, timeoutLength, timeoutUnit, wrapperStrategy);
+ graph = wrapperGraph;
+ super.param = param;
+ State.setState(state, State.BUILDING, State.INIT);
+ }
+
+ @Override
+ public Set> getNextWrappers() {
+ if (nextWrappersCache == null) {
+ synchronized (this) {
+ if (nextWrappersCache == null) {
+ nextWrappersCache = graph.getRelationFrom(this).stream()
+ .map(Graph.Entry::getTo).collect(Collectors.toSet());
+ }
+ }
+ }
+ return nextWrappersCache;
+ }
+
+ @Override
+ public Set> getDependWrappers() {
+ if (dependWrappersCache == null) {
+ synchronized (this) {
+ if (dependWrappersCache == null) {
+ dependWrappersCache = graph.getRelationTo(this).stream()
+ .map(Graph.Entry::getFrom).collect(Collectors.toSet());
+ }
+ }
+ }
+ return dependWrappersCache;
+ }
+
+ @Override
+ void setNextWrappers(Set> nextWrappers) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ void setDependWrappers(Set> dependWrappers) {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapper.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapper.java
similarity index 100%
rename from src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapper.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapper.java
diff --git a/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapperBuilder.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapperBuilder.java
similarity index 90%
rename from src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapperBuilder.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapperBuilder.java
index 5fed611f0d430ac71392e117c672e7d83ef3c3b8..5c0da4a81970f8118ec1d015b14bd126a24057c7 100644
--- a/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapperBuilder.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/StableWorkerWrapperBuilder.java
@@ -1,11 +1,12 @@
package com.jd.platform.async.wrapper;
+import com.jd.platform.async.exception.SkippedException;
import com.jd.platform.async.callback.ICallback;
import com.jd.platform.async.callback.IWorker;
import com.jd.platform.async.worker.WorkResult;
import com.jd.platform.async.wrapper.strategy.depend.DependMustStrategyMapper;
-import com.jd.platform.async.wrapper.strategy.depend.DependWrapperActionStrategy;
-import com.jd.platform.async.wrapper.strategy.depend.DependWrapperStrategyMapper;
+import com.jd.platform.async.wrapper.strategy.depend.DependOnUpWrapperStrategy;
+import com.jd.platform.async.wrapper.strategy.depend.DependOnUpWrapperStrategyMapper;
import com.jd.platform.async.wrapper.strategy.depend.DependenceStrategy;
import com.jd.platform.async.wrapper.strategy.skip.SkipStrategy;
@@ -62,15 +63,15 @@ class StableWorkerWrapperBuilder, DependWrapperActionStrategy> dependWrapperActionStrategyMap;
+ private Map, DependOnUpWrapperStrategy> dependWrapperActionStrategyMap;
/**
* 存储需要特殊对待自己的nextWrapper集合。
*/
- private Map, DependWrapperActionStrategy> selfIsSpecialMap;
+ private Map, DependOnUpWrapperStrategy> selfIsSpecialMap;
/**
* 一个保存以must=true方式传入的WorkerWrapper的集合。
*
- * 该Set将会加入到{@link WorkerWrapper.WrapperStrategy#getDependMustStrategyMapper().mustDependSet}之中
+ * 该Set将会加入到{@link WorkerWrapper.StableWrapperStrategy#getDependMustStrategyMapper().mustDependSet}之中
*/
private Set> mustDependSet;
/**
@@ -172,7 +173,7 @@ class StableWorkerWrapperBuilder wrapper) {
+ public SetDependImpl specialDependWrapper(DependOnUpWrapperStrategy strategy, WorkerWrapper wrapper) {
if (strategy == null || wrapper == null) {
return this;
}
@@ -232,7 +233,7 @@ class StableWorkerWrapperBuilder wrapper) {
+ public SetNextImpl specialToNextWrapper(DependOnUpWrapperStrategy strategy, WorkerWrapper wrapper) {
if (strategy == null || wrapper == null) {
return this;
}
@@ -330,6 +331,7 @@ class StableWorkerWrapperBuilder 0) {
@@ -341,7 +343,7 @@ class StableWorkerWrapperBuilder 0) {
- DependWrapperStrategyMapper mapper = new DependWrapperStrategyMapper();
+ DependOnUpWrapperStrategyMapper mapper = new DependOnUpWrapperStrategyMapper();
dependWrapperActionStrategyMap.forEach(mapper::putMapping);
wrapper.getWrapperStrategy().setDependWrapperStrategyMapper(mapper);
}
@@ -351,12 +353,12 @@ class StableWorkerWrapperBuilder 0) {
selfIsSpecialMap.forEach((next, strategy) -> {
- DependWrapperStrategyMapper dependWrapperStrategyMapper = next.getWrapperStrategy().getDependWrapperStrategyMapper();
- if (dependWrapperStrategyMapper == null) {
+ DependOnUpWrapperStrategyMapper dependOnUpWrapperStrategyMapper = next.getWrapperStrategy().getDependWrapperStrategyMapper();
+ if (dependOnUpWrapperStrategyMapper == null) {
next.getWrapperStrategy().setDependWrapperStrategyMapper(
- dependWrapperStrategyMapper = new DependWrapperStrategyMapper());
+ dependOnUpWrapperStrategyMapper = new DependOnUpWrapperStrategyMapper());
}
- dependWrapperStrategyMapper.putMapping(wrapper, strategy);
+ dependOnUpWrapperStrategyMapper.putMapping(wrapper, strategy);
});
}
}
@@ -381,8 +383,8 @@ class StableWorkerWrapperBuilder... wrappers) {
@@ -397,8 +399,8 @@ class StableWorkerWrapperBuilder wrapper) {
@@ -406,7 +408,7 @@ class StableWorkerWrapperBuilder wrapper, boolean isMust) {
@@ -477,7 +479,7 @@ class StableWorkerWrapperBuilder {
// ========== 固定属性 ==========
@@ -41,7 +58,7 @@ public abstract class WorkerWrapper {
/**
* 各种策略的封装类。
*/
- private final WrapperStrategy wrapperStrategy = new WrapperStrategy();
+ private final WrapperStrategy wrapperStrategy;
/**
* 是否允许被打断
*/
@@ -80,14 +97,14 @@ public abstract class WorkerWrapper {
*/
protected final AtomicReference> workResult = new AtomicReference<>(null);
-
WorkerWrapper(String id,
IWorker worker,
ICallback callback,
boolean allowInterrupt,
boolean enableTimeout,
long timeoutLength,
- TimeUnit timeoutUnit
+ TimeUnit timeoutUnit,
+ WrapperStrategy wrapperStrategy
) {
if (worker == null) {
throw new NullPointerException("async.worker is null");
@@ -103,7 +120,17 @@ public abstract class WorkerWrapper {
this.enableTimeout = enableTimeout;
this.timeoutLength = timeoutLength;
this.timeoutUnit = timeoutUnit;
+ this.wrapperStrategy = wrapperStrategy;
+ }
+ WorkerWrapper(String id,
+ IWorker worker,
+ ICallback callback,
+ boolean allowInterrupt,
+ boolean enableTimeout,
+ long timeoutLength,
+ TimeUnit timeoutUnit) {
+ this(id, worker, callback, allowInterrupt, enableTimeout, timeoutLength, timeoutUnit, new StableWrapperStrategy());
}
// ========== public ==========
@@ -141,7 +168,7 @@ public abstract class WorkerWrapper {
}
public State getState() {
- return State.of(state.get());
+ return of(state.get());
}
/**
@@ -193,7 +220,7 @@ public abstract class WorkerWrapper {
// 就在想CAS的时候，出结果了，就采用新的结果重新判断一次
continue;
}
- fastFail(true, null);
+ fastFail(true, null, false);
}
return -1L;
}
@@ -204,6 +231,16 @@ public abstract class WorkerWrapper {
} while (true);
}
+ /**
+ * 直接取消wrapper运行。
+ * 如果状态在 {@link State#states_of_beforeWorkingEnd}中，则调用 {@link #fastFail(boolean, Exception, boolean)}。
+ */
+ public void cancel() {
+ if (State.setState(state, states_of_beforeWorkingEnd, SKIP, null)) {
+ fastFail(false, new CancelSkippedException(), true);
+ }
+ }
+
public WrapperStrategy getWrapperStrategy() {
return wrapperStrategy;
}
@@ -228,37 +265,39 @@ public abstract class WorkerWrapper {
// 因为抽取成方法反而不好传参、污染类方法，所以就这么干了
final Consumer __function__callbackResult =
success -> {
+ WorkResult _workResult = getWorkResult();
try {
- callback.result(success, param, getWorkResult());
+ callback.result(success, param, _workResult);
} catch (Exception e) {
- if (State.setState(state, states_of_skipOrAfterWork, ERROR, null)) {
- fastFail(false, e);
+ if (setState(state, states_of_skipOrAfterWork, ERROR, null)) {
+ fastFail(false, e, _workResult.getEx() instanceof SkippedException);
}
}
};
- final Runnable __function__callbackResult_beginNext =
+ final Runnable __function__callbackResultOfFalse_beginNext =
() -> {
__function__callbackResult.accept(false);
beginNext(executorService, now, remainTime, group);
};
- final BiConsumer __function__fastFail_callbackResult_beginNext =
+ final BiConsumer __function__fastFail_callbackResult$false_beginNext =
(fastFail_isTimeout, fastFail_exception) -> {
- fastFail(fastFail_isTimeout, fastFail_exception);
- __function__callbackResult_beginNext.run();
+ boolean isSkip = fastFail_exception instanceof SkippedException;
+ fastFail(fastFail_isTimeout && !isSkip, fastFail_exception, isSkip);
+ __function__callbackResultOfFalse_beginNext.run();
};
final Runnable __function__doWork =
() -> {
- if (State.setState(state, STARTED, WORKING)) {
+ if (setState(state, STARTED, WORKING)) {
try {
fire(group);
} catch (Exception e) {
- if (State.setState(state, WORKING, ERROR)) {
- __function__fastFail_callbackResult_beginNext.accept(false, e);
+ if (setState(state, WORKING, ERROR)) {
+ __function__fastFail_callbackResult$false_beginNext.accept(false, e);
}
return;
}
}
- if (State.setState(state, WORKING, AFTER_WORK)) {
+ if (setState(state, WORKING, AFTER_WORK)) {
__function__callbackResult.accept(true);
beginNext(executorService, now, remainTime, group);
}
@@ -266,20 +305,25 @@ public abstract class WorkerWrapper {
// ================================================
// 开始执行
try {
- if (State.isState(state, BUILDING)) {
+ if (isState(state, BUILDING)) {
throw new IllegalStateException("wrapper can't work because state is BUILDING ! wrapper is " + this);
}
- //总的已经超时了，就快速失败，进行下一个
+ // 判断是否整组取消
+ if (group.isWaitingCancel() || group.isCancelled()) {
+ cancel();
+ return;
+ }
+ // 总的已经超时了，就快速失败，进行下一个
if (remainTime <= 0) {
- if (State.setState(state, states_of_checkTimeoutAllowStates, ERROR, null)) {
- __function__fastFail_callbackResult_beginNext.accept(true, null);
+ if (setState(state, states_of_beforeWorkingEnd, ERROR, null)) {
+ __function__fastFail_callbackResult$false_beginNext.accept(true, null);
}
return;
}
- //如果自己已经执行过了。
- //可能有多个依赖，其中的一个依赖已经执行完了，并且自己也已开始执行或执行完毕。当另一个依赖执行完毕，又进来该方法时，就不重复处理了
+ // 如果自己已经执行过了。
+ // 可能有多个依赖，其中的一个依赖已经执行完了，并且自己也已开始执行或执行完毕。当另一个依赖执行完毕，又进来该方法时，就不重复处理了
final AtomicReference oldStateRef = new AtomicReference<>(null);
- if (!State.setState(state, states_of_notWorked, STARTED, oldStateRef::set)) {
+ if (!setState(state, states_of_notWorked, STARTED, oldStateRef::set)) {
return;
}
// 如果wrapper是第一次，要调用callback.begin
@@ -288,8 +332,8 @@ public abstract class WorkerWrapper {
callback.begin();
} catch (Exception e) {
// callback.begin 发生异常
- if (State.setState(state, states_of_checkTimeoutAllowStates, ERROR, null)) {
- __function__fastFail_callbackResult_beginNext.accept(false, e);
+ if (setState(state, states_of_beforeWorkingEnd, ERROR, null)) {
+ __function__fastFail_callbackResult$false_beginNext.accept(false, e);
}
return;
}
@@ -305,22 +349,23 @@ public abstract class WorkerWrapper {
// 每个线程都需要判断是否要跳过自己，该方法可能会跳过正在工作的自己。
final WrapperStrategy wrapperStrategy = getWrapperStrategy();
if (wrapperStrategy.shouldSkip(getNextWrappers(), this, fromWrapper)) {
- if (State.setState(state, STARTED, SKIP)) {
- __function__fastFail_callbackResult_beginNext.accept(false, new SkippedException());
+ if (setState(state, STARTED, SKIP)) {
+ __function__fastFail_callbackResult$false_beginNext.accept(false, new SkippedException());
}
return;
}
// 如果是由其他wrapper调用而运行至此，则使用策略器决定自己的行为
- DependenceAction.WithProperty judge = wrapperStrategy.judgeAction(getDependWrappers(), this, fromWrapper);
+ DependenceAction.WithProperty judge =
+ wrapperStrategy.judgeAction(getDependWrappers(), this, fromWrapper);
switch (judge.getDependenceAction()) {
case TAKE_REST:
return;
case FAST_FAIL:
- if (State.setState(state, STARTED, ERROR)) {
+ if (setState(state, STARTED, ERROR)) {
// 根据FAST_FAIL.fastFailException()设置的属性值来设置fastFail方法的参数
ResultState resultState = judge.getResultState();
- __function__fastFail_callbackResult_beginNext.accept(
+ __function__fastFail_callbackResult$false_beginNext.accept(
resultState == ResultState.TIMEOUT,
judge.getFastFailException()
);
@@ -331,14 +376,16 @@ public abstract class WorkerWrapper {
return;
case JUDGE_BY_AFTER:
default:
- throw new Error("策略配置错误，不应当在WorkerWrapper中返回JUDGE_BY_AFTER或其他无效值 : this=" + this + ",fromWrapper=" + fromWrapper);
+ throw new IllegalStateException(
+ "策略配置错误，不应当在WorkerWrapper中返回JUDGE_BY_AFTER或其他无效值 : this=" + this +
+ ",fromWrapper=" + fromWrapper);
}
} catch (Exception e) {
// wrapper本身抛出了不该有的异常
- State.setState(state, states_all, ERROR, null);
+ setState(state, states_all, ERROR, null);
NotExpectedException ex = new NotExpectedException(e, this);
workResult.set(new WorkResult<>(null, ResultState.EXCEPTION, ex));
- __function__fastFail_callbackResult_beginNext.accept(false, ex);
+ __function__fastFail_callbackResult$false_beginNext.accept(false, ex);
}
}
@@ -360,17 +407,16 @@ public abstract class WorkerWrapper {
} finally {
doWorkingThread.set(null);
}
-
}
/**
* 快速失败。
* 该方法不负责检查状态，请自行控制。
*
- * @param timeout 是否是因为超时而快速失败
- * @param e 设置异常信息到{@link WorkResult#getEx()}
+ * @param isTimeout 是否是因为超时而快速失败
+ * @param e 设置异常信息到{@link WorkResult#getEx()}
*/
- protected void fastFail(boolean timeout, Exception e) {
+ protected void fastFail(boolean isTimeout, Exception e, boolean isSkip) {
// 试图打断正在执行{@link IWorker#action(Object, Map)}的线程
Thread _doWorkingThread;
if ((_doWorkingThread = doWorkingThread.get()) != null
@@ -381,7 +427,7 @@ public abstract class WorkerWrapper {
// 尚未处理过结果则设置
workResult.compareAndSet(null, new WorkResult<>(
worker.defaultValue(),
- timeout ? ResultState.TIMEOUT : ResultState.EXCEPTION,
+ isTimeout ? ResultState.TIMEOUT : (isSkip ? ResultState.DEFAULT : ResultState.EXCEPTION),
e
));
}
@@ -406,7 +452,7 @@ public abstract class WorkerWrapper {
try {
next = nextWrappers.stream().findFirst().get();
group.addWrapper(next);
- State.setState(state, AFTER_WORK, SUCCESS);
+ setState(state, AFTER_WORK, SUCCESS);
} finally {
PollingCenter.getInstance().checkGroup(group.new CheckFinishTask());
if (next != null) {
@@ -421,7 +467,7 @@ public abstract class WorkerWrapper {
nextWrappers.forEach(next -> executorService.submit(() ->
next.work(executorService, this, nextRemainTIme, group))
);
- State.setState(state, AFTER_WORK, SUCCESS);
+ setState(state, AFTER_WORK, SUCCESS);
} finally {
PollingCenter.getInstance().checkGroup(group.new CheckFinishTask());
}
@@ -429,8 +475,6 @@ public abstract class WorkerWrapper {
}
- // ========== private ==========
-
// ========== hashcode and equals ==========
@Override
@@ -460,6 +504,7 @@ public abstract class WorkerWrapper {
/**
* @deprecated 建议使用 {@link #builder()}返回{@link WorkerWrapperBuilder}接口，以调用v1.5之后的规范api
*/
+ @SuppressWarnings("DeprecatedIsStillUsed")
@Deprecated
public Builder() {
}
@@ -477,7 +522,7 @@ public abstract class WorkerWrapper {
public String toString() {
final StringBuilder sb = new StringBuilder(400)
.append("WorkerWrapper{id=").append(id)
- .append(", state=").append(State.of(state.get()))
+ .append(", state=").append(of(state.get()))
.append(", param=").append(param)
.append(", workResult=").append(workResult)
.append(", allowInterrupt=").append(allowInterrupt)
@@ -507,99 +552,54 @@ public abstract class WorkerWrapper {
return sb.toString();
}
- public static class WrapperStrategy implements DependenceStrategy, SkipStrategy {
-
- // ========== 这三个策略器用于链式判断是否要开始工作 ==========
-
- // 从前往后依次判断的顺序为 dependWrapperStrategyMapper -> dependMustStrategyMapper -> dependenceStrategy
-
- /**
- * 对特殊Wrapper专用的依赖响应策略。
- * 该值允许为null
- */
- private DependWrapperStrategyMapper dependWrapperStrategyMapper;
- /**
- * 对必须完成的（must的）Wrapper的依赖响应策略。
- * 该值允许为null
- *
- * 这是一个不得不向历史妥协的属性。用于适配must开关方式。
- */
+ /**
+ * 一个通用的策略器实现类，提供了修改的功能。并兼容之前的代码。
+ */
+ public static class StableWrapperStrategy extends WrapperStrategy.AbstractWrapperStrategy {
+ private DependOnUpWrapperStrategyMapper dependOnUpWrapperStrategyMapper;
private DependMustStrategyMapper dependMustStrategyMapper;
- /**
- * 底层全局策略。
- */
private DependenceStrategy dependenceStrategy;
+ private SkipStrategy skipStrategy;
@Override
- public DependenceAction.WithProperty judgeAction(Set> dependWrappers,
- WorkerWrapper thisWrapper,
- WorkerWrapper fromWrapper) {
- // 如果存在依赖，则调用三层依赖响应策略进行判断
- DependenceStrategy strategy = dependWrapperStrategyMapper;
- if (dependMustStrategyMapper != null) {
- strategy = strategy == null ? dependMustStrategyMapper : strategy.thenJudge(dependMustStrategyMapper);
- }
- if (dependenceStrategy != null) {
- strategy = strategy == null ? dependenceStrategy : strategy.thenJudge(dependenceStrategy);
- }
- if (strategy == null) {
- throw new IllegalStateException("配置无效，三层判断策略均为null，请开发者检查自己的Builder是否逻辑错误！");
- }
- return strategy.judgeAction(dependWrappers, thisWrapper, fromWrapper);
- }
-
- public DependWrapperStrategyMapper getDependWrapperStrategyMapper() {
- return dependWrapperStrategyMapper;
+ public DependOnUpWrapperStrategyMapper getDependWrapperStrategyMapper() {
+ return dependOnUpWrapperStrategyMapper;
}
- public void setDependWrapperStrategyMapper(DependWrapperStrategyMapper dependWrapperStrategyMapper) {
- this.dependWrapperStrategyMapper = dependWrapperStrategyMapper;
+ @Override
+ public void setDependWrapperStrategyMapper(DependOnUpWrapperStrategyMapper dependOnUpWrapperStrategyMapper) {
+ this.dependOnUpWrapperStrategyMapper = dependOnUpWrapperStrategyMapper;
}
+ @Override
public DependMustStrategyMapper getDependMustStrategyMapper() {
return dependMustStrategyMapper;
}
+ @Override
public void setDependMustStrategyMapper(DependMustStrategyMapper dependMustStrategyMapper) {
this.dependMustStrategyMapper = dependMustStrategyMapper;
}
+ @Override
public DependenceStrategy getDependenceStrategy() {
return dependenceStrategy;
}
+ @Override
public void setDependenceStrategy(DependenceStrategy dependenceStrategy) {
this.dependenceStrategy = dependenceStrategy;
}
- // ========== 跳过策略 ==========
-
- private SkipStrategy skipStrategy;
-
@Override
- public boolean shouldSkip(Set> nextWrappers, WorkerWrapper thisWrapper, WorkerWrapper fromWrapper) {
- return skipStrategy != null && skipStrategy.shouldSkip(nextWrappers, thisWrapper, fromWrapper);
- }
-
public SkipStrategy getSkipStrategy() {
return skipStrategy;
}
+ @Override
public void setSkipStrategy(SkipStrategy skipStrategy) {
this.skipStrategy = skipStrategy;
}
-
- // ========== toString ==========
-
- @Override
- public String toString() {
- return "WrapperStrategy{" +
- "dependWrapperStrategyMapper=" + dependWrapperStrategyMapper +
- ", dependMustStrategyMapper=" + dependMustStrategyMapper +
- ", dependenceStrategy=" + dependenceStrategy +
- ", skipStrategy=" + skipStrategy +
- '}';
- }
}
/**
@@ -660,7 +660,7 @@ public abstract class WorkerWrapper {
static final State[] states_of_skipOrAfterWork = new State[]{SKIP, AFTER_WORK};
- static final State[] states_of_checkTimeoutAllowStates = new State[]{INIT, STARTED, WORKING};
+ static final State[] states_of_beforeWorkingEnd = new State[]{INIT, STARTED, WORKING};
static final State[] states_all = new State[]{BUILDING, INIT, STARTED, WORKING, AFTER_WORK, SUCCESS, ERROR, SKIP};
@@ -732,6 +732,7 @@ public abstract class WorkerWrapper {
*
* @param excepts 范围。
*/
+ @SuppressWarnings("unused")
static boolean inStates(AtomicInteger state, State... excepts) {
int current;
boolean inExcepts;
@@ -753,7 +754,7 @@ public abstract class WorkerWrapper {
/**
* CAS的判断是否是某个状态
*/
- static boolean isState(AtomicInteger state, State except) {
+ static boolean isState(AtomicInteger state, @SuppressWarnings("SameParameterValue") State except) {
return state.compareAndSet(except.id, except.id);
}
diff --git a/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperBuilder.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperBuilder.java
similarity index 93%
rename from src/main/java/com/jd/platform/async/wrapper/WorkerWrapperBuilder.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperBuilder.java
index 7a1f909971d94339a19c2d9d6ba10b536be7fe0e..e5307d9b1ba6e7e1d17f908aaec55d3bfc8281cc 100644
--- a/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperBuilder.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperBuilder.java
@@ -2,7 +2,7 @@ package com.jd.platform.async.wrapper;
import com.jd.platform.async.callback.ICallback;
import com.jd.platform.async.callback.IWorker;
-import com.jd.platform.async.wrapper.strategy.depend.DependWrapperActionStrategy;
+import com.jd.platform.async.wrapper.strategy.depend.DependOnUpWrapperStrategy;
import com.jd.platform.async.wrapper.strategy.depend.DependenceStrategy;
import com.jd.platform.async.wrapper.strategy.skip.SkipStrategy;
@@ -112,9 +112,11 @@ public interface WorkerWrapperBuilder {
* @param wrapper 需要设置特殊策略的Wrapper。
* @param strategy 特殊策略。
*/
- SetDepend specialDependWrapper(DependWrapperActionStrategy strategy, WorkerWrapper wrapper);
+ @SuppressWarnings("UnusedReturnValue")
+ SetDepend specialDependWrapper(DependOnUpWrapperStrategy strategy, WorkerWrapper wrapper);
- default SetDepend specialDependWrapper(DependWrapperActionStrategy strategy, WorkerWrapper... wrappers) {
+ @SuppressWarnings("unused")
+ default SetDepend specialDependWrapper(DependOnUpWrapperStrategy strategy, WorkerWrapper... wrappers) {
if (strategy == null || wrappers == null) {
return this;
}
@@ -156,6 +158,7 @@ public interface WorkerWrapperBuilder {
return setDepend().wrapper(wrappers).end();
}
+ @SuppressWarnings("unused")
default WorkerWrapperBuilder depends(Collection wrappers) {
return setDepend().wrapper(wrappers).end();
}
@@ -164,6 +167,7 @@ public interface WorkerWrapperBuilder {
return setDepend().wrapper(wrappers).strategy(strategy).end();
}
+ @SuppressWarnings("unused")
default WorkerWrapperBuilder depends(DependenceStrategy strategy, Collection wrappers) {
return setDepend().wrapper(wrappers).strategy(strategy).end();
}
@@ -204,6 +208,7 @@ public interface WorkerWrapperBuilder {
*/
SetNext mustToNextWrapper(WorkerWrapper wrapper);
+ @SuppressWarnings("unused")
default SetNext requireToNextWrapper(WorkerWrapper wrapper, boolean must) {
return must ? mustToNextWrapper(wrapper) : wrapper(wrapper);
}
@@ -215,7 +220,7 @@ public interface WorkerWrapperBuilder {
* @param wrapper 依赖本Wrapper的下游Wrapper。
* @return 返回Builder自身。
*/
- SetNext specialToNextWrapper(DependWrapperActionStrategy strategy, WorkerWrapper wrapper);
+ SetNext specialToNextWrapper(DependOnUpWrapperStrategy strategy, WorkerWrapper wrapper);
WorkerWrapperBuilder end();
}
@@ -229,6 +234,7 @@ public interface WorkerWrapperBuilder {
return setNext().wrapper(wrappers).end();
}
+ @SuppressWarnings("unused")
default WorkerWrapperBuilder nextOf(Collection wrappers) {
return setNext().wrapper(wrappers).end();
}
diff --git a/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperGroup.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperGroup.java
similarity index 57%
rename from src/main/java/com/jd/platform/async/wrapper/WorkerWrapperGroup.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperGroup.java
index 528d64607098c8cff15ef76ceec82a8c63cd5d70..5ae630bd64f875fc16d36b44db5337b5765d0b90 100644
--- a/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperGroup.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/WorkerWrapperGroup.java
@@ -1,9 +1,7 @@
package com.jd.platform.async.wrapper;
+import com.jd.platform.async.exception.CancelSkippedException;
import com.jd.platform.async.executor.PollingCenter;
-import com.jd.platform.async.util.timer.Timeout;
-import com.jd.platform.async.util.timer.TimerTask;
-import com.jd.platform.async.worker.ResultState;
import java.util.Collection;
import java.util.Map;
@@ -11,9 +9,13 @@ import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;
+import com.jd.platform.async.executor.timer.SystemClock;
+import com.jd.platform.async.openutil.timer.*;
+
/**
* @author create by TcSnZh on 2021/5/9-下午7:21
*/
@@ -36,8 +38,23 @@ public class WorkerWrapperGroup {
* 当全部wrapper都调用结束，它会countDown
*/
private final CountDownLatch endCDL = new CountDownLatch(1);
-
+ /**
+ * 检测到超时，此标记变量将为true。
+ */
private final AtomicBoolean anyTimeout = new AtomicBoolean(false);
+ /**
+ * 结束时间
+ */
+ private volatile long finishTime = -1L;
+ /**
+ * 取消任务状态
+ * 0 - not cancel , 1 - waiting cancel , 2 - already cancel
+ */
+ private final AtomicInteger cancelState = new AtomicInteger();
+
+ public static final int NOT_CANCEL = 0;
+ public static final int WAITING_CANCEL = 1;
+ public static final int ALREADY_CANCEL = 2;
public WorkerWrapperGroup(long groupStartTime, long timeoutLength) {
this.groupStartTime = groupStartTime;
@@ -64,18 +81,38 @@ public class WorkerWrapperGroup {
return forParamUseWrappers;
}
- /**
- * 同步等待这组wrapper执行完成
- *
- * @return false代表有wrapper超时了。true代表全部wrapper没有超时。
- */
- public boolean awaitFinish() throws InterruptedException {
- endCDL.await();
- return !anyTimeout.get();
+ public CountDownLatch getEndCDL() {
+ return endCDL;
+ }
+
+ public long getGroupStartTime() {
+ return groupStartTime;
+ }
+
+ public AtomicBoolean getAnyTimeout() {
+ return anyTimeout;
+ }
+
+ public long getFinishTime() {
+ return finishTime;
+ }
+
+ public boolean isCancelled() {
+ return cancelState.get() == ALREADY_CANCEL;
+ }
+
+ public boolean isWaitingCancel() {
+ return cancelState.get() == WAITING_CANCEL;
+ }
+
+ @SuppressWarnings("UnusedReturnValue")
+ public boolean pleaseCancel() {
+ return cancelState.compareAndSet(NOT_CANCEL, WAITING_CANCEL);
}
public class CheckFinishTask implements TimerTask {
+ @SuppressWarnings("RedundantThrows")
@Override
public void run(Timeout timeout) throws Exception {
// 已经完成了
@@ -84,11 +121,19 @@ public class WorkerWrapperGroup {
}
AtomicBoolean hasTimeout = new AtomicBoolean(false);
// 记录正在运行中的wrapper里，最近的限时时间。
- AtomicLong minDaley = new AtomicLong(Long.MAX_VALUE);
+ final AtomicLong minDaley = new AtomicLong(Long.MAX_VALUE);
final Collection> values = forParamUseWrappers.values();
- final Stream> stream = values.size() > 1024 ? values.parallelStream() : values.stream();
- boolean allFinish = stream
- // 处理超时
+ final Stream> stream = values.size() > 128 ? values.parallelStream() : values.stream();
+ final boolean needCancel = cancelState.get() == WAITING_CANCEL;
+ boolean allFinish_and_notNeedCancel = stream
+ // 需要取消的话就取消
+ .peek(wrapper -> {
+ if (needCancel) {
+ wrapper.cancel();
+ }
+ })
+ // 检查超时并保存最近一次限时时间
+ // 当需要取消时，才会不断遍历。如果不需要取消，则计算一次(或并行流中多次)就因allMatch不满足而退出了。
.peek(wrapper -> {
// time_diff :
// -1 -> already timeout ;
@@ -101,27 +146,39 @@ public class WorkerWrapperGroup {
if (time_diff == 0) {
return;
}
+ // use CAS and SPIN for thread safety in parallelStream .
do {
long getMinDaley = minDaley.get();
- if (getMinDaley <= time_diff || minDaley.compareAndSet(getMinDaley, time_diff)) {
- return;
+ // 需要设置最小时间，但是cas失败，则自旋
+ if (getMinDaley <= time_diff && !minDaley.compareAndSet(getMinDaley, time_diff)) {
+ continue;
}
+ return;
} while (true);
})
- // 判断是否结束，这里如果还有未结束的wrapper则会提前结束流。
- .allMatch(wrapper -> wrapper.getState().finished());
+ // 判断是否不需要取消且全部结束
+ // 在不需要取消时，这里如果还有未结束的wrapper则会提前结束流并返回false
+ // 在需要取消时，会全部遍历一遍并取消掉已经进入链路的wrapper
+ .allMatch(wrapper -> !needCancel && wrapper.getState().finished());
long getMinDaley = minDaley.get();
+ // 如果本次取消掉了任务，或是所有wrapper都已经完成
+ // ( ps : 前后两条件在这里是必定 一真一假 或 两者全假 )
+ if (needCancel || allFinish_and_notNeedCancel) {
+ // 如果这次进行了取消，则设置取消状态为已完成
+ if (needCancel) {
+ cancelState.set(ALREADY_CANCEL);
+ }
+ anyTimeout.set(hasTimeout.get());
+ finishTime = SystemClock.now();
+ endCDL.countDown();
+ }
// 如果有正在运行的wrapper
- if (!allFinish) {
+ else {
// 如果有正在WORKING的wrapper，则计算一下限时时间，限时完成后轮询它。
if (getMinDaley != Long.MAX_VALUE) {
PollingCenter.getInstance().checkGroup(this, getMinDaley);
}
}
- if (allFinish) {
- anyTimeout.set(hasTimeout.get());
- endCDL.countDown();
- }
}
// hashCode and equals will called WorkerWrapperGroup.this
@@ -142,7 +199,7 @@ public class WorkerWrapperGroup {
if (obj == this) {
return true;
}
- if (!(obj instanceof WorkerWrapperGroup.CheckFinishTask)) {
+ if (!(obj instanceof CheckFinishTask)) {
return false;
}
return Objects.equals(WorkerWrapperGroup.this, ((CheckFinishTask) obj).getParent());
diff --git a/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/WrapperStrategy.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/WrapperStrategy.java
new file mode 100644
index 0000000000000000000000000000000000000000..044c8d9cc2883298b7f494ee1af0eab37bae155c
--- /dev/null
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/WrapperStrategy.java
@@ -0,0 +1,135 @@
+package com.jd.platform.async.wrapper.strategy;
+
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.strategy.depend.DependMustStrategyMapper;
+import com.jd.platform.async.wrapper.strategy.depend.DependOnUpWrapperStrategyMapper;
+import com.jd.platform.async.wrapper.strategy.depend.DependenceAction;
+import com.jd.platform.async.wrapper.strategy.depend.DependenceStrategy;
+import com.jd.platform.async.wrapper.strategy.skip.SkipStrategy;
+
+import java.util.Set;
+
+/**
+ * @author create by TcSnZh on 2021/5/17-下午6:23
+ */
+public interface WrapperStrategy extends DependenceStrategy, SkipStrategy {
+ // ========== 这三个策略器用于链式判断是否要开始工作 ==========
+
+ // 从前往后依次判断的顺序为 dependWrapperStrategyMapper -> dependMustStrategyMapper -> dependenceStrategy
+
+ /**
+ * 设置对特殊Wrapper专用的依赖响应策略。
+ *
+ * @return 该值允许为null
+ */
+ DependOnUpWrapperStrategyMapper getDependWrapperStrategyMapper();
+
+ /**
+ * 对必须完成的（must的）Wrapper的依赖响应策略。
+ * 这是一个不得不向历史妥协的属性。用于适配must开关方式。
+ *
+ * @return 该值允许为null
+ * @deprecated 不推荐使用，很有可能被遗弃
+ */
+ @SuppressWarnings("DeprecatedIsStillUsed")
+ @Deprecated
+ default DependMustStrategyMapper getDependMustStrategyMapper() {
+ return null;
+ }
+
+ /**
+ * 底层全局策略。
+ *
+ * @return 该值不允许为null
+ */
+ DependenceStrategy getDependenceStrategy();
+
+ // ========== 这是跳过策略 ==========
+
+ /**
+ * 跳过策略
+ *
+ * @return 不允许为null
+ */
+ SkipStrategy getSkipStrategy();
+
+ @Override
+ default DependenceAction.WithProperty judgeAction(Set> dependWrappers,
+ WorkerWrapper thisWrapper,
+ WorkerWrapper fromWrapper) {
+ // 如果存在依赖，则调用三层依赖响应策略进行判断
+
+ DependenceStrategy strategy = getDependWrapperStrategyMapper();
+ if (getDependMustStrategyMapper() != null) {
+ strategy = strategy == null ? getDependMustStrategyMapper() : strategy.thenJudge(getDependenceStrategy());
+ }
+ if (getDependenceStrategy() != null) {
+ strategy = strategy == null ? getDependenceStrategy() : strategy.thenJudge(getDependenceStrategy());
+ }
+ if (strategy == null) {
+ throw new IllegalStateException("配置无效，三层判断策略均为null，请开发者检查自己的Builder是否逻辑错误！");
+ }
+ return strategy.judgeAction(dependWrappers, thisWrapper, fromWrapper);
+ }
+
+ @Override
+ default boolean shouldSkip(Set> nextWrappers, WorkerWrapper thisWrapper, WorkerWrapper fromWrapper) {
+ return getSkipStrategy() != null && getSkipStrategy().shouldSkip(nextWrappers, thisWrapper, fromWrapper);
+ }
+
+ default void setDependWrapperStrategyMapper(DependOnUpWrapperStrategyMapper dependOnUpWrapperStrategyMapper) {
+ throw new UnsupportedOperationException();
+ }
+
+ default void setDependMustStrategyMapper(DependMustStrategyMapper dependMustStrategyMapper) {
+ throw new UnsupportedOperationException();
+ }
+
+ default void setDependenceStrategy(DependenceStrategy dependenceStrategy) {
+ throw new UnsupportedOperationException();
+ }
+
+ default void setSkipStrategy(SkipStrategy skipStrategy) {
+ throw new UnsupportedOperationException();
+ }
+
+ /**
+ * 抽象策略器，实现了toString
+ */
+ abstract class AbstractWrapperStrategy implements WrapperStrategy {
+ @Override
+ public String toString() {
+ return "WrapperStrategy{" +
+ "dependWrapperStrategyMapper=" + getDependWrapperStrategyMapper() +
+ ", dependMustStrategyMapper=" + getDependMustStrategyMapper() +
+ ", dependenceStrategy=" + getDependenceStrategy() +
+ ", skipStrategy=" + getSkipStrategy() +
+ '}';
+ }
+ }
+
+ /**
+ * 默认策略器，用默认值实现了所有属性。
+ */
+ class DefaultWrapperStrategy extends AbstractWrapperStrategy {
+ @Override
+ public DependOnUpWrapperStrategyMapper getDependWrapperStrategyMapper() {
+ return null;
+ }
+
+ @Override
+ public DependMustStrategyMapper getDependMustStrategyMapper() {
+ return null;
+ }
+
+ @Override
+ public DependenceStrategy getDependenceStrategy() {
+ return DependenceStrategy.ALL_DEPENDENCIES_ALL_SUCCESS;
+ }
+
+ @Override
+ public SkipStrategy getSkipStrategy() {
+ return SkipStrategy.CHECK_ONE_LEVEL;
+ }
+ }
+}
diff --git a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependMustStrategyMapper.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependMustStrategyMapper.java
similarity index 97%
rename from src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependMustStrategyMapper.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependMustStrategyMapper.java
index ff1108a7ab1c29eb17887e90864d1c32834ed46d..b4b1a2407a857e5e68c26dbdba6ab1e073e87b18 100644
--- a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependMustStrategyMapper.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependMustStrategyMapper.java
@@ -26,7 +26,7 @@ public class DependMustStrategyMapper implements DependenceStrategy {
* 如果所有的Wrapper已经完成，本Wrapper将会开始工作。
*
* 如果任一{@link #mustDependSet}中的Wrapper失败，则返回{@link DependenceAction#FAST_FAIL}。
- * 具体超时/异常则根据{@link com.jd.platform.async.worker.ResultState}的值进行判断。
+ * 具体超时/异常则根据{@link ResultState}的值进行判断。
*
* 如果存在Wrapper未完成 且 所有的Wrapper都未失败，则返回{@link DependenceAction#JUDGE_BY_AFTER}。
*
diff --git a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependWrapperActionStrategy.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependOnUpWrapperStrategy.java
similarity index 87%
rename from src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependWrapperActionStrategy.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependOnUpWrapperStrategy.java
index fa5cef26c6c4d004acc7e766a37f5da21d60c771..a5f0d038e70a2cfcfddf1ed93b7d8c3295a0ef39 100644
--- a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependWrapperActionStrategy.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependOnUpWrapperStrategy.java
@@ -3,14 +3,14 @@ package com.jd.platform.async.wrapper.strategy.depend;
import com.jd.platform.async.wrapper.WorkerWrapper;
/**
- * 单参数策略。
+ * 由上游wrapper决定本wrapper行为的单参数策略。
*
* @author create by TcSnZh on 2021/5/1-下午11:16
*/
@FunctionalInterface
-public interface DependWrapperActionStrategy {
+public interface DependOnUpWrapperStrategy {
/**
- * 仅使用一个参数的判断方法
+ * 仅使用一个参数（即调用自身的上游wrapper）的判断方法
*
* @param fromWrapper 调用本Wrapper的上游Wrapper
* @return 返回 {@link DependenceAction.WithProperty}
@@ -24,7 +24,7 @@ public interface DependWrapperActionStrategy {
* 未运行时，休息。
* 失败时，失败。
*/
- DependWrapperActionStrategy SUCCESS_CONTINUE = new DependWrapperActionStrategy() {
+ DependOnUpWrapperStrategy SUCCESS_CONTINUE = new DependOnUpWrapperStrategy() {
@Override
public DependenceAction.WithProperty judge(WorkerWrapper ww) {
switch (ww.getWorkResult().getResultState()) {
@@ -50,7 +50,7 @@ public interface DependWrapperActionStrategy {
* 未运行时，交给下一个策略器判断。
* 失败时，失败。
*/
- DependWrapperActionStrategy SUCCESS_START_INIT_CONTINUE = new DependWrapperActionStrategy() {
+ DependOnUpWrapperStrategy SUCCESS_START_INIT_CONTINUE = new DependOnUpWrapperStrategy() {
@Override
public DependenceAction.WithProperty judge(WorkerWrapper ww) {
switch (ww.getWorkResult().getResultState()) {
diff --git a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependWrapperStrategyMapper.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependOnUpWrapperStrategyMapper.java
similarity index 77%
rename from src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependWrapperStrategyMapper.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependOnUpWrapperStrategyMapper.java
index 802e1c29857c40d368f510afba5473939d7caf45..476aa055eb7a134334d2a51530d79eab4ec11763 100644
--- a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependWrapperStrategyMapper.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependOnUpWrapperStrategyMapper.java
@@ -10,12 +10,12 @@ import java.util.stream.Collectors;
/**
* 对不同的{@link WorkerWrapper}调用者实行个性化依赖响应策略。
*
- * 使用{@link DependWrapperStrategyMapper}本实现类对{@link DependenceStrategy}进行增强，
+ * 使用{@link DependOnUpWrapperStrategyMapper}本实现类对{@link DependenceStrategy}进行增强，
*
* @author create by TcSnZh on 2021/5/1-下午11:12
*/
-public class DependWrapperStrategyMapper implements DependenceStrategy {
- private final Map, DependWrapperActionStrategy> mapper = new ConcurrentHashMap<>(4);
+public class DependOnUpWrapperStrategyMapper implements DependenceStrategy {
+ private final Map, DependOnUpWrapperStrategy> mapper = new ConcurrentHashMap<>(4);
/**
* 设置对应策略
@@ -24,7 +24,8 @@ public class DependWrapperStrategyMapper implements DependenceStrategy {
* @param strategy 要设置的策略
* @return 返回this，链式调用。
*/
- public DependWrapperStrategyMapper putMapping(WorkerWrapper targetWrapper, DependWrapperActionStrategy strategy) {
+ @SuppressWarnings("UnusedReturnValue")
+ public DependOnUpWrapperStrategyMapper putMapping(WorkerWrapper targetWrapper, DependOnUpWrapperStrategy strategy) {
mapper.put(targetWrapper, strategy);
toStringCache = null;
return this;
@@ -33,7 +34,7 @@ public class DependWrapperStrategyMapper implements DependenceStrategy {
/**
* 判断方法。
*
- * 如果fromWrapper在{@link #mapper}中，则返回{@link DependWrapperActionStrategy}的判断返回值。否则返回{@link DependenceAction#JUDGE_BY_AFTER}
+ * 如果fromWrapper在{@link #mapper}中，则返回{@link DependOnUpWrapperStrategy}的判断返回值。否则返回{@link DependenceAction#JUDGE_BY_AFTER}
*
* @param dependWrappers （这里不会使用该值）thisWrapper.dependWrappers的属性值。
* @param thisWrapper （这里不会使用该值）thisWrapper，即为“被催促”的WorkerWrapper
@@ -44,7 +45,7 @@ public class DependWrapperStrategyMapper implements DependenceStrategy {
public DependenceAction.WithProperty judgeAction(Set> dependWrappers,
WorkerWrapper thisWrapper,
WorkerWrapper fromWrapper) {
- DependWrapperActionStrategy strategy = mapper.get(fromWrapper);
+ DependOnUpWrapperStrategy strategy = mapper.get(fromWrapper);
if (strategy == null) {
return DependenceAction.JUDGE_BY_AFTER.emptyProperty();
}
diff --git a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceAction.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceAction.java
similarity index 100%
rename from src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceAction.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceAction.java
diff --git a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceStrategy.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceStrategy.java
similarity index 95%
rename from src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceStrategy.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceStrategy.java
index be000424f0b492fdae42c63003d159f14f059a25..a075369635bf3e67fbe06958b6338338c4056b4f 100644
--- a/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceStrategy.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/depend/DependenceStrategy.java
@@ -1,10 +1,13 @@
package com.jd.platform.async.wrapper.strategy.depend;
+import com.jd.platform.async.exception.SkippedException;
import com.jd.platform.async.worker.ResultState;
import com.jd.platform.async.worker.WorkResult;
import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapperGroup;
-import java.util.*;
+import java.util.Collections;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
@@ -35,7 +38,7 @@ public interface DependenceStrategy {
* @param fromWrapper 调用来源Wrapper。
*
* 该参数不会为null。
- * 因为在{@link WorkerWrapper#work(ExecutorService, long, Map, WrapperEndingInspector)}方法中传入的的第一批无依赖的Wrapper，
+ * 因为在{@link WorkerWrapper#work(ExecutorService, long, WorkerWrapperGroup)}方法中传入的的第一批无依赖的Wrapper，
* 不会被该策略器所判断，而是不论如何直接执行。
*
* @return 返回枚举值内部类，WorkerWrapper将会根据其值来决定自己如何响应这次调用。 {@link DependenceAction.WithProperty}
@@ -135,6 +138,7 @@ public interface DependenceStrategy {
case EXCEPTION:
resultState = !hasFailed ? workResult.getResultState() : resultState;
fastFailException = !hasFailed ? workResult.getEx() : fastFailException;
+ // 跳过不算失败
hasFailed = true;
break;
default:
@@ -154,8 +158,10 @@ public interface DependenceStrategy {
};
/**
- * 如果被依赖的工作中任一失败，则立即失败。否则就开始工作（不论之前的工作有没有开始）。
+ * 如果被依赖的工作中任一失败，则立即失败。
+ * 否则就开始工作（不论之前的工作有没有开始）。
*/
+ @SuppressWarnings("unused")
DependenceStrategy ALL_DEPENDENCIES_NONE_FAILED = new DependenceStrategy() {
@Override
public DependenceAction.WithProperty judgeAction(Set> dependWrappers,
@@ -166,7 +172,7 @@ public interface DependenceStrategy {
switch (workResult.getResultState()) {
case TIMEOUT:
case EXCEPTION:
- return DependenceAction.FAST_FAIL.fastFailException(workResult.getResultState(), workResult.getEx());
+ return DependenceAction.FAST_FAIL.fastFailException(workResult.getResultState(), workResult.getEx());
default:
}
}
@@ -187,6 +193,7 @@ public interface DependenceStrategy {
* @param theseWrapper 该方法唯一有效参数。
* @return 返回生成的 {@link DependenceAction.WithProperty)
*/
+ @SuppressWarnings("unused")
static DependenceStrategy theseWrapperAllSuccess(Set> theseWrapper) {
return new DependenceStrategy() {
private final Set> theseWrappers;
@@ -239,6 +246,7 @@ public interface DependenceStrategy {
*
* @deprecated 不推荐使用must开关
*/
+ @SuppressWarnings("DeprecatedIsStillUsed")
@Deprecated
DependenceStrategy IF_MUST_SET_NOT_EMPTY_ALL_SUCCESS_ELSE_ANY = new DependenceStrategy() {
@Override
diff --git a/src/main/java/com/jd/platform/async/wrapper/strategy/skip/SkipStrategy.java b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/skip/SkipStrategy.java
similarity index 98%
rename from src/main/java/com/jd/platform/async/wrapper/strategy/skip/SkipStrategy.java
rename to asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/skip/SkipStrategy.java
index 13bd5a137d45bb30591901e7e1275064065861b7..1b4f981d01ffb09e0bd2242b32372f0d66d7c090 100644
--- a/src/main/java/com/jd/platform/async/wrapper/strategy/skip/SkipStrategy.java
+++ b/asyncTool-core/src/main/java/com/jd/platform/async/wrapper/strategy/skip/SkipStrategy.java
@@ -2,7 +2,7 @@ package com.jd.platform.async.wrapper.strategy.skip;
import com.jd.platform.async.wrapper.WorkerWrapper;
-import java.util.*;
+import java.util.Set;
/**
* @author create by TcSnZh on 2021/5/6-下午3:02
diff --git a/src/test/java/beforev14/depend/DeWorker.java b/asyncTool-core/src/test/java/beforev14/depend/DeWorker.java
similarity index 100%
rename from src/test/java/beforev14/depend/DeWorker.java
rename to asyncTool-core/src/test/java/beforev14/depend/DeWorker.java
diff --git a/src/test/java/beforev14/depend/DeWorker1.java b/asyncTool-core/src/test/java/beforev14/depend/DeWorker1.java
similarity index 100%
rename from src/test/java/beforev14/depend/DeWorker1.java
rename to asyncTool-core/src/test/java/beforev14/depend/DeWorker1.java
diff --git a/src/test/java/beforev14/depend/DeWorker2.java b/asyncTool-core/src/test/java/beforev14/depend/DeWorker2.java
similarity index 100%
rename from src/test/java/beforev14/depend/DeWorker2.java
rename to asyncTool-core/src/test/java/beforev14/depend/DeWorker2.java
diff --git a/src/test/java/beforev14/depend/LambdaTest.java b/asyncTool-core/src/test/java/beforev14/depend/LambdaTest.java
similarity index 98%
rename from src/test/java/beforev14/depend/LambdaTest.java
rename to asyncTool-core/src/test/java/beforev14/depend/LambdaTest.java
index 68a8386338a7096ff4011d2512ba30ed69da3466..7dbaecac0d7c53a5b6e68c9966489704877802d8 100644
--- a/src/test/java/beforev14/depend/LambdaTest.java
+++ b/asyncTool-core/src/test/java/beforev14/depend/LambdaTest.java
@@ -10,6 +10,7 @@ import com.jd.platform.async.wrapper.WorkerWrapper;
* @author sjsdfg
* @since 2020/6/14
*/
+@SuppressWarnings({"deprecation", "RedundantStringFormatCall"})
class LambdaTest {
public static void main(String[] args) throws Exception {
WorkerWrapper, String> workerWrapper2 = new WorkerWrapper.Builder, String>()
diff --git a/src/test/java/beforev14/depend/Test.java b/asyncTool-core/src/test/java/beforev14/depend/Test.java
similarity index 100%
rename from src/test/java/beforev14/depend/Test.java
rename to asyncTool-core/src/test/java/beforev14/depend/Test.java
diff --git a/src/test/java/beforev14/depend/User.java b/asyncTool-core/src/test/java/beforev14/depend/User.java
similarity index 100%
rename from src/test/java/beforev14/depend/User.java
rename to asyncTool-core/src/test/java/beforev14/depend/User.java
diff --git a/src/test/java/beforev14/dependnew/DeWorker.java b/asyncTool-core/src/test/java/beforev14/dependnew/DeWorker.java
similarity index 100%
rename from src/test/java/beforev14/dependnew/DeWorker.java
rename to asyncTool-core/src/test/java/beforev14/dependnew/DeWorker.java
diff --git a/src/test/java/beforev14/dependnew/DeWorker1.java b/asyncTool-core/src/test/java/beforev14/dependnew/DeWorker1.java
similarity index 100%
rename from src/test/java/beforev14/dependnew/DeWorker1.java
rename to asyncTool-core/src/test/java/beforev14/dependnew/DeWorker1.java
diff --git a/src/test/java/beforev14/dependnew/DeWorker2.java b/asyncTool-core/src/test/java/beforev14/dependnew/DeWorker2.java
similarity index 100%
rename from src/test/java/beforev14/dependnew/DeWorker2.java
rename to asyncTool-core/src/test/java/beforev14/dependnew/DeWorker2.java
diff --git a/src/test/java/beforev14/dependnew/Test.java b/asyncTool-core/src/test/java/beforev14/dependnew/Test.java
similarity index 100%
rename from src/test/java/beforev14/dependnew/Test.java
rename to asyncTool-core/src/test/java/beforev14/dependnew/Test.java
diff --git a/src/test/java/beforev14/dependnew/User.java b/asyncTool-core/src/test/java/beforev14/dependnew/User.java
similarity index 100%
rename from src/test/java/beforev14/dependnew/User.java
rename to asyncTool-core/src/test/java/beforev14/dependnew/User.java
diff --git a/src/test/java/beforev14/parallel/ParTimeoutWorker.java b/asyncTool-core/src/test/java/beforev14/parallel/ParTimeoutWorker.java
similarity index 100%
rename from src/test/java/beforev14/parallel/ParTimeoutWorker.java
rename to asyncTool-core/src/test/java/beforev14/parallel/ParTimeoutWorker.java
diff --git a/src/test/java/beforev14/parallel/ParWorker.java b/asyncTool-core/src/test/java/beforev14/parallel/ParWorker.java
similarity index 100%
rename from src/test/java/beforev14/parallel/ParWorker.java
rename to asyncTool-core/src/test/java/beforev14/parallel/ParWorker.java
diff --git a/src/test/java/beforev14/parallel/ParWorker1.java b/asyncTool-core/src/test/java/beforev14/parallel/ParWorker1.java
similarity index 100%
rename from src/test/java/beforev14/parallel/ParWorker1.java
rename to asyncTool-core/src/test/java/beforev14/parallel/ParWorker1.java
diff --git a/src/test/java/beforev14/parallel/ParWorker2.java b/asyncTool-core/src/test/java/beforev14/parallel/ParWorker2.java
similarity index 100%
rename from src/test/java/beforev14/parallel/ParWorker2.java
rename to asyncTool-core/src/test/java/beforev14/parallel/ParWorker2.java
diff --git a/src/test/java/beforev14/parallel/ParWorker3.java b/asyncTool-core/src/test/java/beforev14/parallel/ParWorker3.java
similarity index 100%
rename from src/test/java/beforev14/parallel/ParWorker3.java
rename to asyncTool-core/src/test/java/beforev14/parallel/ParWorker3.java
diff --git a/src/test/java/beforev14/parallel/ParWorker4.java b/asyncTool-core/src/test/java/beforev14/parallel/ParWorker4.java
similarity index 100%
rename from src/test/java/beforev14/parallel/ParWorker4.java
rename to asyncTool-core/src/test/java/beforev14/parallel/ParWorker4.java
diff --git a/src/test/java/beforev14/parallel/TestPar.java b/asyncTool-core/src/test/java/beforev14/parallel/TestPar.java
similarity index 100%
rename from src/test/java/beforev14/parallel/TestPar.java
rename to asyncTool-core/src/test/java/beforev14/parallel/TestPar.java
diff --git a/src/test/java/beforev14/seq/SeqTimeoutWorker.java b/asyncTool-core/src/test/java/beforev14/seq/SeqTimeoutWorker.java
similarity index 100%
rename from src/test/java/beforev14/seq/SeqTimeoutWorker.java
rename to asyncTool-core/src/test/java/beforev14/seq/SeqTimeoutWorker.java
diff --git a/src/test/java/beforev14/seq/SeqWorker.java b/asyncTool-core/src/test/java/beforev14/seq/SeqWorker.java
similarity index 100%
rename from src/test/java/beforev14/seq/SeqWorker.java
rename to asyncTool-core/src/test/java/beforev14/seq/SeqWorker.java
diff --git a/src/test/java/beforev14/seq/SeqWorker1.java b/asyncTool-core/src/test/java/beforev14/seq/SeqWorker1.java
similarity index 100%
rename from src/test/java/beforev14/seq/SeqWorker1.java
rename to asyncTool-core/src/test/java/beforev14/seq/SeqWorker1.java
diff --git a/src/test/java/beforev14/seq/SeqWorker2.java b/asyncTool-core/src/test/java/beforev14/seq/SeqWorker2.java
similarity index 100%
rename from src/test/java/beforev14/seq/SeqWorker2.java
rename to asyncTool-core/src/test/java/beforev14/seq/SeqWorker2.java
diff --git a/src/test/java/beforev14/seq/TestSequential.java b/asyncTool-core/src/test/java/beforev14/seq/TestSequential.java
similarity index 100%
rename from src/test/java/beforev14/seq/TestSequential.java
rename to asyncTool-core/src/test/java/beforev14/seq/TestSequential.java
diff --git a/src/test/java/beforev14/seq/TestSequentialTimeout.java b/asyncTool-core/src/test/java/beforev14/seq/TestSequentialTimeout.java
similarity index 100%
rename from src/test/java/beforev14/seq/TestSequentialTimeout.java
rename to asyncTool-core/src/test/java/beforev14/seq/TestSequentialTimeout.java
diff --git a/asyncTool-core/src/test/java/v15/cases/Case0.java b/asyncTool-core/src/test/java/v15/cases/Case0.java
new file mode 100644
index 0000000000000000000000000000000000000000..52fa6cf1cd4181120a692df65d0475286a0a4f66
--- /dev/null
+++ b/asyncTool-core/src/test/java/v15/cases/Case0.java
@@ -0,0 +1,37 @@
+package v15.cases;
+
+import com.jd.platform.async.executor.Async;
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapperBuilder;
+
+import java.util.concurrent.ExecutionException;
+
+/**
+ * @author tcsnzh[zh.jobs@foxmail.com] create this in 2021/5/25-下午9:10
+ */
+class Case0 {
+ static WorkerWrapperBuilder builder(String id) {
+ return WorkerWrapper.builder()
+ .id(id)
+ .worker((param, allWrappers) -> {
+ System.out.println("wrapper(id=" + id + ") is working");
+ return null;
+ });
+ }
+
+ public static void main(String[] args) {
+ WorkerWrapper a = builder("A").build();
+ WorkerWrapper b = builder("B").build();
+ WorkerWrapper c = builder("C").build();
+ try {
+ Async.work(100, a, b, c).awaitFinish();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ /* 输出:
+ wrapper(id=A) is working
+ wrapper(id=B) is working
+ wrapper(id=C) is working
+ */
+ }
+}
diff --git a/asyncTool-core/src/test/java/v15/cases/Case01.java b/asyncTool-core/src/test/java/v15/cases/Case01.java
new file mode 100644
index 0000000000000000000000000000000000000000..cf1194f6934e8f775a38c5583d1c2e1b854ecb16
--- /dev/null
+++ b/asyncTool-core/src/test/java/v15/cases/Case01.java
@@ -0,0 +1,37 @@
+package v15.cases;
+
+import com.jd.platform.async.executor.Async;
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapperBuilder;
+
+/**
+ * @author tcsnzh[zh.jobs@foxmail.com] create this in 2021/5/25-下午9:13
+ */
+class Case01 {
+ static WorkerWrapperBuilder builder(String id) {
+ return WorkerWrapper.builder()
+ .id(id)
+ .worker((param, allWrappers) -> {
+ System.out.println("wrapper(id=" + id + ") is working");
+ return null;
+ });
+ }
+
+ public static void main(String[] args) {
+ WorkerWrapper a = builder("A").build();
+ WorkerWrapper b = builder("B").depends(a).build();
+ WorkerWrapper c = builder("C").depends(a).build();
+ WorkerWrapper f = builder("F").depends(b, c).build();
+ try {
+ Async.work(100, a).awaitFinish();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ /* 输出:
+ wrapper(id=A) is working
+ wrapper(id=C) is working
+ wrapper(id=B) is working
+ wrapper(id=F) is working
+ */
+ }
+}
diff --git a/asyncTool-core/src/test/java/v15/cases/Case02.java b/asyncTool-core/src/test/java/v15/cases/Case02.java
new file mode 100644
index 0000000000000000000000000000000000000000..0c9a65d3becc10bacef2d6284e7e7f909e1c84ca
--- /dev/null
+++ b/asyncTool-core/src/test/java/v15/cases/Case02.java
@@ -0,0 +1,38 @@
+package v15.cases;
+
+import com.jd.platform.async.executor.Async;
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapperBuilder;
+
+/**
+ * @author tcsnzh[zh.jobs@foxmail.com] create this in 2021/5/25-下午9:15
+ */
+class Case02 {
+ static WorkerWrapperBuilder builder(String id) {
+ return WorkerWrapper.builder()
+ .id(id)
+ .worker((param, allWrappers) -> {
+ System.out.println("wrapper(id=" + id + ") is working");
+ return null;
+ });
+ }
+
+ public static void main(String[] args) {
+ WorkerWrapper f = builder("F").build();
+ WorkerWrapper a = builder("A")
+ .nextOf(builder("B").nextOf(f).build())
+ .nextOf(builder("C").nextOf(f).build())
+ .build();
+ try {
+ Async.work(100, a).awaitFinish();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ /* 输出:
+ wrapper(id=A) is working
+ wrapper(id=B) is working
+ wrapper(id=C) is working
+ wrapper(id=F) is working
+ */
+ }
+}
diff --git a/src/test/java/v15/cases/Case1.java b/asyncTool-core/src/test/java/v15/cases/Case1.java
similarity index 94%
rename from src/test/java/v15/cases/Case1.java
rename to asyncTool-core/src/test/java/v15/cases/Case1.java
index 11075ba86fa6f08a93b8a104610e0e4d70fe0135..c0a1b74ddb14c5002aba0c563690c4f980419563 100644
--- a/src/test/java/v15/cases/Case1.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case1.java
@@ -43,8 +43,8 @@ class Case1 {
)
.build();
try {
- Async.beginWork(1000, a, d);
- } catch (ExecutionException | InterruptedException e) {
+ Async.work(1000, a, d).awaitFinish();
+ } catch (InterruptedException e) {
e.printStackTrace();
}
/* 输出:
diff --git a/asyncTool-core/src/test/java/v15/cases/Case10.java b/asyncTool-core/src/test/java/v15/cases/Case10.java
new file mode 100644
index 0000000000000000000000000000000000000000..2f6fcc60a664d7c43e9f16d8318947152083eeb9
--- /dev/null
+++ b/asyncTool-core/src/test/java/v15/cases/Case10.java
@@ -0,0 +1,72 @@
+package v15.cases;
+
+import com.jd.platform.async.callback.ICallback;
+import com.jd.platform.async.executor.Async;
+import com.jd.platform.async.worker.OnceWork;
+import com.jd.platform.async.worker.WorkResult;
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapperBuilder;
+
+import java.util.concurrent.ExecutionException;
+
+/**
+ * @author tcsnzh[zh.jobs@foxmail.com] create this in 2021/5/26-下午4:07
+ */
+class Case10 {
+ private static WorkerWrapperBuilder builder(String id) {
+ return builder(id, -1L);
+ }
+
+ private static WorkerWrapperBuilder builder(String id, long sleepTime) {
+ return WorkerWrapper.builder()
+ .id(id)
+ .worker((param, allWrappers) -> {
+ System.out.println("\twrapper(id=" + id + ") is working");
+ if (sleepTime > 0) {
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ return null;
+ })
+ .callback((new ICallback() {
+ @Override
+ public void begin() {
+ System.out.println("wrapper(id=" + id + ") has begin . ");
+ }
+
+ @Override
+ public void result(boolean success, String param, WorkResult workResult) {
+ System.out.println("\t\twrapper(id=" + id + ") callback "
+ + (success ? "success " : "fail ")
+ + ", workResult is " + workResult);
+ }
+ }))
+ .allowInterrupt(true);
+ }
+
+ /**
+ * A(10ms) ==> B(10ms) ==> C(10ms)
+ */
+ public static void main(String[] args) throws ExecutionException, InterruptedException {
+ final WorkerWrapper c;
+ final WorkerWrapper b;
+ final WorkerWrapper a = builder("A", 10)
+ .nextOf(b = builder("B", 10)
+ .nextOf(c = builder("C", 10).build())
+ .build())
+ .build();
+ final OnceWork onceWork = Async.work(40, a);
+ Thread.sleep(25);
+ onceWork.pleaseCancelAndAwaitFinish();
+ System.out.println("任务b信息 " + b);
+ System.out.println("任务c信息 " + c);
+ System.out.println("OnceWork信息 " + onceWork);
+ /*
+ 可以看到C的state为SKIP，workResult.ex为CancelSkippedException，即被取消了。
+ 不过有时程序运行慢，导致B被取消了，那么C就不会执行，其状态就为INIT了。
+ */
+ }
+}
diff --git a/src/test/java/v15/cases/Case2.java b/asyncTool-core/src/test/java/v15/cases/Case2.java
similarity index 97%
rename from src/test/java/v15/cases/Case2.java
rename to asyncTool-core/src/test/java/v15/cases/Case2.java
index bd2758de25cb052be87abac997ffb3f90b9d31be..1c930c7f1cc5921dce68452beaa047ae7d64c535 100644
--- a/src/test/java/v15/cases/Case2.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case2.java
@@ -46,7 +46,7 @@ class Case2 {
.id("id:200").worker(new AddWork()).param(200).build();
WorkerWrapper add = WorkerWrapper.builder().id("id:add")
.worker(new AddWork("id:100", "id:200")).depends(wrapper100, wrapper200).build();
- Async.beginWork(20,wrapper100,wrapper200);
+ Async.work(20,wrapper100,wrapper200).awaitFinish();
System.out.println(add.getWorkResult());
// 输出WorkResult{result=300, resultState=SUCCESS, ex=null}
}
diff --git a/src/test/java/v15/cases/Case3.java b/asyncTool-core/src/test/java/v15/cases/Case3.java
similarity index 94%
rename from src/test/java/v15/cases/Case3.java
rename to asyncTool-core/src/test/java/v15/cases/Case3.java
index d149e071473cf16f52ead2a2fe0564471bd5b167..8c5dc985e1a18be7eaa16ef5a33560edbb4279ea 100644
--- a/src/test/java/v15/cases/Case3.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case3.java
@@ -45,7 +45,7 @@ class Case3 {
// 这里用线程数较少的线程池做示例，对于ALL_DEPENDENCIES_ANY_SUCCESS“仅需一个”的效果会好一点
ExecutorService pool = Executors.newFixedThreadPool(2);
try {
- Async.beginWork(1000, pool, a);
+ Async.work(1000, pool, a).awaitFinish();
} finally {
pool.shutdown();
}
@@ -56,6 +56,8 @@ class Case3 {
wrapper(id=B2) is working
wrapper(id=C2) is working
wrapper(id=C1) is working
+ wrapper(id=B4) is working
+ // 我们看到B5被跳过了，没有执行callback
*/
}
}
diff --git a/src/test/java/v15/cases/Case4.java b/asyncTool-core/src/test/java/v15/cases/Case4.java
similarity index 98%
rename from src/test/java/v15/cases/Case4.java
rename to asyncTool-core/src/test/java/v15/cases/Case4.java
index 657f7a7255179cf273b0c744646236ec5f0b5a3c..e78a59937ece96e8beee4ee969f14e6b07c38c20 100644
--- a/src/test/java/v15/cases/Case4.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case4.java
@@ -51,7 +51,7 @@ class Case4 {
}
ExecutorService pool = Executors.newFixedThreadPool(2);
try {
- Async.beginWork(1000, pool, a);
+ Async.work(1000, pool, a).awaitFinish();
} finally {
pool.shutdown();
}
diff --git a/src/test/java/v15/cases/Case5.java b/asyncTool-core/src/test/java/v15/cases/Case5.java
similarity index 98%
rename from src/test/java/v15/cases/Case5.java
rename to asyncTool-core/src/test/java/v15/cases/Case5.java
index 7a96e6d372a3e30b55dc3098c9b987458465790b..5b3c99b1f69740c259cee052709a42f3259a2e58 100644
--- a/src/test/java/v15/cases/Case5.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case5.java
@@ -56,7 +56,7 @@ class Case5 {
WorkerWrapper start = builder("start").nextOf(a1, a2, a3, a4, a5, a6, a7, a8, a9, a10).build();
ExecutorService pool = Executors.newFixedThreadPool(2);
try {
- Async.beginWork(1000, pool, start);
+ Async.work(1000, pool, start).awaitFinish();
} finally {
pool.shutdown();
}
diff --git a/src/test/java/v15/cases/Case6.java b/asyncTool-core/src/test/java/v15/cases/Case6.java
similarity index 88%
rename from src/test/java/v15/cases/Case6.java
rename to asyncTool-core/src/test/java/v15/cases/Case6.java
index d54d855c10659f211fe2fcbab3ad4c7d57da32b1..9de0258b33edb383c8c1ba91a6a1beb78e038eb2 100644
--- a/src/test/java/v15/cases/Case6.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case6.java
@@ -1,17 +1,13 @@
package v15.cases;
+import com.jd.platform.async.callback.ICallback;
import com.jd.platform.async.executor.Async;
import com.jd.platform.async.worker.ResultState;
import com.jd.platform.async.wrapper.WorkerWrapper;
import com.jd.platform.async.wrapper.WorkerWrapperBuilder;
-import com.jd.platform.async.wrapper.strategy.depend.DependWrapperActionStrategy;
import com.jd.platform.async.wrapper.strategy.depend.DependenceAction;
-import com.jd.platform.async.wrapper.strategy.depend.DependenceStrategy;
-import java.util.Set;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
/**
* 示例：自定义依赖策略--对单个wrapper设置“上克下”策略--简单使用与示例
@@ -34,12 +30,14 @@ class Case6 {
}
public static void main(String[] args) throws ExecutionException, InterruptedException {
+ //noinspection unchecked
WorkerWrapper b = builder("B")
// 这里设置了，不论a怎么样b都会快速失败。但是，a设置的对wrapper的特殊策略把它覆盖了。
.depends((dependWrappers, thisWrapper, fromWrapper) ->
DependenceAction.FAST_FAIL
.fastFailException(ResultState.EXCEPTION, new RuntimeException("b 必定失败，除非有上游wrapper救他"))
)
+ .callback(ICallback.PRINT_EXCEPTION_STACK_TRACE)
.build();
WorkerWrapper a = builder("A")
.setNext()
@@ -48,7 +46,7 @@ class Case6 {
.specialToNextWrapper(fromWrapper -> DependenceAction.START_WORK.emptyProperty(), b)
.wrapper(b)
.end().build();
- Async.beginWork(1000, a);
+ Async.work(1000, a).awaitFinish();
System.out.println(a.getWorkResult());
System.out.println(b.getWorkResult());
/* 输出：
diff --git a/src/test/java/v15/cases/Case7.java b/asyncTool-core/src/test/java/v15/cases/Case7.java
similarity index 98%
rename from src/test/java/v15/cases/Case7.java
rename to asyncTool-core/src/test/java/v15/cases/Case7.java
index 1be8b93674c56cdad29a35b8faee202363a4a731..14c2e852fd184ba69d86980ba15d1679fb009228 100644
--- a/src/test/java/v15/cases/Case7.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case7.java
@@ -50,7 +50,7 @@ class Case7 {
.build(),
builder("E", 5).nextOf(d).build()
).build();
- Async.beginWork(1000, a);
+ Async.work(1000, a).awaitFinish();
/* 输出：
wrapper(id=A) is working
wrapper(id=E) is working
diff --git a/src/test/java/v15/cases/Case8.java b/asyncTool-core/src/test/java/v15/cases/Case8.java
similarity index 98%
rename from src/test/java/v15/cases/Case8.java
rename to asyncTool-core/src/test/java/v15/cases/Case8.java
index 2972c9d3627fa1776e6817c48401f93135771bd8..bbc5defdd8533c9a8c4e2e602ac4a845a0603cc7 100644
--- a/src/test/java/v15/cases/Case8.java
+++ b/asyncTool-core/src/test/java/v15/cases/Case8.java
@@ -55,7 +55,7 @@ class Case8 {
.nextOf(builder("C", 20).build())
.build())
.build();
- Async.beginWork(15, a);
+ Async.work(20, a).awaitFinish();
/* 输出：
wrapper(id=A) has begin .
wrapper(id=A) is working
@@ -67,6 +67,7 @@ class Case8 {
wrapper(id=C) callback fail , workResult is WorkResult{result=null, resultState=TIMEOUT, ex=null}
java.lang.InterruptedException: sleep interrupted
at java.lang.Thread.sleep(Native Method)
+ ...
以下异常信息省略
*/
}
diff --git a/asyncTool-core/src/test/java/v15/cases/Case9.java b/asyncTool-core/src/test/java/v15/cases/Case9.java
new file mode 100644
index 0000000000000000000000000000000000000000..d58c4d7461572ee0df1c1862632b375e570e5942
--- /dev/null
+++ b/asyncTool-core/src/test/java/v15/cases/Case9.java
@@ -0,0 +1,65 @@
+package v15.cases;
+
+import com.jd.platform.async.callback.DefaultCallback;
+import com.jd.platform.async.executor.Async;
+import com.jd.platform.async.openutil.collection.CommonDirectedGraph;
+import com.jd.platform.async.openutil.collection.DirectedGraph;
+import com.jd.platform.async.wrapper.QuickBuildWorkerWrapper;
+import com.jd.platform.async.wrapper.WorkerWrapper;
+import com.jd.platform.async.wrapper.strategy.WrapperStrategy;
+
+import java.util.concurrent.*;
+
+/**
+ * 快速构造示例。
+ *
+ * @author create by TcSnZh on 2021/5/17-下午5:23
+ */
+class Case9 {
+ public static void main(String[] args) throws ExecutionException, InterruptedException {
+ DirectedGraph, Object> graph = DirectedGraph.synchronizedDigraph(new CommonDirectedGraph<>());
+ QuickBuildWorkerWrapper w1 = new QuickBuildWorkerWrapper<>("id1",
+ null,
+ (object, allWrappers) -> {
+ System.out.println("I am IWorker 1");
+ return null;
+ },
+ new DefaultCallback<>(),
+ false,
+ true,
+ 100,
+ TimeUnit.MILLISECONDS,
+ new WrapperStrategy.DefaultWrapperStrategy(),
+ graph
+ );
+ QuickBuildWorkerWrapper w2 = new QuickBuildWorkerWrapper<>("id2",
+ null,
+ (object, allWrappers) -> {
+ System.out.println("I am IWorker 2");
+ return null;
+ },
+ new DefaultCallback<>(),
+ false,
+ true,
+ 100,
+ TimeUnit.MILLISECONDS,
+ new WrapperStrategy.DefaultWrapperStrategy(),
+ graph
+ );
+ graph.addNode(w1, w2);
+ graph.putRelation(w1, new Object(), w2);
+
+// System.out.println(graph);
+
+ Async.work(200, w1).awaitFinish();
+
+ System.out.println(" Begin work end .\n w1 : " + w1 + "\n w2 : " + w2 + "\n");
+ /* 输出：
+ I am IWorker 1
+ I am IWorker 2
+ Begin work end .
+ w1 : 省略
+ w2 : 省略
+ */
+ }
+}
diff --git a/src/test/java/v15/wrappertest/Test.java b/asyncTool-core/src/test/java/v15/wrappertest/Test.java
similarity index 100%
rename from src/test/java/v15/wrappertest/Test.java
rename to asyncTool-core/src/test/java/v15/wrappertest/Test.java
diff --git a/asyncTool-openutil/pom.xml b/asyncTool-openutil/pom.xml
new file mode 100644
index 0000000000000000000000000000000000000000..e0603b8a8450d79bd06637c1077b5ae378497438
--- /dev/null
+++ b/asyncTool-openutil/pom.xml
@@ -0,0 +1,19 @@
+
+
+
+ asyncTool
+ com.jd.platform
+ 1.5.1-SNAPSHOT
+
+ 4.0.0
+
+ asyncTool-openutil
+
+
+ 8
+ 8
+
+
+
\ No newline at end of file
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/BiInt.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/BiInt.java
new file mode 100644
index 0000000000000000000000000000000000000000..5902bd6915a5a68998a2871800f8d0329decc7ce
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/BiInt.java
@@ -0,0 +1,109 @@
+package com.jd.platform.async.openutil;
+
+import java.util.Comparator;
+
+/**
+ * 两个int值包装类。重写了{@link #hashCode()}与{@link #equals(Object)}
+ *
+ * @author create by TcSnZh on 2021/5/16-上午1:50
+ */
+public final class BiInt {
+ // properties
+
+ private final int m;
+ private final int n;
+
+ public static final Comparator cmp_m_asc = Comparator.comparingInt(BiInt::getM);
+ public static final Comparator cmp_n_asc = Comparator.comparingInt(BiInt::getN);
+ public static final Comparator cmp_m_desc = cmp_m_asc.reversed();
+ public static final Comparator cmp_n_desc = cmp_n_asc.reversed();
+ public static final Comparator cmp_m_asc_n_asc =
+ cmp_m_asc.thenComparing(cmp_n_asc);
+ public static final Comparator cmp_m_asc_n_desc =
+ cmp_m_asc.thenComparing(cmp_n_desc);
+ public static final Comparator cmp_m_desc_n_asc =
+ cmp_m_desc.thenComparing(cmp_n_asc);
+ public static final Comparator cmp_m_desc_n_desc =
+ cmp_m_desc.thenComparing(cmp_n_desc);
+ public static final Comparator cmp_n_asc_m_asc =
+ cmp_n_asc.thenComparing(cmp_m_asc);
+ public static final Comparator cmp_n_asc_m_desc =
+ cmp_n_asc.thenComparing(cmp_m_desc);
+ public static final Comparator cmp_n_desc_m_asc =
+ cmp_n_desc.thenComparing(cmp_m_asc);
+ public static final Comparator cmp_n_desc_m_desc =
+ cmp_n_desc.thenComparing(cmp_m_desc);
+
+ /**
+ * private constructor , please use {@link #of(int, int)} to build Idx object.
+ */
+ private BiInt(int m, int n) {
+ this.m = m;
+ this.n = n;
+ }
+
+ // getter
+
+ public int getM() {
+ return m;
+ }
+
+ public int getN() {
+ return n;
+ }
+
+ // hashcode and equals
+
+ @Override
+ public int hashCode() {
+ return m ^ n;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o)
+ return true;
+ if (!(o instanceof BiInt))
+ return false;
+ BiInt idx = (BiInt) o;
+ return m == idx.m && n == idx.n;
+ }
+
+ // toString
+
+ @Override
+ public String toString() {
+ return "(" + m + ',' + n + ')';
+ }
+
+ // ========== static ==========
+
+
+ // 工厂方法
+
+ public static BiInt of(int m, int n) {
+ if (m == Integer.MIN_VALUE && n == Integer.MAX_VALUE) {
+ return MIN_TO_MAX;
+ }
+ if (m >= 0 && m < CACHE_RANGE_M && n >= 0 && n < CACHE_RANGE_M) {
+ return cache[m * CACHE_RANGE_M + n];
+ }
+ return new BiInt(m, n);
+ }
+
+ // 缓存区间
+
+ private static final BiInt MIN_TO_MAX = new BiInt(Integer.MIN_VALUE, Integer.MAX_VALUE);
+ private static final BiInt[] cache; // m from 0 to 31 , n from 0 to 31 , total 1023 .
+ private static final int CACHE_RANGE_M = 32; // 0 to 31
+ private static final int CACHE_RANGE_N = 32; // 0 to 31
+
+ static {
+ cache = new BiInt[CACHE_RANGE_M * CACHE_RANGE_N];
+ for (int i = 0; i < CACHE_RANGE_M; i++) {
+ for (int j = 0; j < CACHE_RANGE_N; j++) {
+ cache[i * CACHE_RANGE_M + j] = new BiInt(i, j);
+ }
+ }
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractArray2D.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractArray2D.java
new file mode 100644
index 0000000000000000000000000000000000000000..66c893939548ff20890c20e408a05b63172fadc1
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractArray2D.java
@@ -0,0 +1,74 @@
+package com.jd.platform.async.openutil.collection;
+
+import com.jd.platform.async.openutil.BiInt;
+
+import java.util.Iterator;
+
+/**
+ * @author create by TcSnZh on 2021/5/14-下午9:51
+ */
+public abstract class AbstractArray2D implements Array2D {
+ /**
+ * 用于代替null
+ */
+ protected static final Object NULL = new Object() {
+ @Override
+ public String toString() {
+ return "null";
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ @SuppressWarnings("EqualsDoesntCheckParameterClass")
+ @Override
+ public boolean equals(Object obj) {
+ //noinspection ConstantConditions
+ return obj == null || obj == NULL || obj.equals(null);
+ }
+ };
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder(64).append(this.getClass().getSimpleName()).append('{');
+ Iterator> it = iterator();
+ if (it.hasNext()) {
+ while (true) {
+ Point point = it.next();
+ sb.append('{').append(point.getIdx()).append(':').append(point.getElement()).append('}');
+ if (!it.hasNext()) {
+ break;
+ }
+ sb.append(", ");
+ }
+ }
+ return sb.append('}').toString();
+ }
+
+ public static class PointImpl implements Point {
+ private final BiInt idx;
+ private final E element;
+
+ public PointImpl(BiInt idx, E element) {
+ this.idx = idx;
+ this.element = element;
+ }
+
+ @Override
+ public BiInt getIdx() {
+ return idx;
+ }
+
+ @Override
+ public E getElement() {
+ return element;
+ }
+
+ @Override
+ public String toString() {
+ return "{" + idx + ":" + element + "}";
+ }
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractDirectedGraph.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractDirectedGraph.java
new file mode 100644
index 0000000000000000000000000000000000000000..cb971d5f9f9a81e6fa39602bbedfeb491dd5d823
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractDirectedGraph.java
@@ -0,0 +1,88 @@
+package com.jd.platform.async.openutil.collection;
+
+import java.util.AbstractSet;
+import java.util.Iterator;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * 抽象有向图
+ *
+ * @author create by TcSnZh on 2021/5/13-上午11:37
+ */
+public abstract class AbstractDirectedGraph implements DirectedGraph {
+
+ @Override
+ public String toString() {
+ Set nv = nodesView();
+ Set> rSet = getRelations();
+ StringBuilder sb = new StringBuilder(nv.size() * 10 + rSet.size() * 20)
+ .append(this.getClass().getSimpleName()).append("{nodes=[");
+ Iterator nit = nodesView().iterator();
+ if (nit.hasNext()) {
+ for (; ; ) {
+ sb.append(nit.next());
+ if (!nit.hasNext()) {
+ break;
+ }
+ sb.append(", ");
+ }
+ }
+ sb.append("], relations=[");
+ Iterator> eit = rSet.iterator();
+ if (eit.hasNext()) {
+ for (; ; ) {
+ sb.append(eit.next());
+ if (!eit.hasNext()) {
+ break;
+ }
+ sb.append(", ");
+ }
+ }
+ return sb.append("]}").toString();
+ }
+
+ public abstract class AbstractNodesView extends AbstractSet {
+ @Override
+ public boolean add(N n) {
+ return AbstractDirectedGraph.this.addNode(n);
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ N o1;
+ //noinspection unchecked
+ if (!AbstractDirectedGraph.this.containsNode(o1 = (N) o)) {
+ return false;
+ }
+ AbstractDirectedGraph.this.removeNode(o1);
+ return true;
+ }
+ }
+
+ public static abstract class AbstractEntry implements Entry {
+ @Override
+ public int hashCode() {
+ return this.getFrom().hashCode() ^ this.getTo().hashCode() ^ this.getRelation().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj == this) {
+ return true;
+ }
+ if (!(obj instanceof Graph.Entry)) {
+ return false;
+ }
+ Entry obj1 = (Entry) obj;
+ return Objects.equals(this.getFrom(), obj1.getFrom())
+ && Objects.equals(this.getTo(), obj1.getTo())
+ && Objects.equals(this.getRelation(), obj1.getRelation());
+ }
+
+ @Override
+ public String toString() {
+ return "{from=" + getFrom() + ", relation=" + getRelation() + ", to=" + getTo() + "]";
+ }
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractStoreArk.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractStoreArk.java
new file mode 100644
index 0000000000000000000000000000000000000000..6caea3c07769ab3c681ae594846c65d3b6f47cdb
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/AbstractStoreArk.java
@@ -0,0 +1,22 @@
+package com.jd.platform.async.openutil.collection;
+
+/**
+ * @author create by TcSnZh on 2021/5/14-上午2:33
+ */
+public abstract class AbstractStoreArk implements StoreArk {
+
+ @Override
+ public boolean isEmpty() {
+ return size() <= 0;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder(size() * 10).append(this.getClass().getSimpleName()).append("{");
+ if (!isEmpty()) {
+ stream().forEach(entry -> sb.append(entry.getKey()).append(":").append(entry.getValue()).append(", "));
+ sb.delete(sb.length() - 2, sb.length());
+ }
+ return sb.append("}").toString();
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/Array2D.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/Array2D.java
new file mode 100644
index 0000000000000000000000000000000000000000..c4bcc642d85a3c5808c6ab78dddc29bc9a8ed9bf
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/Array2D.java
@@ -0,0 +1,169 @@
+package com.jd.platform.async.openutil.collection;
+
+import com.jd.platform.async.openutil.BiInt;
+
+import java.util.*;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+/**
+ * 二维数组
+ *
+ * @author create by TcSnZh on 2021/5/14-下午9:50
+ */
+@SuppressWarnings("unused")
+public interface Array2D extends Iterable> {
+ /**
+ * 有多少行
+ */
+ int lineLength();
+
+ /**
+ * 有多少列
+ */
+ int columnLength();
+
+ /**
+ * 添加元素到指定位置
+ *
+ * @param line 行
+ * @param column 列
+ * @param element 元素
+ * @return 如果之前添加过元素，将返回替换掉的之前的元素
+ * @throws IndexOutOfBoundsException 行列超出范围
+ */
+ E add(int line, int column, E element);
+
+ /**
+ * 如果不存在的话则添加元素
+ *
+ * {@link #add(int, int, Object)}
+ *
+ * @return 不存在且成功添加，返回true。
+ */
+ default boolean addIfAbsent(int line, int column, E element) {
+ if (get(line, column) != null) {
+ return false;
+ }
+ add(line, column, element);
+ return true;
+ }
+
+ /**
+ * 删除元素
+ *
+ * @param line 行
+ * @param column 列
+ * @return 返回移出的元素
+ * @throws IndexOutOfBoundsException 行列超出返回
+ * @throws IllegalArgumentException 如果原本不存在元素
+ */
+ E remove(int line, int column);
+
+ /**
+ * 存在则移除，不存在则返回null
+ *
+ * @param line 行
+ * @param column 列
+ * @return 如果不存在，返回null。存在则返回被移出的元素。
+ * @throws IndexOutOfBoundsException 行列超出范围
+ */
+ default E removeIfAbsent(int line, int column) {
+ if (get(line, column) == null) {
+ return null;
+ }
+ return remove(line, column);
+ }
+
+ /**
+ * 获取元素
+ *
+ * @param line 行
+ * @param column 列
+ * @return 如果存在，返回该元素。不存在则返回null。
+ * @throws IndexOutOfBoundsException 行列超出范围
+ */
+ E get(int line, int column);
+
+ /**
+ * 是否包含元素
+ *
+ * @param element 元素
+ * @return 有这个元素就返回true。
+ */
+ boolean containsElement(E element);
+
+ /**
+ * 获取整行的元素
+ *
+ * @param line 行号
+ * @return 返回key为列号，value为元素的Map
+ * @throws IndexOutOfBoundsException 行号超出范围
+ */
+ Map fullLine(int line);
+
+ /**
+ * 获取整列的元素
+ *
+ * @param column 列号
+ * @return 返回key为行号，value为元素的Map
+ * @throws IndexOutOfBoundsException 列号超出范围
+ */
+ Map fullColumn(int column);
+
+ /**
+ * 迭代器
+ *
+ * @param foreachOrder 遍历顺序
+ * @return 如果本容器不允许null值存在，只需返回存在的元素的键即可。如果允许null值存在，仅需返回包括人工放入的null值的键即可。
+ */
+ Iterator> iterator(Comparator foreachOrder);
+
+ @Override
+ default Iterator> iterator() {
+ //noinspection unchecked
+ return (Iterator) iterator(BiInt.cmp_m_asc_n_asc);
+ }
+
+ /**
+ * 流
+ */
+ default Stream> stream() {
+ return StreamSupport.stream(spliterator(), false);
+ }
+
+ default Stream> parallelStream() {
+ return StreamSupport.stream(spliterator(), true);
+ }
+
+ default Spliterator> spliterator(Comparator foreachOrder) {
+ return Spliterators.spliteratorUnknownSize(iterator(foreachOrder), 0);
+ }
+
+ default Stream> stream(Comparator foreachOrder) {
+ return StreamSupport.stream(spliterator(foreachOrder), false);
+ }
+
+ default Stream> parallelStream(Comparator foreachOrder) {
+ return StreamSupport.stream(spliterator(foreachOrder), true);
+ }
+
+ /**
+ * 端点
+ *
+ * @param 元素泛型
+ */
+ interface Point {
+ BiInt getIdx();
+
+ default int getLine() {
+ return getIdx().getM();
+ }
+
+ default int getColumn() {
+ return getIdx().getN();
+ }
+
+ E getElement();
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CachedStoreArk.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CachedStoreArk.java
new file mode 100644
index 0000000000000000000000000000000000000000..a8a1362d9484bf2413a587cb47f4a0580318b2d6
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CachedStoreArk.java
@@ -0,0 +1,60 @@
+package com.jd.platform.async.openutil.collection;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.function.Supplier;
+
+/**
+ * 一个缓存元素位置的储存柜。
+ *
+ * @author create by TcSnZh on 2021/5/14-上午2:37
+ */
+public class CachedStoreArk extends AbstractStoreArk {
+ private final StoreArk inner;
+
+ private final Map cacheMap = new HashMap<>();
+
+ public CachedStoreArk() {
+ this(CommonStoreArk::new);
+ }
+
+ private CachedStoreArk(Supplier> sup) {
+ this.inner = sup.get();
+ }
+
+ @Override
+ public int store(E element) {
+ int id = inner.store(element);
+ cacheMap.put(element, id);
+ return id;
+ }
+
+ @Override
+ public E peek(int id) {
+ return inner.peek(id);
+ }
+
+ @Override
+ public E takeOut(int id) {
+ E e = inner.takeOut(id);
+ cacheMap.remove(e);
+ return e;
+ }
+
+ @Override
+ public int size() {
+ return inner.size();
+ }
+
+ @Override
+ public Iterator> iterator() {
+ return inner.iterator();
+ }
+
+ @Override
+ public int findId(E element) {
+ Integer idNullable = cacheMap.get(element);
+ return idNullable == null ? -1 : idNullable;
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CommonDirectedGraph.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CommonDirectedGraph.java
new file mode 100644
index 0000000000000000000000000000000000000000..0683d2ca71c3da9831039b0b48a39c80008adc84
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CommonDirectedGraph.java
@@ -0,0 +1,145 @@
+package com.jd.platform.async.openutil.collection;
+
+import java.util.*;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * 线程不安全的有向图。
+ *
+ * 不允许放入null。
+ *
+ * @author create by TcSnZh on 2021/5/14-上午2:22
+ */
+public class CommonDirectedGraph extends AbstractDirectedGraph {
+
+ // ========== properties ==========
+
+ private final StoreArk nodes = new CachedStoreArk<>();
+ private final Array2D arr = new SparseArray2D<>();
+
+ // ========== methods ==========
+
+ @Override
+ public boolean addNode(N node) {
+ if (containsNode(Objects.requireNonNull(node))) {
+ return false;
+ }
+ nodes.store(node);
+ return true;
+ }
+
+ @Override
+ public boolean containsNode(N node) {
+ return node != null && findNodeId(node, false) >= 0;
+ }
+
+ @Override
+ public Set> removeNode(N node) {
+ int id = findNodeId(Objects.requireNonNull(node), true);
+ LinkedHashSet> res = new LinkedHashSet<>();
+ // 查找node为from的键
+ arr.fullLine(id).forEach((toNodeId, relation) -> {
+ res.add(new OuterEntry<>(node, nodes.peek(toNodeId), relation));
+ arr.remove(id, toNodeId);
+ });
+ // 查找node为to的键
+ arr.fullColumn(id).forEach((fromNodeId, relation) -> {
+ // 在上一次遍历中，fromNodeId为id，
+ if (fromNodeId == id) {
+ return;
+ }
+ res.add(new OuterEntry<>(nodes.peek(fromNodeId), node, relation));
+ arr.remove(fromNodeId, id);
+ });
+ nodes.takeOut(id);
+ return res;
+ }
+
+ @Override
+ public R putRelation(N fromNode, R relation, N toNode) {
+ return arr.add(
+ findNodeId(Objects.requireNonNull(fromNode), true),
+ findNodeId(Objects.requireNonNull(toNode), true),
+ Objects.requireNonNull(relation)
+ );
+ }
+
+ @Override
+ public Set> getRelationFrom(N from) {
+ int id = findNodeId(Objects.requireNonNull(from), true);
+ LinkedHashSet> res = new LinkedHashSet<>();
+ // 查找node为from的键
+ arr.fullLine(id).forEach((toNodeId, relation) -> res.add(new OuterEntry<>(from, nodes.peek(toNodeId), relation)));
+ return res;
+ }
+
+ @Override
+ public Set> getRelationTo(N to) {
+ int id = findNodeId(Objects.requireNonNull(to), true);
+ LinkedHashSet> res = new LinkedHashSet<>();
+ // 查找node为to的键
+ arr.fullColumn(id).forEach((fromNodeId, relation) ->
+ res.add(new OuterEntry<>(nodes.peek(fromNodeId), to, relation)));
+ return res;
+ }
+
+ @Override
+ public Set nodesView() {
+ return new AbstractNodesView() {
+ @Override
+ public Iterator iterator() {
+ return nodes.stream().map(Map.Entry::getValue).iterator();
+ }
+
+ @Override
+ public int size() {
+ return nodes.size();
+ }
+ };
+ }
+
+ @Override
+ public Set> getRelations() {
+ return arr.stream().map((Function, Entry>) rPoint -> new OuterEntry<>(
+ nodes.peek(rPoint.getLine()),
+ nodes.peek(rPoint.getColumn()),
+ rPoint.getElement()
+ )).collect(Collectors.toSet());
+ }
+
+ private int findNodeId(N node, boolean mustExistElseThrowEx) {
+ int id = nodes.findId(Objects.requireNonNull(node));
+ if (mustExistElseThrowEx && id < 0) {
+ throw new IllegalArgumentException("No node exists : " + node);
+ }
+ return id;
+ }
+
+ private static class OuterEntry extends AbstractEntry {
+ private final N from;
+ private final N to;
+ private final R relation;
+
+ public OuterEntry(N from, N to, R relation) {
+ this.from = from;
+ this.to = to;
+ this.relation = relation;
+ }
+
+ @Override
+ public N getFrom() {
+ return from;
+ }
+
+ @Override
+ public N getTo() {
+ return to;
+ }
+
+ @Override
+ public R getRelation() {
+ return relation;
+ }
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CommonStoreArk.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CommonStoreArk.java
new file mode 100644
index 0000000000000000000000000000000000000000..ae05f6fbc056a8e605a1a564be9b4a77db1af4f6
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/CommonStoreArk.java
@@ -0,0 +1,159 @@
+package com.jd.platform.async.openutil.collection;
+
+import java.util.*;
+
+/**
+ * 自动扩容的id储物柜，线程不安全。
+ *
+ * @author create by TcSnZh on 2021/5/13-下午1:24
+ */
+public class CommonStoreArk extends AbstractStoreArk {
+ private Object[] elements;
+
+ /**
+ * 已经分配的下标数
+ */
+ private int allocSize = 0;
+
+ /**
+ * 保存着最小空元素的队列
+ */
+ private final Queue emptyPoints = new PriorityQueue<>(Integer::compareTo);
+
+ public CommonStoreArk(int initialCapacity) {
+ elements = new Object[initialCapacity];
+ }
+
+ public CommonStoreArk() {
+ this(10);
+ }
+
+ @Override
+ public int store(E element) {
+ int id;
+ elements[id = pollId()] = element;
+ return id;
+ }
+
+ @Override
+ public E peek(int id) {
+ if (id < 0) {
+ throw new IllegalArgumentException("id " + id + " can't be negative");
+ }
+ if (id >= elements.length) {
+ return null;
+ }
+ //noinspection unchecked
+ return (E) elements[id];
+ }
+
+ @Override
+ public E takeOut(int id) {
+ if (id < 0) {
+ throw new IllegalArgumentException("id " + id + " can't be negative");
+ }
+ if (id >= elements.length) {
+ return null;
+ }
+ //noinspection unchecked
+ E out = (E) elements[id];
+ elements[id] = null;
+ if (id == allocSize - 1) {
+ allocSize--;
+ } else {
+ emptyPoints.add(id);
+ }
+ return out;
+ }
+
+ @Override
+ public int size() {
+ return allocSize - emptyPoints.size();
+ }
+
+ @Override
+ public Iterator> iterator() {
+ return new Iterator>() {
+ private final Map.Entry[] items;
+
+ private int idx = 0;
+
+ {
+ //noinspection unchecked
+ items = new Map.Entry[size()];
+ int itemsIdx = 0;
+ Iterator emptyPointItr = emptyPoints.iterator();
+ for (int i = 0; i < allocSize; i++) {
+ Object element = elements[i];
+ if (element == null) {
+ continue;
+ }
+ final int _i = i;
+ //noinspection unchecked
+ items[itemsIdx++] = new Map.Entry() {
+ private final int k = _i;
+ private E v = (E) element;
+
+ @Override
+ public Integer getKey() {
+ return k;
+ }
+
+ @Override
+ public E getValue() {
+ return v;
+ }
+
+ @Override
+ public E setValue(E value) {
+ E _v = this.v;
+ this.v = value;
+ return _v;
+ }
+
+ @Override
+ public String toString() {
+ return "{" + k + ':' + v + '}';
+ }
+ };
+ }
+ }
+
+
+ @Override
+ public boolean hasNext() {
+ return idx < items.length;
+ }
+
+ @Override
+ public Map.Entry next() {
+ return items[idx++];
+ }
+ };
+ }
+
+ @Override
+ public int findId(E element) {
+ int i = 0;
+ for (Object o : elements) {
+ if (Objects.equals(o, element)) {
+ return i;
+ }
+ i++;
+ }
+ return -1;
+ }
+
+ private int pollId() {
+ if (!emptyPoints.isEmpty()) {
+ return emptyPoints.poll();
+ }
+ int id = allocSize++;
+ int length = elements.length;
+ if (id >= length) {
+ // 扩容
+ elements = Arrays.copyOf(elements, Math.max(length + 1, length + (length >> 1)));
+ }
+ return id;
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/DirectedGraph.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/DirectedGraph.java
new file mode 100644
index 0000000000000000000000000000000000000000..5ac6dc7c0ecb630348a057d190746ea219c1aacd
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/DirectedGraph.java
@@ -0,0 +1,182 @@
+package com.jd.platform.async.openutil.collection;
+
+import java.util.*;
+
+/**
+ * @author create by TcSnZh on 2021/5/16-下午11:27
+ */
+public interface DirectedGraph extends Graph {
+ @Override
+ default boolean isDirected() {
+ return true;
+ }
+
+ static DirectedGraph readOnlyDigraph(DirectedGraph source) {
+ return new ReadOnlyDirectedGraph<>(source);
+ }
+
+ static DirectedGraph synchronizedDigraph(DirectedGraph source) {
+ return synchronizedDigraph(source, new Object());
+ }
+
+ static DirectedGraph synchronizedDigraph(DirectedGraph source, Object mutex) {
+ return new SyncDirectedGraph<>(source, mutex);
+ }
+
+ class ReadOnlyDirectedGraph extends AbstractDirectedGraph {
+ private final DirectedGraph source;
+
+ public ReadOnlyDirectedGraph(DirectedGraph source) {
+ this.source = source;
+ }
+
+ private static UnsupportedOperationException readOnlyGraph() {
+ return new UnsupportedOperationException("readOnly graph");
+ }
+
+ @Override
+ public boolean addNode(N node) {
+ throw readOnlyGraph();
+ }
+
+ @Override
+ public boolean containsNode(N node) {
+ return source.containsNode(node);
+ }
+
+ @Override
+ public Set> removeNode(N node) {
+ throw readOnlyGraph();
+ }
+
+ @Override
+ public R putRelation(N fromNode, R relation, N toNode) {
+ throw readOnlyGraph();
+ }
+
+ @Override
+ public Set> getRelationFrom(N from) {
+ return source.getRelationFrom(from);
+ }
+
+ @Override
+ public Set> getRelationTo(N to) {
+ return source.getRelationTo(to);
+ }
+
+ @Override
+ public Set nodesView() {
+ return new AbstractSet() {
+ private final Set nodesViewSource = source.nodesView();
+
+ @Override
+ public Iterator iterator() {
+ return new Iterator() {
+ private final Iterator iteratorSource = nodesViewSource.iterator();
+
+ @Override
+ public boolean hasNext() {
+ return iteratorSource.hasNext();
+ }
+
+ @Override
+ public N next() {
+ return iteratorSource.next();
+ }
+
+ @Override
+ public void remove() {
+ throw readOnlyGraph();
+ }
+ };
+ }
+
+ @Override
+ public int size() {
+ return nodesViewSource.size();
+ }
+
+ @Override
+ public boolean add(N n) {
+ throw readOnlyGraph();
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ throw readOnlyGraph();
+ }
+ };
+ }
+
+ @Override
+ public Set> getRelations() {
+ return source.getRelations();
+ }
+ }
+
+ class SyncDirectedGraph extends AbstractDirectedGraph {
+ private final DirectedGraph source;
+ private final Object mutex;
+
+ public SyncDirectedGraph(DirectedGraph source, Object mutex) {
+ this.source = source;
+ this.mutex = mutex;
+ }
+
+ @Override
+ public boolean addNode(N node) {
+ synchronized (mutex) {
+ return source.addNode(node);
+ }
+ }
+
+ @Override
+ public boolean containsNode(N node) {
+ synchronized (mutex) {
+ return source.containsNode(node);
+ }
+ }
+
+ @Override
+ public Set> removeNode(N node) {
+ synchronized (mutex) {
+ return source.removeNode(node);
+ }
+ }
+
+ @Override
+ public R putRelation(N fromNode, R relation, N toNode) {
+ synchronized (mutex) {
+ return source.putRelation(fromNode, relation, toNode);
+ }
+ }
+
+ @Override
+ public Set> getRelationFrom(N from) {
+ synchronized (mutex) {
+ return source.getRelationFrom(from);
+ }
+ }
+
+ @Override
+ public Set> getRelationTo(N to) {
+ synchronized (mutex) {
+ return source.getRelationTo(to);
+ }
+ }
+
+ @Override
+ public Set nodesView() {
+ synchronized (mutex) {
+ return Collections.synchronizedSet(source.nodesView());
+ }
+ }
+
+ @Override
+ public Set> getRelations() {
+ synchronized (mutex) {
+ return source.getRelations();
+ }
+ }
+ }
+}
diff --git a/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/Graph.java b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/Graph.java
new file mode 100644
index 0000000000000000000000000000000000000000..e9903ff07b9514af47c6f8aec3844f6ecb000b1a
--- /dev/null
+++ b/asyncTool-openutil/src/main/java/com/jd/platform/async/openutil/collection/Graph.java
@@ -0,0 +1,106 @@
+package com.jd.platform.async.openutil.collection;
+
+import java.util.Set;
+
+/**
+ * 图数据结构
+ *
+ * @author create by TcSnZh on 2021/5/13-上午11:37
+ */
+@SuppressWarnings("unused")
+public interface Graph