手写并发任务执行框架

刺骨的言语ヽ痛彻心扉 2023-07-21 13:27 11阅读 0赞

### 手写并发任务执行框架 ###

*  目标：
 *  实现：
 *   *   *  思路：
     *  一、框架开发
     *   *  1、添加依赖
         *  2、任务处理器接口
         *  3、超时机制包装类
         *  4、任务信息类
         *  5、任务结果类型
         *  6、任务结果类
         *  7、框架主体处理类
         *  8、过期任务清理类
     *  二、框架测试
     *   *  1、自定义工作任务
         *  2、测试
         *  3、测试结果分析
     *  三、总结
     *   *  流程图
         *  重点及易错点
         *   *  1、避免传递引用
             *  2、清除任务一定要终止线程

# 目标： #

> 利用多线程批量完成用户提交的任务。  
> 要求不同任务相互隔离、不互相干扰。  
> 同时自动清除已完成和过期任务。

# 实现： #

### 思路： ###

首先定义框架主体类，提供注册任务接口供用户调用。  
其次提供提交具体任务接口，使得用户将任务真正提交给并发任务执行框架。  
最后提供结果查询接口，使得用户在并发任务执行框架执行完可以获取结果。

## 一、框架开发 ##

按照上述思路完成并发任务执行框架的基本功能。

### 1、添加依赖 ###

本项目使用`junit`对框架进行测试，为了快速开发也继承了`Lombok`插件，所以需要在`pom.xml`配置文件中添加`junit`和`Lombok`等相关依赖。源码如下：

> pom.xml

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>muti_thread_shizhan</groupId>
        <artifactId>muti_thread_shizhan</artifactId>
        <version>1.0-SNAPSHOT</version>
    
    
        <properties>
            
            <jdk.version>1.8</jdk.version>
    
        </properties>
    
        <dependencies>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.13</version>
            </dependency>
    
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.12</version>
            </dependency>
    
        </dependencies>
    
        <build>
            <plugins>
                <plugin>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>${jdk.version}</source>
                        <target>${jdk.version}</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    
    </project>

### 2、任务处理器接口 ###

定义`ITaskProcesserChen` 类，要求框架使用者实现该任务接口，因为任务的性质在调用时才知道，所以传入的参数和方法的返回值均使用泛型。超时机制中需要解释进程，所以需要抛出中断异常。简单的说就是实现输入数据并返回结果的功能。源码如下：

> ITaskProcesserChen.java

package com.chen.mutithread.vo;
    
    /** * * *类说明：要求框架使用者实现的任务接口，因为任务的性质在调用时才知道， *所以传入的参数和方法的返回值均使用泛型 */
    public interface ITaskProcesserChen<T, R>{ 
    	/** * @param data 调用方法需要使用的业务数据 * @return 方法执行后业务方法的结果 */
    	TaskResult<R> taskExecute(T data) throws InterruptedException;
    }

### 3、超时机制包装类 ###

定义`ItemVo` 类，用于存放任务名称和超时时间，便于延时队列管理。实现`Delayed`接口，引入超时机制。源码如下：

> ItemVo.java

package com.chen.mutithread.vo;
    
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /** * 类说明：存放到队列的元素 */
    public class ItemVo<T> implements Delayed { 
    
        private long activeTime;//到期时间，单位毫秒
        private T date;
    
        //activeTime是个过期时长
        public ItemVo(long activeTime, T date) { 
            super();
            this.activeTime = TimeUnit.NANOSECONDS.convert(activeTime,
                    TimeUnit.MILLISECONDS) + System.nanoTime();//将传入的时长转换为超时的时刻
            this.date = date;
        }
    
        public long getActiveTime() { 
            return activeTime;
        }
    
        public T getDate() { 
            return date;
        }
    
        //按照剩余时间排序
        @Override
        public int compareTo(Delayed o) { 
            long d = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
            return (d == 0) ? 0 : ((d > 0) ? 1 : -1);
        }
    
        //返回元素的剩余时间
        @Override
        public long getDelay(TimeUnit unit) { 
            long d = unit.convert(this.activeTime - System.nanoTime(),
                    TimeUnit.NANOSECONDS);
            return d;
        }
    
    
    }

### 4、任务信息类 ###

定义`JobInfoChen` 类，任务信息类主要包含任务名称、任务个数、任务处理器、成功处理的任务数、已处理的任务数、结果队列、超时时间等基本信息。通过任务名称、任务个数、任务处理器、超时时间对其进行初始化。提供`getSuccessCount()`方法返回成功处理的结果数，注意这里不能直接返回 `successCount` 因为这是一个引用，会导致线程不安全，以下方法同理、 `getTaskProcesserCount()`方法返回当前已处理任务的结果数、`getTaskDetail()`方法返回结果队列中每个任务的处理详情、`addTaskResult()`方法将处理完成的任务结果放到结果队列，如果任务达到工作任务大小，则开始定期清除任务。`getTotalProcess()`方法返回当前并发任务执行框架工作状态。`getTaskProcesser()`方法返回任务处理器。源码如下：

> JobInfoChen.java

package com.chen.mutithread.vo;
    
    import com.chen.mutithread.CheckJobProcesser;
    import com.chen.mutithread.CheckJobProcesserChen;
    
    import java.util.LinkedList;
    import java.util.List;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.atomic.AtomicInteger;
    
    
    /** * 类说明：提交给框架执行的工作实体类,工作：表示本批次需要处理的同性质任务(Task)的一个集合 * 注意：返回给用户的不能是引用，防止线程安全事故发生。即不允许用户修改底层框架使用的东西 */
    public class JobInfoChen<R> { 
        //区分唯一的工作
        private final String jobName;
        //工作的任务个数
        private final int jobLength;
        //这个工作的任务处理器
        private final ITaskProcesserChen<?, ?> taskProcesser;
        //成功处理的任务数
        private final AtomicInteger successCount;
        //已处理的任务数
        private final AtomicInteger taskProcesserCount;
        //拿结果从头拿，放结果从尾部放 保存工作中每个任务的返回结果,只有处理完成成放进去
        // 注意这是一个结果队列
        private final LinkedBlockingDeque<TaskResult<R>> taskDetailQueue;
        //工作的完成保存的时间，超过这个时间从缓存中清除
        private final long expireTime;
    
        //阻塞队列不应该由调用者传入，应该内部生成，长度为工作的任务个数
        public JobInfoChen(String jobName, int jobLength,
                           ITaskProcesserChen<?, ?> taskProcesser,
                           long expireTime) { 
            super();
            this.jobName = jobName;
            this.jobLength = jobLength;
            this.taskProcesser = taskProcesser;
            this.successCount = new AtomicInteger(0);
            this.taskProcesserCount = new AtomicInteger(0);
            this.taskDetailQueue = new LinkedBlockingDeque<>(jobLength);
            this.expireTime = expireTime;
        }
    
        //返回成功处理的结果数
        //不能直接返回 successCount 因为这是一个引用，会导致线程不安全，以下同理
        public int getSuccessCount() { 
            return successCount.get();
        }
    
        //返回当前已处理的结果数
        public int getTaskProcesserCount() { 
            return taskProcesserCount.get();
        }
    
        //提供工作中失败的次数，为了方便调用者使用
        public int getFailCount() { 
            return taskProcesserCount.get() - successCount.get();
        }
    
        //获得工作中每个任务的处理详情
        //注意由于阻塞队列是加锁的，所以不存在写的时候读，多线程冲突
        //由于是框架，不能讲引用传给用户，防止出现线程安全问题
        public List<TaskResult<R>> getTaskDetail() { 
            List<TaskResult<R>> taskList = new LinkedList<>();
            TaskResult<R> taskResult;
            //从阻塞队列中拿任务的结果，反复取，一直取到null为止，说明目前队列中最新的任务结果已经取完，可以不取了
            //既然是任务结果，肯定已经处理完成了，所以全部拿完是对的
            while ((taskResult = taskDetailQueue.pollFirst()) != null) { 
                taskList.add(taskResult);
            }
            return taskList;
        }
    
        //从业务应用角度来说，保证最终一致性即可，不需要对方法加锁.
        //注意 从jvm内存角度，只有传递进来的参数需要操作的时候才需要新建一个变量接受，即栈变量
        public void addTaskResult(TaskResult<R> result, CheckJobProcesserChen checkJob) { 
            if (TaskResultType.Success.equals(result.getResultType())) { 
                successCount.incrementAndGet();
            }
            taskDetailQueue.addLast(result);
            taskProcesserCount.incrementAndGet();
    
            //工作队列满了，放入过期队列
            //这里一般假设用户只是提交 jobLength 个任务到工作框架中，提交完毕定期清除
            if (taskProcesserCount.get() == jobLength) { 
                checkJob.putJob(jobName, expireTime);
            }
    
    // if (taskProcesserCount.get() >= jobLength) { 
    // if (result.getReturnValue() == null) { 
    // System.out.println("chen: " + taskProcesserCount.get());
    // } else { 
    // System.out.println("chen: " + taskProcesserCount.get() + " : " + result.getReturnValue());
    // }
    //
    // }
        }
    
        /** * 返回当前并发任务执行框架工作状态 * @return */
        public String getTotalProcess() { 
            return "Success[" + successCount.get() + "]/Current["
                    + taskProcesserCount.get() + "] Total[" + jobLength + "]";
        }
    
        /** * 返回处理器 * @return */
        public ITaskProcesserChen<?, ?> getTaskProcesser() { 
            return taskProcesser;
        }
    }

### 5、任务结果类型 ###

定义`TaskResultType` 类，主要定义任务处理的3种状态，即：`Success`、`Failure`、`Exception`。源码如下：

> TaskResultType .java

package com.chen.mutithread.vo;
    
    /** * 类说明：方法本身运行是否正确的结果类型 */
    public enum TaskResultType { 
        //方法成功执行并返回了业务人员需要的结果
        Success,
        //方法成功执行但是返回的是业务人员不需要的结果
        Failure,
        //方法执行抛出了Exception
        Exception;
    }

### 6、任务结果类 ###

定义`TaskResult` 类，主要包含任务处理结果类型、任务处理业务结果数据、如果失败了还有失败的原因等基本信息。同时提供这些基本信息的`get()`方法供用户使用。源码如下：

> TaskResult.java

package com.chen.mutithread.vo;
    
    /** * *类说明：任务处理返回结果实体类 */
    public class TaskResult<R> { 
    	//方法本身运行是否正确的结果类型
    	private final TaskResultType resultType;
    	//方法的业务结果数据；
    	private final R returnValue;
    	//这里放方法失败的原因
    	private final String reason;
    	
    	public TaskResult(TaskResultType resultType, R returnValue, String reason) { 
    		super();
    		this.resultType = resultType;
    		this.returnValue = returnValue;
    		this.reason = reason;
    	}
    	
    	//方便业务人员使用，这个构造方法表示业务方法执行成功返回的结果
    	public TaskResult(TaskResultType resultType, R returnValue) { 
    		super();
    		this.resultType = resultType;
    		this.returnValue = returnValue;
    		this.reason = "Success";
    	}
    
    	public TaskResultType getResultType() { 
    		return resultType;
    	}
    
    	public R getReturnValue() { 
    		return returnValue;
    	}
    
    	public String getReason() { 
    		return reason;
    	}
    
    
    	@Override
    	public String toString() { 
    		return "TaskResult{" +
    				"resultType=" + resultType +
    				", returnValue=" + returnValue +
    				", reason='" + reason + '\'' +
    				'}';
    	}
    }

### 7、框架主体处理类 ###

定义`PendingJobPoolChen` 类，主要包含线程数、任务队列、线程池、任务信息容器、线程容器等基本信息。其中任务队列要避免使用无界队列 防止用户无限制上传。使用懒加载设置线程安全的单例模式、限制用户使用的线程数，防止应用崩掉。定义`PendingTask`内部类，对工作中的任务进行包装，提交给线程池使用，并处理任务的结果，写入缓存以供查询。提供`putTask()`方法调用者可以提交工作中的任务到框架中。`registerJob()`方法主要实现调用者注册任务工作，如工作名，任务的处理器等功能,其中名称唯一，相当于ID。`getTaskDetail()`方法返回每个任务的处理详情。`getTaskProgess()`方法返回工作的整体处理进度。源码如下：

> PendingJobPoolChen .java

package com.chen.mutithread;
    
    import com.chen.mutithread.vo.*;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.*;
    
    
    /** * 框架的主体类，也是调用者主要使用的类 */
    public class PendingJobPoolChen { 
    
        //保守估计
        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);
        //job的存放容器
        //G<?> 与 G<? extends Object>等同，如List<?> 与List<? extends Objext>等同。
        private static ConcurrentHashMap<String, JobInfoChen<?>> jobInfoMap
                = new ConcurrentHashMap<>();
    
        //线程和处理任务对应
        private static ConcurrentHashMap<String, List<?>> threadMap = new ConcurrentHashMap<>();
    // private static ConcurrentHashMap<String, List<?>> threadMap2 = new ConcurrentHashMap<>();
    
    
        public static ConcurrentHashMap<String, List<?>> getThreadMap() { 
            return threadMap;
        }
    
        private static CheckJobProcesserChen checkJob
                = CheckJobProcesserChen.getInstance();
    
        public static Map<String, JobInfoChen<?>> getMap() { 
            return jobInfoMap;
        }
    
        //单例模式------注意：限制用户使用的线程数，防止应用崩掉
        private PendingJobPoolChen() { 
        }
    
        //懒加载式的线程安全
        private static class JobPoolHolder { 
            public static PendingJobPoolChen pool = new PendingJobPoolChen();
        }
    
        public static PendingJobPoolChen getInstance() { 
            return JobPoolHolder.pool;
        }
        //单例模式------
    
        //对工作中的任务进行包装，提交给线程池使用，并处理任务的结果，写入缓存以供查询
        public static class PendingTask<T, R> implements Runnable { 
    
            private JobInfoChen<R> jobInfo;
            private T processData;
            private Thread currentThread;
    
            public PendingTask(JobInfoChen<R> jobInfo, T processData) { 
                super();
                this.jobInfo = jobInfo;
                this.processData = processData;
                this.currentThread = Thread.currentThread();
            }
    
            public void quit() { 
                this.currentThread.interrupt();
            }
    
            //告诉编译器忽略指定的警告，不用在编译完成后出现警告信息。
            @SuppressWarnings("unchecked")
            @Override
            public void run() { 
                R r = null;
                ITaskProcesserChen<T, R> taskProcesser =
                        (ITaskProcesserChen<T, R>) jobInfo.getTaskProcesser();
                TaskResult<R> result = null;
    
                //用户的代码是最不可信的，所以防止用户程序抛出异常，需要try catch处理
                try { 
                    //调用业务人员实现的具体方法
                    result = taskProcesser.taskExecute(processData);
                    //要做检查，防止开发人员处理不当
                    if (result == null) { 
                        result = new TaskResult<R>(TaskResultType.Exception, r,
                                "result is null");
                    }
                    if (result.getResultType() == null) { 
                        if (result.getReason() == null) { 
                            result = new TaskResult<R>(TaskResultType.Exception, r, "reason is null");
                        } else { 
                            result = new TaskResult<R>(TaskResultType.Exception, r,
                                    "result is null,but reason:" + result.getReason());
                        }
                    }
                } catch (InterruptedException e) { 
                    System.out.println("线程超时终止！");
                } catch (Exception e) { 
                    e.printStackTrace();
                    result = new TaskResult<R>(TaskResultType.Exception, r,
                            e.getMessage());
                } finally { 
                    jobInfo.addTaskResult(result, checkJob);
                }
            }
        }
    
        //根据工作名称检索工作
        @SuppressWarnings("unchecked")
        private <R> JobInfoChen<R> getJob(String jobName) { 
            JobInfoChen<R> jobInfo = (JobInfoChen<R>) jobInfoMap.get(jobName);
            if (null == jobInfo) { 
                throw new RuntimeException(jobName + "是个非法任务、或该任务已经过期！");
            }
            return jobInfo;
        }
    
        public <T, R> void putTaskAndThread(String jobName, PendingTask<T, R> task) { 
            ArrayList<PendingTask<T, R>> pendingTasks = new ArrayList<>();
            pendingTasks.add(task);
            //添加任务，将任务和线程绑定
            List<PendingTask<T, R>> objects = (List<PendingTask<T, R>>) threadMap.putIfAbsent(jobName, pendingTasks);
            if (objects != null) { 
                objects.addAll(pendingTasks);
                // 叠加到已有任务末尾
                threadMap.put(jobName, objects);
            }
    
        }
    
        //调用者提交工作中的任务
        public <T, R> void putTask(String jobName, T t) { 
            JobInfoChen<R> jobInfo = getJob(jobName);
            PendingTask<T, R> task = new PendingTask<>(jobInfo, t);
    
            //添加任务，将任务和线程绑定
            putTaskAndThread(jobName, task);
    
            taskExecutor.execute(task);
    // Future<?> future = taskExecutor.submit(task);
        }
    
        //调用者注册工作，如工作名，任务的处理器等等,名称唯一，相当于ID
        public <R> void registerJob(String jobName, int jobLength,
                                    ITaskProcesserChen<?, ?> taskProcesser, long expireTime) { 
            JobInfoChen<R> jobInfo = new JobInfoChen<>(jobName, jobLength, taskProcesser, expireTime);
            // 不能使用put,防止任务在处理的时候，用户修改map,导致处理异常
            if (jobInfoMap.putIfAbsent(jobName, jobInfo) != null) { 
                throw new RuntimeException(jobName + "已经注册了！");
            }
        }
    
        //获得每个任务的处理详情
        public <R> List<TaskResult<R>> getTaskDetail(String jobName) { 
            JobInfoChen<R> jobInfo = getJob(jobName);
            return jobInfo.getTaskDetail();
        }
    
        //获得工作的整体处理进度
        public <R> String getTaskProgess(String jobName) { 
            JobInfoChen<R> jobInfo = getJob(jobName);
            return jobInfo.getTotalProcess();
        }
    
    }

### 8、过期任务清理类 ###

定义`CheckJobProcesserChen` 类，实现名称和到期时间绑定，将过期任务定期清除，注意：这里清除时一定要终止工作线程，释放资源。源码如下：

> CheckJobProcesserChen .java

package com.chen.mutithread;
    
    import com.chen.mutithread.vo.ItemVo;
    
    import java.util.List;
    import java.util.concurrent.DelayQueue;
    
    
    /** * 类说明：任务完成后,在一定的时间供查询，之后为释放资源节约内存，需要定期处理过期的任务 */
    public class CheckJobProcesserChen { 
        //DelayQueue是一个无界阻塞队列，只有在延迟期满时才能从中提取元素
        private static DelayQueue<ItemVo<String>> queue
                = new DelayQueue<>();//存放已完成任务等待过期的队列
    
        //单例模式------
        private CheckJobProcesserChen() { 
        }
    
        private static class ProcesserHolder { 
            public static CheckJobProcesserChen processer = new CheckJobProcesserChen();
        }
    
        public static CheckJobProcesserChen getInstance() { 
            return ProcesserHolder.processer;
        }
        //单例模式------
    
        //处理队列中到期任务的实行
        private static class FetchJob implements Runnable { 
    
            @Override
            public void run() { 
                while (true) { 
                    try { 
                        //拿到已经过期的任务 take()是阻塞方法
                        ItemVo<String> item = queue.take();
                        String jobName = item.getDate();
                        //删除工作信息
                        PendingJobPoolChen.getMap().remove(jobName);
                        //中断正在工作的任务线程
                        shutDownThread(jobName);
                        System.out.println(jobName + " is out of date,remove from map!");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        public static <T, R> void shutDownThread(String jobName) { 
            List<?> objects = PendingJobPoolChen.getThreadMap().get(jobName);
            if (objects == null) { 
                return;
            }
    
            List<PendingJobPoolChen.PendingTask<T, R>> threads = (List<PendingJobPoolChen.PendingTask<T, R>>) objects;
    
            for (PendingJobPoolChen.PendingTask<T, R> thread : threads) { 
                // 逐个终止，即使线程运行结束也没有影响
                thread.quit();
            }
            //删除
            PendingJobPoolChen.getThreadMap().remove(jobName);
    
        }
    
        /*任务完成后，放入队列，经过expireTime时间后，从整个框架中移除*/
        public void putJob(String jobName, long expireTime) { 
            //实现名称和到期时间绑定
            ItemVo<String> item = new ItemVo<String>(expireTime, jobName);
            queue.offer(item);
            System.out.println("Job[" + jobName + "已经放入了过期检查缓存，过期时长：" + expireTime);
        }
    
        static { 
            Thread thread = new Thread(new FetchJob());
            thread.setDaemon(true);
            thread.start();
            System.out.println("开启任务过期检查守护线程................");
        }
    
    
    }

## 二、框架测试 ##

将自定义任务批量提交给并发任务执行框架，并查看任务处理结果。

### 1、自定义工作任务 ###

定义`MyTaskChen` 类，实现将数值加上一个随机数，并休眠随机时间的功能，模拟用户提交的任务。源码如下：

> MyTaskChen .java

package com.chen.mutithread.demo;
    
    import com.chen.mutithread.tools.SleepTools;
    import com.chen.mutithread.vo.ITaskProcesser;
    import com.chen.mutithread.vo.ITaskProcesserChen;
    import com.chen.mutithread.vo.TaskResult;
    import com.chen.mutithread.vo.TaskResultType;
    
    import java.util.Random;
    
    /** * * <p> * 类说明：一个实际任务类，将数值加上一个随机数，并休眠随机时间 */
    public class MyTaskChen implements ITaskProcesserChen<Integer, Integer> { 
    
        @Override
        public TaskResult<Integer> taskExecute(Integer data) { 
            Random r = new Random();
            int flag = r.nextInt(500);
            SleepTools.ms(flag);
            if (flag <= 300) { //正常处理的情况
                Integer returnValue = data.intValue() + flag;
                return new TaskResult<Integer>(TaskResultType.Success, returnValue);
            } 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());
                }
            }
        }
    
    }

### 2、测试 ###

定义`AppTestChen` 类，实现向并发任务执行框架提交`1000`个任务，并通过QueryResult线程查询框架执行结果。源码如下：

> AppTestChen .java

package com.chen.mutithread.demo;
    
    import com.chen.mutithread.PendingJobPool;
    import com.chen.mutithread.PendingJobPoolChen;
    import com.chen.mutithread.vo.TaskResult;
    
    import java.util.List;
    import java.util.Random;
    
    /** * 类说明：模拟一个应用程序，提交工作和任务，并查询任务进度 */
    public class AppTestChen { 
    
        private final static String JOB_NAME = "计算数值";
        private final static int JOB_LENGTH = 1000;
    
        //查询任务进度的线程
        private static class QueryResult implements Runnable { 
    
            private PendingJobPoolChen pool;
    
            public QueryResult(PendingJobPoolChen pool) { 
                super();
                this.pool = pool;
            }
    
            @Override
            public void run() { 
                int num = 0;
                List<TaskResult<String>> taskDetail;
                try { 
                    while ((taskDetail = pool.getTaskDetail(JOB_NAME)) != null) { 
    
                        if (!taskDetail.isEmpty()) { 
                            System.out.println(pool.getTaskProgess(JOB_NAME));
                            System.out.println(taskDetail);
                            num += taskDetail.size();
                        }
                    }
                } catch (Exception e) { 
    
                }
    
                System.out.println("chen num : " + num);
            }
    
        }
    
        public static void main(String[] args) { 
            MyTaskChen myTask = new MyTaskChen();
            //拿到框架的实例
            PendingJobPoolChen pool = PendingJobPoolChen.getInstance();
            //注册job
            pool.registerJob(JOB_NAME, JOB_LENGTH, myTask, 1000 * 5);
            Random r = new Random();
            // 假定用户不按套路出牌
            for (int i = 0; i < JOB_LENGTH; i++) { 
                //依次推入Task
                pool.putTask(JOB_NAME, r.nextInt(1000));
            }
            Thread t = new Thread(new QueryResult(pool));
            t.start();
        }
    }

### 3、测试结果分析 ###

中间大部分重复信息被我删除。在程序运行一开始，我们就开启了任务过期检查守护线程。当任务执行完毕，并且时间过期，我们发现任务被成功清除。

> Console

开启任务过期检查守护线程................
    Success[1]/Current[1] Total[1000]
    [TaskResult{ resultType=Success, returnValue=462, reason='Success'}]
    Success[2]/Current[2] Total[1000]
    [TaskResult{ resultType=Success, returnValue=620, reason='Success'}]
    Success[3]/Current[3] Total[1000]
    [TaskResult{ resultType=Success, returnValue=1081, reason='Success'}]
    ...
    [TaskResult{ resultType=Exception, returnValue=-1, reason='异常发生了！！'}]
    Success[579]/Current[981] Total[1000]
    [TaskResult{ resultType=Failure, returnValue=-1, reason='Failure'}]
    Success[580]/Current[982] Total[1000]
    [TaskResult{ resultType=Success, returnValue=541, reason='Success'}]
    Success[580]/Current[983] Total[1000]
    [TaskResult{ resultType=Failure, returnValue=-1, reason='Failure'}]
    Success[580]/Current[984] Total[1000]
    [TaskResult{ resultType=Exception, returnValue=-1, reason='异常发生了！！'}]
    Success[580]/Current[985] Total[1000]
    [TaskResult{ resultType=Exception, returnValue=-1, reason='异常发生了！！'}]
    Success[581]/Current[986] Total[1000]
    [TaskResult{ resultType=Success, returnValue=612, reason='Success'}]
    Success[582]/Current[987] Total[1000]
    [TaskResult{ resultType=Success, returnValue=367, reason='Success'}]
    Success[583]/Current[988] Total[1000]
    [TaskResult{ resultType=Success, returnValue=894, reason='Success'}]
    Success[583]/Current[989] Total[1000]
    [TaskResult{ resultType=Exception, returnValue=-1, reason='异常发生了！！'}]
    Success[584]/Current[990] Total[1000]
    [TaskResult{ resultType=Success, returnValue=209, reason='Success'}]
    Success[584]/Current[991] Total[1000]
    [TaskResult{ resultType=Failure, returnValue=-1, reason='Failure'}]
    Success[585]/Current[992] Total[1000]
    [TaskResult{ resultType=Success, returnValue=698, reason='Success'}]
    Success[585]/Current[993] Total[1000]
    [TaskResult{ resultType=Failure, returnValue=-1, reason='Failure'}]
    Success[586]/Current[994] Total[1000]
    [TaskResult{ resultType=Success, returnValue=373, reason='Success'}]
    Success[586]/Current[995] Total[1000]
    [TaskResult{ resultType=Exception, returnValue=-1, reason='异常发生了！！'}]
    Success[587]/Current[996] Total[1000]
    [TaskResult{ resultType=Success, returnValue=538, reason='Success'}]
    Success[587]/Current[997] Total[1000]
    [TaskResult{ resultType=Failure, returnValue=-1, reason='Failure'}]
    Success[587]/Current[998] Total[1000]
    [TaskResult{ resultType=Exception, returnValue=-1, reason='异常发生了！！'}]
    Success[588]/Current[999] Total[1000]
    [TaskResult{ resultType=Success, returnValue=733, reason='Success'}]
    Success[588]/Current[1000] Total[1000]
    [TaskResult{ resultType=Failure, returnValue=-1, reason='Failure'}]
    Job[计算数值已经放入了过期检查缓存，过期时长：5000
    chen num : 1000
    计算数值 is out of date,remove from map!

## 三、总结 ##

大致说下流程，基本就是用户先获取框架实例、然后通过框架实例注册工作任务Job、接着批量提交任务给框架、框架接受到该任务之后封装成线程类以便在线程池中批量运行、然后任务处理完成将处理结果放到结果队列、处理的同时用户也可以查询任务处理结果、如果任务处理达到处理阈值则放入延时队列定期清除。下面是流程图，配合理解。

### 流程图 ###

Created with Raphaël 2.2.0开始拿到框架的实例并注册Job将任务推入Task框架获取到该task，并封装成线程类PendingTask以提交给线程池线程池接受并处理该任务任务处理完成，放入结果队列是否已经达到处理数量上限?将任务加入过期删除队列,并超时删除结束yesno

### 重点及易错点 ###

#### 1、避免传递引用 ####

//返回成功处理的结果数
        public int getSuccessCount() { 
            return successCount.get();
        }

上述代码不能直接返回 `successCount` 因为这是一个引用，会导致线程不安全的问题。

#### 2、清除任务一定要终止线程 ####

//处理队列中到期任务的实行
        private static class FetchJob implements Runnable { 
    
            @Override
            public void run() { 
                while (true) { 
                    try { 
                        //拿到已经过期的任务 take()是阻塞方法
                        ItemVo<String> item = queue.take();
                        String jobName = item.getDate();
                        //删除工作信息
                        PendingJobPoolChen.getMap().remove(jobName);
                        //中断正在工作的任务线程
                        shutDownThread(jobName);
                        System.out.println(jobName + " is out of date,remove from map!");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        public static <T, R> void shutDownThread(String jobName) { 
            List<?> objects = PendingJobPoolChen.getThreadMap().get(jobName);
            if (objects == null) { 
                return;
            }
    
            List<PendingJobPoolChen.PendingTask<T, R>> threads = (List<PendingJobPoolChen.PendingTask<T, R>>) objects;
    
            for (PendingJobPoolChen.PendingTask<T, R> thread : threads) { 
                // 逐个终止，即使线程运行结束也没有影响
                thread.quit();
            }
            //删除
            PendingJobPoolChen.getThreadMap().remove(jobName);
    
        }

清除任务一定要终止线程，同时一定要将该任务在线程容器中删除，便于`JVM`优化。

写小轮子太烧脑了，有问题欢迎各位读者批评指正。

点个赞再走呗！欢迎留言哦！