当前位置：
java 多线程处理任务，返回带进度条结果 封装为工具类
机器助手 | 2025-07-13
100人浏览
在我们写业务代码中，经常会想到多线程处理复杂耗时任务，但又想返回处理进度，下面代码就是实现这个业务。
以下代码是执行两个任务，分别输出两个任务的实时进度和最后结果，包括定时从缓存移除
类说明
需自定义类：
- MyRun---------------程序测试入口
- MyTask--------------具体需执行业务类
- RequestData--------请求参数实体
- ResponseRet-------返回结果实体
框架类（无需修改，直接用）：
- PendingJobPool-----------框架的主体类，也是调用者主要使用的类
- CheckJobProcesser------任务完成后,在一定的时间供查询，之后为释放资源节约内存，需要定期处理过期的任务
- ITaskProcesser-------------要求框架使用者实现的任务接口，因为任务的性质在调用时才知道
- ItemDelayed----------------存放到延时队列的元素
- JobInfo-----------------------任务信息实体
- ResultEnum-----------------返回结果枚举类
- TaskResult------------------任务处理返回结果实体类
MyRun类：
/**
* 类说明：模拟一个应用程序，提交工作和任务，并查询任务进度
* @author hyun
*
*/
public class MyRun {
public static void main(String[] args) throws Exception {
//模拟两个任务工作场景（例如两个人同时进行文件校验）
MyTask task = new MyTask();
String taskName = "test-1";
String taskNameTwo = "test-2";
//每个任务需要50个子任务（比如一个文件校验 可分为多个线程去校验）
int taskLength = 50;
//过期时间(比如页面需要查询校验进度，当页面不小心关了 可直接根据任务名称直接查询 不用再次校验，这个过期时间就是当当前未过期时能查询到)
int expiredTime =10;
//拿到框架的实例
PendingJobPool<ResponseRet, RequestData> pool = PendingJobPool.getPendingTaskPool(ResponseRet.class, RequestData.class);
PendingJobPool<ResponseRet, RequestData> poolTwo = PendingJobPool.getPendingTaskPool(ResponseRet.class, RequestData.class);
//注册job
pool.registerTask(taskLength, taskName, task, expiredTime, TimeUnit.SECONDS);
poolTwo.registerTask(taskLength, taskNameTwo, task, expiredTime, TimeUnit.SECONDS);
for(int i = 0; i < taskLength; i++) {
RequestData data = new RequestData();
data.setId(UUID.randomUUID().toString());
data.setName("name1-" + i);
//依次推入Task
pool.putTask(taskName, data);
}
for(int i = 0; i < taskLength; i++) {
RequestData data = new RequestData();
data.setId(UUID.randomUUID().toString());
data.setName("name2-" + i);
pool.putTask(taskNameTwo, data);
}
//以下为查询结果操作
Thread readData = new Thread(()->{
for(;;) {
JobInfo<ResponseRet, RequestData> result = pool.getJobInfoMap().get(taskName);
JobInfo<ResponseRet, RequestData> resultTwo = pool.getJobInfoMap().get(taskNameTwo);
System.out.println(resultTwo.getTaskName() + ":" + resultTwo.toString());
System.out.println(result.getTaskName() + ":" + result.toString());
if((result.sucessNum.get() + result.failNum.get()) >= taskLength &&
(resultTwo.sucessNum.get() + resultTwo.failNum.get()) >= taskLength) {
System.out.println(result.getTaskName() + ":" + result.getBlokingDeque().size() + ":" + result.getBlokingDequeDetail());
System.out.println(resultTwo.getTaskName() + ":" + resultTwo.getBlokingDeque().size() + ":" + resultTwo.getBlokingDequeDetail());
break;
}
Random random = new Random();
try {
Thread.sleep(random.nextInt(30));
} catch (Exception e) {
e.printStackTrace();
}
}
});
readData.start();
}
}
MyTask类：
/**
* 类说明：一个实际任务类，将数值加上一个随机数，并休眠随机时间 根据实际业务需修改类
* @author hyun
*
*/
public class MyTask implements ITaskProcesser<ResponseRet, RequestData>{
@Override
public TaskResult<ResponseRet> taskExecute(RequestData reqData) {
TaskResult<ResponseRet> result = new TaskResult<ResponseRet>();
String id = reqData.getId();
String name = reqData.getName();
Random rand = new Random();
int randNum = rand.nextInt(100);
//模拟实际操作 随机休眠100毫秒以内
try {
Thread.sleep(randNum);
//如果20>随机数>=0 当做失败
if(20 > randNum && randNum >= 0) {
result.setResult(ResultEnum.FAIL);
result.setReason("失败结果");
} else if(randNum >= 90) {//如果随机数>=90 当异常处理
result.setResult(ResultEnum.EXCEPTION);
result.setReason("异常信息");
} else {//其它情况当成功处理
result.setResult(ResultEnum.SUCESS);
result.setReason("Sucess");
ResponseRet responseRet = new ResponseRet();
responseRet.setResult(id);
result.setData(responseRet);
}
} catch (InterruptedException e) {
//异常处理
e.printStackTrace();
result.setResult(ResultEnum.EXCEPTION);
result.setReason(e.getMessage());
}
return result;
}
}
RequestData类：
/**
* 请求参数实体 根据业务需修改类
* @author hyun
*
*/
public class RequestData {
private String id;
private String name;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public String toString() {
return "id:" + id + ",name:" + name;
}
}
ResponseRet类：
/**
* 返回结果实体 根据需要业务需修改类
* @author hyun
*
*/
public class ResponseRet {
private String result;
public String getResult() {
return result;
}
public void setResult(String result) {
this.result = result;
}
@Override
public String toString() {
return "[result:" + result + "]";
}
}
PendingJobPool类：
/**
* 框架的主体类，也是调用者主要使用的类
* @author hyun
*
* @param <R> 返回结果类型
* @param <T> 参数类型
*/
public class PendingJobPool<R, T> {
/**
* cpu核心数
*/
private final static int PROCESSOR_NUM = Runtime.getRuntime().availableProcessors();
/**
* 初始化线程数
*/
private static int INIT_THREAD_NUM = 2;
/**
* keepAliveTime when the number of threads is greater than
* the core, this is the maximum time that excess idle threads
* will wait for new tasks before terminating
*/
private static int ALIVE_TIME = 5;
/**
* 存放任务结果双端队列
*/
private static BlockingQueue<Runnable> bq = new ArrayBlockingQueue<>(2000);
/**
* 线程池，固定大小，有界队列
*/
private Executor executor = new ThreadPoolExecutor(INIT_THREAD_NUM, PROCESSOR_NUM, ALIVE_TIME, TimeUnit.SECONDS, bq);
/**
* job的存放容器
*/
public ConcurrentHashMap<String, JobInfo<R, T>> jobInfoMap = new ConcurrentHashMap<>();
/**
* 单例模式
* @author hyun
*
* @param <R> 结果类型
* @param <T> 参数类型
*/
private static class InitPendingSingleton<R, T>{
public static PendingJobPool<?,?> pendingTaskPool = new PendingJobPool<>();
}
@SuppressWarnings("unchecked")
public static <E,K>PendingJobPool<E,K> getPendingTaskPool(Class<?> E, Class<?> K) {
return (PendingJobPool<E,K>)InitPendingSingleton.pendingTaskPool;
}
/**
* 调用者注册工作，如工作名，任务的处理器等等
* @param taskLength
* @param taskName
* @param task
* @param expiredTime
* @param unit
* @throws Exception
*/
public void registerTask(int taskLength, String taskName, ITaskProcesser<R, T> task, long expiredTime, TimeUnit unit) throws Exception {
JobInfo<R, T> jobInfo = new JobInfo<R, T>(taskLength, taskName, task, expiredTime, unit);
JobInfo<R, T> oldJobInfo = this.jobInfoMap.putIfAbsent(jobInfo.getTaskName(), jobInfo);
if(oldJobInfo != null) {
throw new Exception("任务名称已存在");
}
}
/**
* 对工作中的任务进行包装，提交给线程池使用，并处理任务的结果，写入缓存以供查询
* @author hyun
*
* @param <R> 结果类型
* @param <T> 参数类型
*/
private static class TaskThread<R, T> implements Runnable {
final T data;
private JobInfo<R, T> jobInfo;
public TaskThread(JobInfo<R, T> jobInfo, T data) {
this.data = data;
this.jobInfo = jobInfo;
}
@Override
public void run() {
ITaskProcesser<R, T> taskProcesser = jobInfo.getTaskProcesser();
TaskResult<R> result;
try {
//调用业务人员实现的具体方法
result = taskProcesser.taskExecute(data);
//要做检查，防止开发人员处理不当
if (result == null) {
result = new TaskResult<R>(ResultEnum.EXCEPTION, "result is null");
}
if(ResultEnum.SUCESS.equals(result.getResult())) {
jobInfo.sucessNum.getAndIncrement();
} else {
jobInfo.failNum.getAndIncrement();
}
} catch (Exception e1) {
e1.printStackTrace();
jobInfo.failNum.getAndIncrement();
result = new TaskResult<R>(ResultEnum.EXCEPTION, e1.getMessage());
}
try {
jobInfo.getBlokingDeque().putLast(result);
if((jobInfo.sucessNum.get() + jobInfo.failNum.get()) == jobInfo.getTaskLength() ) {
ItemDelayed<String> delayed = new ItemDelayed<>(jobInfo.getExpiredTime(), jobInfo.getUnit(), jobInfo.getTaskName());
CheckJobProcesser.bq.put(delayed);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 调用者提交工作中的任务
* @param name
* @param data
*/
public void putTask(String name, T data) {
JobInfo<R, T> jobInfo = this.jobInfoMap.get(name);
executor.execute(new TaskThread<>(jobInfo, data));
}
public ConcurrentHashMap<String, JobInfo<R, T>> getJobInfoMap() {
return jobInfoMap;
}
public void setJobInfoMap(ConcurrentHashMap<String, JobInfo<R, T>> jobInfoMap) {
this.jobInfoMap = jobInfoMap;
}
}
CheckJobProcesser类：
/**
* 类说明：任务完成后,在一定的时间供查询，之后为释放资源节约内存，需要定期处理过期的任务
* @author hyun
*
*/
public class CheckJobProcesser {
public static BlockingQueue<ItemDelayed<String>> bq = new DelayQueue<ItemDelayed<String>>();
private static class InitCheckJobProcesser {
public static CheckJobProcesser checkJobProcesser = new CheckJobProcesser();
}
/**
* 单例模式
* @return
*/
public static CheckJobProcesser getCheckJobProcesser() {
return InitCheckJobProcesser.checkJobProcesser;
}
static {
Thread thread = new Thread(new CheckJobThread());
thread.start();
}
/**
* 处理队列中到期任务的执行
* @author root
*
*/
private static class CheckJobThread implements Runnable {
@Override
public void run() {
try {
System.out.println("检查任务过期线程启动");
for(;;) {
ItemDelayed<String> removeObj = bq.take();
String taskName = removeObj.getData();
PendingJobPool<Object,Object> pendingTaskPool = PendingJobPool.getPendingTaskPool(Object.class, Object.class);
pendingTaskPool.getJobInfoMap().remove(taskName);
System.out.println("任务被移除：" + taskName);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
ITaskProcesser类：
/**
*类说明：要求框架使用者实现的任务接口，因为任务的性质在调用时才知道，
*所以传入的参数和方法的返回值均使用泛型
* @author hyun
*
* @param <R> 返回结果类型
* @param <T> 参数类型
*/
public interface ITaskProcesser<R, T> {
/**
* @param data 调用方法需要使用的业务数据
* @return 方法执行后业务方法的结果
*/
public TaskResult<R> taskExecute(T data);
}
ItemDelayed类：
/**
* 存放到延时队列的元素
* @author hyun
*
* @param <T> 放入的数据类型
*/
public class ItemDelayed<T> implements Delayed {
/**
* 到期时间
*/
private long activeTime;
/**
* 数据实体
*/
private T data;
public ItemDelayed(long activeTime, TimeUnit unit, T data) {
this.data = data;
//将传入的时长转换为超时的时刻
this.activeTime = TimeUnit.NANOSECONDS.convert(activeTime, unit) + System.nanoTime();//将传入的时长转换为超时的时刻
}
@Override
public int compareTo(Delayed o) {
long diffTime = this.getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
return diffTime > 0 ? 1 : ((diffTime == 0 ? 0 : -1));
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(this.activeTime - System.nanoTime(), TimeUnit.NANOSECONDS);
}
public T getData() {
return data;
}
public void setData(T data) {
this.data = data;
}
}
JobInfo类：
/**
* 任务信息实体
* @author hyun
*
* @param <R> 返回结果类型
* @param <T> 参数类型
*/
public class JobInfo<R, T> {
//工作的任务个数
private final int taskLength;
//区分唯一的工作
private final String taskName;
//成功处理的任务数
public final AtomicInteger sucessNum = new AtomicInteger(0);
//处理失败的任务数
public final AtomicInteger failNum = new AtomicInteger(0);
//这个工作的任务处理器
private final ITaskProcesser<R, T> taskProcesser;
//结果队列，拿结果从头拿，放结果从尾部放
private final LinkedBlockingDeque<TaskResult<R>> blokingDeque;
//工作的完成保存的时间，超过这个时间从缓存中清除
private final long expiredTime;
//过期时间单位
private final TimeUnit unit;
/**
* 阻塞队列不应该由调用者传入，应该内部生成，长度为工作的任务个数
* @param taskLength
* @param taskName
* @param taskProcesser
* @param expiredTime
* @param unit
*/
public JobInfo(int taskLength, String taskName, ITaskProcesser<R, T> taskProcesser, long expiredTime, TimeUnit unit) {
super();
this.taskLength = taskLength;
this.taskName = taskName;
this.sucessNum.set(0);
this.failNum.set(0);
this.taskProcesser = taskProcesser;
this.blokingDeque = new LinkedBlockingDeque<TaskResult<R>>(taskLength);
this.expiredTime = expiredTime;
this.unit = unit;
}
/**
* 返回当前任务明细的结果 根据实际修改
* @return
*/
public String getBlokingDequeDetail() {
StringBuffer sb = new StringBuffer();
for(;;) {
TaskResult<R> result = this.blokingDeque.pollFirst();
if(result == null) {
break;
}
sb.append(result.toString());
}
return sb.toString();
}
@Override
public String toString() {
return "Success[" + sucessNum.get() + "]/Fail[" + failNum.get() + "] Total[" + taskLength + "]";
}
public int getTaskLength() {
return taskLength;
}
public String getTaskName() {
return taskName;
}
public ITaskProcesser<R, T> getTaskProcesser() {
return taskProcesser;
}
public LinkedBlockingDeque<TaskResult<R>> getBlokingDeque() {
return blokingDeque;
}
public long getExpiredTime() {
return expiredTime;
}
public TimeUnit getUnit() {
return unit;
}
}
ResultEnum类：
/**
* 返回结果枚举类
* @author hyun
*
*/
public enum ResultEnum {
//方法成功执行并返回了业务人员需要的结果
SUCESS,
//方法成功执行但是返回的是业务人员不需要的结果
FAIL,
//方法执行抛出了Exception
EXCEPTION;
}
TaskResult类：
/**
* 任务处理返回结果实体类
* @author root
*
* @param <R> 返回结果类型
*/
public class TaskResult<R> {
/**
* 方法的业务结果数据；
*/
private R data;
/**
* 这里放方法失败的原因
*/
private String reason;
/**
* 方法本身运行是否正确的结果类型
*/
private ResultEnum result;
public TaskResult() {
}
public TaskResult(R data, String reason, ResultEnum result) {
super();
this.data = data;
this.reason = reason;
this.result = result;
}
public TaskResult(ResultEnum result, String reason) {
super();
this.reason = reason;
this.result = result;
}
/**
* 方便业务人员使用，这个构造方法表示业务方法执行成功返回的结果
* @param data
*/
public TaskResult(R data) {
super();
this.data = data;
this.reason = "sucess";
this.result = ResultEnum.SUCESS;
}
@Override
public String toString() {
return "[result:" + result.name() + ",reason:" + reason + ",data:" + (data != null ? data.toString() : "") + "] ";
}
public String getReason() {
return reason;
}
public void setReason(String reason) {
this.reason = reason;
}
public ResultEnum getResult() {
return result;
}
public void setResult(ResultEnum result) {
this.result = result;
}
public void setData(R data) {
this.data = data;
}
public R getData() {
return data;
}
}