如何实现一个并发任务执行框架？

/**
 * @Author SunnyBear
 * @Description 方法本身运行是否正确的结果类型
 */
public enum TaskResultType {
    /**
     * 方法执行完成，业务结果也正确
     */
    SUCCESS,
    /**
     * 方法执行完成，业务结果错误
     */
    FAILURE,
    /**
     * 方法执行抛出异常
     */
    EXCEPTION
}

/**
 * @Author SunnyBear
 * @Description 任务处理后返回的结果实体类
 */
public class TaskResult<R> {

    /**
     * 方法执行结果
     */
    private final TaskResultType taskResultType;
    /**
     * 方法执行后的结果数据
     */
    private final R returnValue;
    /**
     * 如果方法执行失败，可以在这里填充原因
     */
    private final String reason;

    public TaskResult(TaskResultType taskResultType, R returnValue, String reason) {
        this.taskResultType = taskResultType;
        this.returnValue = returnValue;
        this.reason = reason;
    }

    public TaskResultType getTaskResultType() {
        return taskResultType;
    }

    public R getReturnValue() {
        return returnValue;
    }

    public String getReason() {
        return reason;
    }

    @Override
    public String toString() {
        return "TaskResult{" +
                "taskResultType=" + taskResultType +
                ", returnValue=" + returnValue +
                ", reason='" + reason + '\'' +
                '}';
    }
}

/**
 * @Author SunnyBear
 * @Description 要求框架使用者实现的任务接口，因为任务的性质在调用时才知道，所以传入的参数(T)和方法(R)的返回值均使用泛型
 */
public interface ITaskProcesser<T, R> {

    TaskResult<R> taskExecute(T data);
}

/**
 * @Author SunnyBear
 * @Description 存放的队列的元素
 */
public class Item<T> implements Delayed {

    /**
     * 到期时间(秒)
     */
    private long activeTime;
    /**
     * 业务数据
     */
    private T data;

    public Item(long activeTime, T data) {
        this.activeTime = activeTime * 1000 + System.currentTimeMillis();
        this.data = data;
    }

    public long getActiveTime() {
        return activeTime;
    }

    public T getData() {
        return data;
    }

    /**
     * 从这个方法返回到激活日期的剩余时间
     */
    @Override
    public long getDelay(TimeUnit unit) {
        long time = unit.convert(this.activeTime - System.currentTimeMillis(), unit);
        return time;
    }

    /**
     * Delayed接口继承了Comparable接口，按剩余时间排序，实际计算考虑精度为纳秒数
     */
    @Override
    public int compareTo(Delayed o) {
        long d = (getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        if (d == 0) {
            return 0;
        } else {
            if (d < 0) {
                return -1;
            } else {
                return 1;
            }
        }
    }
}

/**
 * @Author SunnyBear
 * @Description 提交给框架执行的工作实体类，工作：表示本批次需要处理的同性质任务(Task)的一个集合
 */
public class JobInfo<R> {

    /**
     * 工作名，用以区分框架中唯一的工作
     */
    private final String jobName;
    /**
     * 工作中任务的长度，即一个工作多少个任务
     */
    private final int jobLength;
    /**
     * 处理工作中任务的处理器
     */
    private final ITaskProcesser<?, ?> taskProcesser;
    /**
     * 任务的成功次数
     */
    private AtomicInteger successCount;
    /**
     * 工作中任务目前已经处理的次数
     */
    private AtomicInteger taskProcessCount;
    /**
     * 存放每个任务的处理结果，供查询用
     */
    private LinkedBlockingDeque<TaskResult<R>> taskDetailQueues;
    /**
     * 保留工作结果信息供查询的时长
     */
    private final long expireTime;
    /**
     * checkJobProcesser，用于添加处理的结果到延时队列
     */
    private CheckJobProcesser checkJobProcesser = CheckJobProcesser.getInstance();

    public JobInfo(String jobName, int jobLength,
                   ITaskProcesser<?, ?> taskProcesser,
                   long expireTime) {
        this.jobName = jobName;
        this.jobLength = jobLength;
        this.taskProcesser = taskProcesser;
        this.successCount = new AtomicInteger(0);
        this.taskProcessCount = new AtomicInteger(0);
        this.taskDetailQueues = new LinkedBlockingDeque<>(jobLength);
        this.expireTime = expireTime;
    }

    public String getJobName() {
        return jobName;
    }

    public int getJobLength() {
        return jobLength;
    }

    public long getExpireTime() {
        return expireTime;
    }

    public ITaskProcesser<?, ?> getTaskProcesser() {
        return taskProcesser;
    }

    public AtomicInteger getTaskProcessCount() {
        return taskProcessCount;
    }

    public AtomicInteger getSuccessCount() {
        return successCount;
    }

    /**
     * 失败次数
     */
    public long getFailureCount() {
        return taskProcessCount.get() - successCount.get();
    }

    /**
     * 工作的整体进度信息
     */
    public String getTotalProcess() {
        return "Success[ " + successCount.get() + " ] / Current[ " + taskProcessCount.get() + " ] Total[ " + jobLength + " ]";
    }

    /**
     * 提供工作中每个任务的处理结果
     */
    public List<TaskResult<R>> getTaskDetail() {
        List<TaskResult<R>> taskResultList = new LinkedList<>();
        TaskResult<R> taskResult;
        while ((taskResult = taskDetailQueues.pollFirst()) != null) {
            taskResultList.add(taskResult);
        }
        return taskResultList;
    }

    /**
     * 个任务处理完成后，记录任务的处理结果，因为从业务应用的角度来说，
     * 对查询任务进度数据的一致性要不高
     * 我们保证最终一致性即可，无需对整个方法加锁
     */
    public void addTaskResult(TaskResult<R> taskResult){
        if(TaskResultType.SUCCESS.equals(taskResult.getTaskResultType())){
            successCount.incrementAndGet();
        }
        taskProcessCount.incrementAndGet();
        taskDetailQueues.addLast(taskResult);
        if(taskProcessCount.get() == jobLength){
            checkJobProcesser.putJob(expireTime, jobName);
        }
    }
}

/**
 * @Author SunnyBear
 * @Description 框架的主题类，也是调用者主要使用的类
 */
public class PendingJobPool {

    /**
     * 框架运行时的线程数，与机器的CPU数相同
     */
    private static final int THREAD_COUNTS = Runtime.getRuntime().availableProcessors();
    /**
     * 用以存放待处理的任务，供线程池使用
     */
    private static BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(5000);
    /**
     * 线程池，固定大小，有界队列
     */
    private static ExecutorService taskExecutor
            = new ThreadPoolExecutor(THREAD_COUNTS, THREAD_COUNTS, 60, TimeUnit.SECONDS, taskQueue);
    /**
     * 工作信息的存放容器
     */
    private static ConcurrentHashMap<String, JobInfo<?>> jobInfoMap = new ConcurrentHashMap<>();

    public static Map<String, JobInfo<?>> getMap() {
        return jobInfoMap;
    }

    private PendingJobPool() {

    }

    /**
     * 单例模式
     */
    private static PendingJobPool threadPool = new PendingJobPool();

    public static PendingJobPool getInstance() {
        return threadPool;
    }

    private static class PendingTask<T, R> implements Runnable {

        private JobInfo<R> jobInfo;
        // 任务参数
        private T processData;

        public PendingTask(JobInfo<R> jobInfo, T processData) {
            this.jobInfo = jobInfo;
            this.processData = processData;
        }

        @Override
        public void run() {
            R r = null;
            ITaskProcesser<T, R> taskProcesser = (ITaskProcesser<T, R>) jobInfo.getTaskProcesser();
            TaskResult<R> result = null;
            try {
                result = taskProcesser.taskExecute(processData);
                if (result == null) {
                    result = new TaskResult<R>(TaskResultType.EXCEPTION, r, "结果为空");
                }
                if (result.getTaskResultType() == null) {
                    if (result.getReason() == null) {
                        result = new TaskResult<R>(TaskResultType.EXCEPTION, r, "结果为空");
                    } else {
                        result = new TaskResult<R>(TaskResultType.EXCEPTION, r, "结果为空, 原因：" + result.getReason());
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                result = new TaskResult<R>(TaskResultType.EXCEPTION, r, e.getMessage());
            } finally {
                // 江结果添加到延时队列及结果队列，同时计数
                jobInfo.addTaskResult(result);
            }
        }
    }

    /**
     * 调用者提交工作中的任务
     */
    public <T, R> void putTask(String jobName, T t) {
        JobInfo<R> jobInfo = getJob(jobName);
        PendingTask<T, R> task = new PendingTask<>(jobInfo, t);
        taskExecutor.execute(task);
    }

    /**
     * 根据工作名称检索工作
     */
    private <R> JobInfo<R> getJob(String jobName) {
        JobInfo<R> jobInfo = (JobInfo<R>) jobInfoMap.get(jobName);
        if (null == jobInfo) {
            throw new RuntimeException(jobName + "是非法任务！");
        }
        return jobInfo;
    }

    /**
     * 调用者注册工作，如工作名，任务的处理器等等
     */
    public <R> void registerJob(String jobName, int jobLength,
                                ITaskProcesser<?, ?> taskProcesser, long expireTime) {
        JobInfo<R> jobInfo = new JobInfo<R>(jobName, jobLength, taskProcesser, expireTime);
        // putIfAbsent 如果传入key对应的value已经存在，就返回存在的value，不进行替换。如果不存在，就添加key和value，返回null
        if (jobInfoMap.putIfAbsent(jobName, jobInfo) != null) {
            throw new RuntimeException(jobName + "已经注册！");
        }
    }

    /**
     * 获得工作的整体处理进度
     */
    public <R> String getTaskProgess(String jobName) {
        JobInfo<R> jobInfo = getJob(jobName);
        return jobInfo.getTotalProcess();
    }

    /**
     * 获得每个任务的处理详情
     */
    public <R> List<TaskResult<R>> getTaskDetail(String jobName) {
        JobInfo<R> jobInfo = getJob(jobName);
        return jobInfo.getTaskDetail();
    }
}

/**
 * @Author SunnyBear
 * @Description 任务完成后, 在一定的时间供查询结果，之后为释放资源节约内存，需要定期处理过期的任务
 */
public class CheckJobProcesser {

    /**
     * 延时队列，存放处理好的工作任务结果
     */
    private static DelayQueue<Item<String>> queue = new DelayQueue<>();

    private static CheckJobProcesser processer = new CheckJobProcesser();

    private CheckJobProcesser() {
    }

    /**
     * 使用单例
     */
    public static CheckJobProcesser getInstance() {
        return processer;
    }

    /**
     * 处理队列中的到期任务
     */
    public static class FetchJob implements Runnable {
        private static DelayQueue<Item<String>> queue = CheckJobProcesser.queue;
        private static Map<String, JobInfo<?>> jobInfoMap = PendingJobPool.getMap();

        @Override
        public void run() {
            while (true) {
                try {
                    // 从延时队列DelayQueue中获取任务
                    Item<String> item = queue.take();
                    String jobName = item.getData();
                    jobInfoMap.remove(jobName);
                    System.out.println(jobName + " 过期了，从缓存中清除");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 任务完成后，放入队列，经过expireTime时间后，从整个框架中移除
     */
    public void putJob(long expireTime, String jobName) {
        Item<String> item = new Item<>(expireTime, jobName);
        queue.offer(item);
        System.out.println(jobName + " 已经放入过期检查缓存，时长：" + expireTime);
    }

    /**
     * 处理过期任务设置为守护线程
     */
    static {
        Thread thread = new Thread(new FetchJob());
        thread.setDaemon(true);
        thread.start();
        System.out.println("开启过期检查的任务线程。。。");
    }

}

/**
 * @Author SunnyBear
 * @Description 一个实际任务类（即用户要执行的任务类），将数值加上一个随机数，并休眠随机时间，模拟任务执行过程中的情况
 */
public class MyTask implements ITaskProcesser<Integer, Integer> {

    @Override
    public TaskResult<Integer> taskExecute(Integer data) {
        Random r = new Random();
        int flag = r.nextInt(500);
        try {
            Thread.sleep(flag);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (flag <= 300) {
            // 正常处理的情况
            Integer returnValue = data.intValue() + flag;
            return new TaskResult<Integer>(TaskResultType.SUCCESS, returnValue, "success");
        } else if (flag > 301 && flag <= 400) {
            // 处理失败的情况
            return new TaskResult<Integer>(TaskResultType.FAILURE, -1, "failure");
        } else {
            // 发生异常的情况
            try {
                throw new RuntimeException("异常发生了！！");
            } catch (Exception e) {
                return new TaskResult<Integer>(TaskResultType.EXCEPTION,
                        -1, e.getMessage());
            }
        }
    }
}

/**
 * @Author SunnyBear
 * @Description 模拟一个应用程序，提交工作和任务，并查询任务进度
 */
public class AppTest {
    private final static String JOB_NAME = "工作组一";
    /**
     * 一个工作组中的任务数量
     */
    private final static int JOB_LENGTH = 1000;

    /**
     * 查询任务进度的线程
     */
    private static class QueryResult implements Runnable {

        private PendingJobPool pool;

        public QueryResult(PendingJobPool pool) {
            super();
            this.pool = pool;
        }

        @Override
        public void run() {
            int i = 0;
            while (i < 350) {
                List<TaskResult<String>> taskDetail = pool.getTaskDetail(JOB_NAME);
                if (!taskDetail.isEmpty()) {
                    System.out.println("当前进度：" + pool.getTaskProgess(JOB_NAME));
                    System.out.println("任务详情: " + taskDetail);
                }
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                i++;
            }
        }

    }

    public static void main(String[] args) {
        MyTask myTask = new MyTask();
        PendingJobPool pool = PendingJobPool.getInstance();
        pool.registerJob(JOB_NAME, JOB_LENGTH, myTask, 5);
        Random r = new Random();
        for (int i = 0; i < JOB_LENGTH; i++) {
            pool.putTask(JOB_NAME, r.nextInt(1000));
        }
        Thread t = new Thread(new QueryResult(pool));
        t.start();
    }
}