java 并发框架_纯Java——简易高并发框架(示例代码)

转自：https://blog.csdn.net/MonkeyDCoding/article/details/81369610

0.源代码

github-简易高并发框架

注：本篇博客知识来自于网课。

1.问题来源以及w

对于一个题库系统。考试组要有批量的离线文档要生成。题库组批量的题目要进行排重，且要根据条件批量修改题目内容。对于

痛点：

批量任务完成缓慢

所有的问题都围绕着“查询”，即查询进度影响总体性能

我们希望尽量使用友好(如果用多线程来提高性能，我们希望能屏蔽细节)

因此我们需要一个可以提供查询进度通用的框架。

2.我们该怎么做？

这里先要明确“任务”(Task)和“工作”(Job)的关系。对于一个工作，他内部可能须有许多的任务，任务是他的子元素(属性、字段)。

用并发安全的类确保每个工作的属性和工作下的每个任务信息，也意味着工作和任务的注册机制。

需要并发安全的类保存每个任务的处理结果(TaskResult)。

需要提供查询接口，供外部的使用。

这里我们不处理对于工作的检查。有兴趣的可以实现。

3.总体流程

这里不按照流程讲解，而是按照类关系从下而上讲解。

4.目录结构

5.TaskResultType

package me.hcFramework.pool.vo;

//这个类只是用来作为标志的信息。

public enum TaskResultType {

SUCCESS, //表示任务成功

FAILSURE, //表示任务失败

EXCEPTION; //表示发生了异常，这里我们不去详尽判断，只用这个标示来笼统表示

}

6.TaskResult

package me.hcFramework.pool.vo;

/**

*

* @param 业务方法处理后的业务结果数据的类型

*

* 对属性使用final修饰是为了使其不可改

*/

public class TaskResult {

//用户业务是否成功完成

private final TaskResultType resultType;

//业务方法处理后的业务结果数据

private final R returnType;

//如果失败，则失败原因

private final String reason;

//针对任务失败的构造方法

public TaskResult(TaskResultType resultType , R returnType , String reason) {

this.resultType = resultType;

this.returnType = returnType;

this.reason = reason;

}

//针对任务成功的构造方法

public TaskResult(TaskResultType resultType , R returnType) {

this.resultType = resultType;

this.returnType = returnType;

this.reason = "success";

}

//因为我们希望字段不可改，设置为了final。所以只提供getters

public TaskResultType getResultType() {

return resultType;

}

public R getReturnType() {

return returnType;

}

public String getReason() {

return reason;

}

@Override

public String toString() {

return "TaskResult [resultType=" + resultType + ", returnType=" + returnType + ", reason=" + reason + "]";

}

}

在这里其实可以发生一点小改动。即：把错误信息归并到TaskResultType中。这样一个TaskResultType包括成功，错误/异常以及其原因就完整了。这里不过多介绍。

7.JobInfo

package me.hcFramework.pool.vo;

import java.util.LinkedList;

import java.util.List;

import java.util.concurrent.LinkedBlockingDeque;

import java.util.concurrent.atomic.AtomicInteger;

/**

* 可以看作是一堆Task的打包+信息控制

* 与TaskResult一样，一旦设置好了就不许再次更改

*/

public class JobInfo {

//唯一性标志

private final String jobName;

//任务处理器，要求业务人员实现接口

private final ITaskProcessor, ?> taskProcessor;

//工作(Job)中任务(Task)的数量

private final int jobLength;

//以下两个类保证操作原子性

//任务总执行成功个数

private AtomicInteger successCount;

//已执行的任务总数

private AtomicInteger taskProcessCount;

//每个任务的处理结果，供查询调用

private LinkedBlockingDeque> taskDetailQueue;

public JobInfo(String jobName , int jobLength , ITaskProcessor,?> taskProcessor) {

this.jobName = jobName;

this.jobLength = jobLength;

this.taskProcessor = taskProcessor;

this.successCount = new AtomicInteger(0);

this.taskProcessCount = new AtomicInteger(0);

this.taskDetailQueue = new LinkedBlockingDeque>(jobLength);

}

//提供工作的整体进度信息

public String getTotalProcess() {

return "success[" + successCount.get()+"]/current[" + taskProcessCount.get() + "],Total=[" + jobLength + "]";

}

//取得工作中每个任务的详情

public List> getTaskDetail() {

List> taskDetailList = new LinkedList>();

TaskResult taskResult;

//pollFirst()方法返回双端队列的第一个元素，返回的元素会从列表中移除

while((taskResult = taskDetailQueue.pollFirst()) != null) {

taskDetailList.add(taskResult);

}

return taskDetailList;

}

//放入工作详情，只需要保证最终个数正确即可，不需要加锁

public void addTaskResult(TaskResult result) {

if(TaskResultType.SUCCESS == result.getResultType()) {

successCount.getAndIncrement();

}

taskProcessCount.getAndIncrement();

taskDetailQueue.add(result);

}

public String getJobName() {

return jobName;

}

public ITaskProcessor, ?> getTaskProcessor() {

return taskProcessor;

}

public int getJobLength() {

return jobLength;

}

public int getSuccessCount() {

return successCount.get();

}

public int getTaskProcessCount() {

return taskProcessCount.get();

}

@Override

public String toString() {

return "JobInfo [jobName=" + jobName + ", taskProcessor=" + taskProcessor + ", jobLength=" + jobLength

+ ", successCount=" + successCount + ", taskProcessCount=" + taskProcessCount + ", taskDetailQueue="

+ taskDetailQueue + "]";

}

}

关于LinkedBlockingDeque的说明：他是线程安全的。他是双端队列，任何一端都可以进行元素的出入。

8.ITaskProcessor

package me.hcFramework.pool.vo;

/**

* 定义接口，所有需要完成的任务都需要实现此接口进行

*

* @param 业务方法需要的数据

* @param 业务方法处理后的业务结果数据的类型

*/

public interface ITaskProcessor {

TaskResult taskExecute(T data);

}

9.真正的黑箱子：PendingJobPool

package me.hcFramework.pool;

import java.util.List;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.ConcurrentHashMap;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.ThreadPoolExecutor;

import java.util.concurrent.TimeUnit;

import me.hcFramework.pool.vo.ITaskProcessor;

import me.hcFramework.pool.vo.JobInfo;

import me.hcFramework.pool.vo.TaskResult;

import me.hcFramework.pool.vo.TaskResultType;

/**

*

* 这是框架主体类

*/

public class PendingJobPool {

//key = 每个工作的名字 jobInfo.jobName

//工作的存放容器，用于完成工作的注册

private static ConcurrentHashMap> jobInfoMap =

new ConcurrentHashMap>();

//单例模式组合拳：类内部实例化+私有构造方法+静态get方法

private static PendingJobPool pool = new PendingJobPool();

private PendingJobPool(){

}

public static PendingJobPool getPool() {

//这里是为了完善逻辑，且为日后框架加入检查功能预留空间

//当然这里也可成为后续版本AOP的切点

//checkJob.initCheck(jobInfoMap);

return pool;

}

//根据工作名称，拿工作的实体

@SuppressWarnings("unchecked")

public JobInfo getJob(String jobName) {

JobInfo jobInfo = (JobInfo) jobInfoMap.get(jobName);

if(null == jobInfo) {

throw new RuntimeException(jobName + "是非法任务！");

}

return jobInfo;

}

//获得处理详情，这里不对与jobName作出检查

public List> getTaskDetail(String jobName) {

JobInfo jobInfo = getJob(jobName);

return jobInfo.getTaskDetail();

}

//获得处理进度

public String getTaskProgess(String jobName) {

return getJob(jobName).getTotalProcess();

}

//获得当前已处理多少个任务

public int getDoneCount(String jobName) {

return getJob(jobName).getTaskProcessCount();

}

/**

* 注册方法：注册工作(job)

* @param jobName 名字

* @param jobLength 工作中任务的长度

* @param taskProcessor 业务处理器

*/

public void registerJob(String jobName , int jobLength , ITaskProcessor,?> taskProcessor) {

JobInfo jobInfo = new JobInfo(jobName, jobLength, taskProcessor);

//putIfAbsent()如果map中没有该工作，则放入且返回null；如果已有会返回对象

if(jobInfoMap.putIfAbsent(jobName, jobInfo) != null) {

throw new RuntimeException(jobName + "已经注册过了");

}

}

/**

* 提交任务

* @param jobName 任务所对应的工作名

* @param t任务数据

*/

public void putTask(String jobName , T t) {

JobInfo jobInfo = getJob(jobName);

PendingTask task = new PendingTask(jobInfo , t);

taskExecutor.execute(task);

}

//取得当前机器上的CPU数量

private static final int THREAD_COUNTS = Runtime.getRuntime().availableProcessors();

//阻塞队列，线程池使用，用以存放待处理的任务

private static BlockingQueue taskQueue = new ArrayBlockingQueue<>(5000);

//线程池，固定大小，有界队列

private static ExecutorService taskExecutor = new ThreadPoolExecutor(THREAD_COUNTS, THREAD_COUNTS, 60, TimeUnit.SECONDS, taskQueue);

public void closePool() {

taskExecutor.shutdown();

}

//交给我们框架执行的任务

private static class PendingTask implements Runnable {

private JobInfo jobInfo;

private T processData;

public PendingTask(JobInfo jobInfo , T processData) {

this.jobInfo = jobInfo;

this.processData = processData;

}

@SuppressWarnings("unchecked")

@Override

public void run() {

ITaskProcessor taskProcessor = (ITaskProcessor) jobInfo.getTaskProcessor();

TaskResult result = null;

try{

result = taskProcessor.taskExecute(processData);

if(result== null) {

result = new TaskResult(TaskResultType.EXCEPTION, null , "is null");

}else if(result.getResultType() == null) {

//如果你看懂这个判断，就会觉得很厉害同时又会感到羞辱

if(result.getReason() == null) {

result = new TaskResult(TaskResultType.EXCEPTION, null , "reason is null");

} else {

result = new TaskResult(TaskResultType.EXCEPTION, null , "type is null");

}

}

} catch (Exception e) {

result = new TaskResult(TaskResultType.EXCEPTION, null ,

"task exception" + e.getMessage());

} finally {

jobInfo.addTaskResult(result);

}

}

}

}

如果读者了解Spring的实现，会知道bean的注册过程其实也就是放入了Map中。或者读者也曾经开发过一些需要注册功能的应用，无疑都是使用了Map。除了Map的高性能，真的可以说是：聪明人都只用一种聪明法。

10.测试

自己实现ITaskProcessor接口

public class MyTask implements ITaskProcessor{

@Override

public TaskResult 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(TaskResultType.SUCCESS, returnValue);

} else if(flag > 300 && flag <= 400) {//处理失败的情况

return new TaskResult(TaskResultType.FAILSURE, -1 , "Failsure");

} else {

try {

throw new RuntimeException("异常发生了!!");

} catch(Exception e) {

return new TaskResult(TaskResultType.EXCEPTION, -1 ,e.getMessage());

}

}

}

}

Test类

public class AppTest {

private final static String JOB_NAME="计算数值";

//private final static String JOB_OTHER_NAME = "字符串";

private final static int JOB_LENGTH = 150;

private static class QueryResult implements Runnable {

private PendingJobPool pool;

private String jobName;

public QueryResult(PendingJobPool pool , String jobName) {

this.pool = pool;

this.jobName = jobName;

}

@Override

public void run() {

while(pool.getDoneCount(jobName) <= JOB_LENGTH) {

List> taskDetail = pool.getTaskDetail(jobName);

if(!taskDetail.isEmpty()) {

System.out.println(pool.getTaskProgess(jobName));

System.out.println(taskDetail);

}

if(pool.getDoneCount(jobName) == JOB_LENGTH) {

break;

}

try {

Thread.sleep(100);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

public static void main(String[] args) {

PendingJobPool pool = PendingJobPool.getPool();

MyTask myTask = new MyTask();

pool.registerJob(JOB_NAME, JOB_LENGTH, myTask);

Random r = new Random();

for(int i = 0 ; i < JOB_LENGTH ; i++) {

pool.putTask(JOB_NAME, r.nextInt(1000));

}

new Thread(new QueryResult(pool, JOB_NAME)).start();

}

}

Test类中实现了一个用来查询的线程。

---------------------

作者：MonkeyDCoding

来源：CSDN

原文：https://blog.csdn.net/MonkeyDCoding/article/details/81369610

版权声明：本文为博主原创文章，转载请附上博文链接！