基于Java的多线程并发编程框架gobrs-async设计源码

package com.gobrs.async.core; import com.gobrs.async.core.callback.ErrorCallback; import com.gobrs.async.core.common.domain.AnyConditionResult; import com.gobrs.async.core.common.domain.AsyncParam; import com.gobrs.async.core.common.domain.TaskResult; import com.gobrs.async.core.common.domain.TaskStatus; import com.gobrs.async.core.common.enums.ExpState; import com.gobrs.async.core.common.enums.ResultState; import com.gobrs.async.core.common.enums.TaskEnum; import com.gobrs.async.core.common.exception.GobrsForceStopException; import com.gobrs.async.core.common.exception.ManualStopException; import com.gobrs.async.core.config.ConfigManager; import com.gobrs.async.core.log.LogWrapper; import com.gobrs.async.core.log.TraceUtil; import com.gobrs.async.core.task.AsyncTask; import com.gobrs.async.core.task.TaskUtil; import com.gobrs.async.core.timer.GobrsFutureTask; import com.gobrs.async.core.timer.GobrsTimer; import com.gobrs.async.core.timer.TimerListener; import lombok.extern.slf4j.Slf4j; import org.springframework.util.CollectionUtils; import java.lang.ref.Reference; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.Future; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import static com.gobrs.async.core.common.def.DefaultConfig.*; import static com.gobrs.async.core.common.enums.InterruptEnum.INTERRUPTED; import static com.gobrs.async.core.common.enums.InterruptEnum.INTERRUPTTING; import static com.gobrs.async.core.task.ReUsing.reusing; import static com.gobrs.async.core.timer.GobrsFutureTask.STOP_STAMP; /** * The type Task actuator. * * @param <Param> the type parameter * @param <Result> the type parameter */ @Slf4j public class TaskActuator<Param, Result> implements Callable, Cloneable { /** * The Support. */ public TaskSupport support; /** * Tasks to be performed */ public final AsyncTask<Param, Result> task; /** * Upstream dependent quantity */ public volatile int upstreamDepdends; /** * depend com.gobrs.async.com.gobrs.async.test.task */ public final List<AsyncTask<?, ?>> subTasks; /** * The Param. */ public AsyncParam<Param> param; private Lock lock; private Map<AsyncTask<?, ?>, List<AsyncTask<?, ?>>> upwardTasksMap; /** * Instantiates a new Task actuator. * * @param asyncTask the async com.gobrs.async.com.gobrs.async.test.task * @param depends the depends * @param subTasks the sub tasks */ TaskActuator(AsyncTask<Param, Result> asyncTask, int depends, List<AsyncTask<?, ?>> subTasks) { this.task = asyncTask; this.upstreamDepdends = depends > 1 & task.isAny() ? 1 : depends; this.subTasks = subTasks; } /** * Instantiates a new Task actuator. * * @param asyncTask the async com.gobrs.async.com.gobrs.async.test.task * @param depends the depends * @param subTasks the sub tasks * @param upwardTasksMap the upward tasks map */ TaskActuator(AsyncTask<Param, Result> asyncTask, int depends, List<AsyncTask<?, ?>> subTasks, Map<AsyncTask<?, ?>, List<AsyncTask<?, ?>>> upwardTasksMap) { this.task = asyncTask; this.upstreamDepdends = depends > 1 & task.isAny() ? 1 : depends; this.subTasks = subTasks; this.upwardTasksMap = upwardTasksMap; } /** * Initialize the object cloned from prototype. * * @param support the support * @param param the param */ void init(TaskSupport support, AsyncParam param) { this.support = support; this.param = param; } @Override public Object call() throws Exception { Param parameter = getParameter(task); preparation(); TaskLoader taskLoader = support.getTaskLoader(); /** * If the conditions are not met * no execution is performed */ Result result = null; try { /** * 判断任务是否有必要执行 * 1、necessary 返回true * 2、如果具备执行结果 则无需执行 */ if (executeNecessary(parameter, task)) { task.prepare(parameter); /** * Unified front intercept * 统一前置处理 */ taskLoader.preInterceptor(parameter, task.getName()); /** * Perform a com.gobrs.async.com.gobrs.async.test.task * 执行核心任务处理 */ result = task.taskAdapter(parameter, support); /** * Setting Task Results * 设置任务结果 */ result(result); stopAsync0(parameter, support); /** * 状态改变 */ change(); /** * 数量统计 */ statisticsOptimalCount(); /** * Post-processing of tasks * 后置任务 */ taskLoader.postInterceptor(result, task.getName()); /** * Success com.gobrs.async.callback * 执行成功回调 */ task.onSuccess(support); } noRepeat(taskLoader, result); } catch (Exception e) { try { exceptionProcess(parameter, taskLoader, e); } catch (Exception exception) { if (log.isErrorEnabled()) { log.error("<{}> [{}] exceptionProcess error {} ", TraceUtil.get(), task.getName(), e); } taskLoader.stopSingleTaskLine(subTasks); } } finally { clear(); stopOrRelease(parameter, taskLoader); } return result; } private void result(Result result) { if (TaskEnum.METHOD.getType().equals(task.getType())) { support.getResultMap().put(task.getName(), buildSuccessResult(result)); } else { support.getResultMap().put(task.getClass(), buildSuccessResult(result)); } } private void stopAsync0(Object parameter, TaskSupport support) { AtomicInteger diagnose = support.getTaskLoader().getINTERRUPTFLAG(); if (diagnose.get() == INTERRUPTTING.getState() && diagnose.compareAndSet(INTERRUPTTING.getState(), INTERRUPTED.getState())) { ErrorCallback<Object> errorCallback = new ErrorCallback<Object>(() -> parameter, null, support, task); LogWrapper logWrapper = support.getLogWrapper(); if (Objects.nonNull(logWrapper)) { logWrapper.setStopTaskName(task.getName()); } support.taskLoader.setExpCode(new AtomicInteger(ExpState.STOP_ASYNC.getCode())); support.getResultMap().put(task.getName(), buildErrorResult(null, new ManualStopException("Manually executing stopAsync Exception"))); support.getTaskLoader().isRunning().set(false); support.getTaskLoader().errorInterrupted(errorCallback); } } /** * 停止任务 或 释放资源 * * @param parameter * @param taskLoader */ private void stopOrRelease(Object parameter, TaskLoader<Param, Result> taskLoader) throws Exception { if (task.getTimeoutInMilliseconds() > TASK_TIME_OUT) { futureStopRelease(parameter, taskLoader); } if (task.isExclusive()) { releaseFutureTasks(); } } /** * 根据中断位强制释放资源 针对开发者使用死循环等问题fix