多任务处理器更新

2020年06月16日更新

源码以及详细说明请移步
更新后的版本号&＃xff1a;

implementation &＃39;com.ripple.component:task:0.0.6&＃39;


一、背景

咱们来分析一下多任务&＃xff0c;在使用者的角度可以简单理解为其是一个黑盒&＃xff0c;使用者放入之后经过黑盒处理之后再取出这样就达到了最终的结果。

二、分析抽象

既然是这样我们可以抽象一下&＃xff0c;因为在linux中万物皆文件&＃xff0c;所以咱们传入的其实是一个废弃&＃xff1a;sourcePath&＃xff0c;新增&＃xff1a;source:S&＃xff0c;那么下一步我们就考虑我们想要的是什么了&＃xff0c;然后咱们可以把那个黑盒理解为规则&＃xff0c;那么可以抽象为&＃xff0c;废弃&＃xff1a;fun parse(sourcePath:String,targetPath:String?):String&＃xff0c;
新增&＃xff1a;fun parse(source: S, target: T?): T这里估计大家会疑问为什么会有targetPath&＃xff0c;不是已经有处理结果了么&＃xff0c;这其实是使其更具扩展性&＃xff0c;比如要处理一个文件&＃xff0c;使用者在处理之前就已经为其定好了targetPath&＃xff0c;那么在使用时直接传入即可&＃xff0c;但是还有一种比如将图片转为base64那么知道的只是规则结果是未知的&＃xff0c;这时候就需要去取这个返回值了。
经过以上的分析这个库的主干就出来了&＃xff0c;那么下一步就是要为其装饰了。
使用者在使用时肯定想的是这个库能够处理批量任务并且能够有相应的回调通知&＃xff0c;这样使用者只需要自己定义好处理规则封装为对象&＃xff0c;传入这个多任务处理器引擎中&＃xff0c;得到相应的回调。
更新后定义的接口&＃xff1a;
但是更新后的调用基本没有变化&＃xff0c;回调结果由泛型推导出实际类型

/*** Author: fanyafeng* Data: 2020/6/3 19:15* Email: fanyafeng&＃64;live.cn* Description:*/
interface ProcessModel : Serializable {companion object {const val PROCESS_ITEM &＃61; "process_item"const val PROCESS_LIST &＃61; "process_list"}/*** 获取需要处理的源路径*/fun getSource(): S/*** 目标路径*/fun getTarget(): T?/*** 处理后的目标路径不能为空*/fun setTarget(target: T)/*** 任务解析器* 这里按道理说如果有了targetPath那么这个返回值是可以不需要的* 但是就是因为如果你去处理一个任务但是有规则没有输出那么这个返回值就是必须的了* 而且不能为空** 分为以下两种以下两种情况&＃xff1a;* 1.处理文件类* &＃64;param source 源* &＃64;param target 结果** 2.有处理规则&＃xff0c;和原路径&＃xff0c;那么方法的返回值就是处理结果* &＃64;param source 源**/fun parse(source: S, target: T?): T/*** 简化接口调用*/abstract class ProcessSimpleModel(var sourcePath: S, var targetPath: T) : ProcessModel {override fun getSource(): S {return sourcePath}override fun getTarget(): T? {return targetPath}override fun setTarget(target: T) {this.targetPath &＃61; target}}
}


三、多任务处理器结构图

四、伪代码真调用

这里以kotlin为例&＃xff0c;当然也支持java&＃xff0c;只是kotlin使用起来会更简洁&＃xff0c;来模拟一下咱们想要调用

handleTaskList(taskList){onSuccess{ successResult->}onFailed{ failedResult->}
}


以上只是简单写了几个回调&＃xff0c;基本调用方式是这样&＃xff0c;可能使用者还想要自定义一下线程池&＃xff0c;并行&＃xff0c;串行等

五、编码

因为要写的是一个框架类的库&＃xff0c;需要对扩展开放对修改关闭&＃xff0c;此时再根据上面的抽象以及图便可以很清晰明了的将其完成&＃xff0c;下面来细看

5.1 任务Model封装

因为是批量任务处理&＃xff0c;并且还需要兼容不同类型&＃xff0c;那么这个Model必须要实现ProcessModel接口&＃xff0c;设计之初为了灵活设计了三个接口&＃xff0c;但是又发现用起来比较麻烦&＃xff0c;后来又将其粒度变大&＃xff0c;有舍有得吧。下面来看一下接口的实现&＃xff1a;
PS&＃xff1a;注释中有详细的说明不再做细说明

/*** Author: fanyafeng* Data: 2020/6/3 19:15* Email: fanyafeng&＃64;live.cn* Description: 单项任务接口*/
interface ProcessModel : Serializable {companion object {const val PROCESS_ITEM &＃61; "process_item"const val PROCESS_LIST &＃61; "process_list"}/*** 获取需要处理的源路径* 不能为空皮之不存毛将焉附*/fun getSourcePath(): String/*** 目标路径可以为空* 因为有时只知道规则不知道结果*/fun getTargetPath(): String?/*** 处理后的目标路径不能为空*/fun setTargetPath(target: String)/*** 任务解析器* 这里按道理说如果有了targetPath那么这个返回值是可以不需要的* 但是就是因为如果你去处理一个任务但是有规则没有输出那么这个返回值就是必须的了* 而且不能为空** 分为以下两种以下两种情况&＃xff1a;* 1.处理文件类* &＃64;param sourcePath 文件原路径* &＃64;param targetPath 文件目标路径** 2.有处理规则&＃xff0c;和原路径&＃xff0c;那么方法的返回值就是处理结果* &＃64;param sourcePath 文件原路径**/fun parse(sourcePath: String, targetPath: String?): String
}


5.2 引擎处理设置

这里使用的是java的线程池的主要接口ExecutorService&＃xff0c;主要是其中为我们封装好了我们需要的一些通用的方法&＃xff0c;我这里还是把任务交给线程去处理&＃xff0c;如果是单线程则是串行&＃xff0c;多的话就是并行处理了&＃xff0c;还可以实现接口进行自定义
更新后不仅不会每次都新建线程&＃xff0c;而是去复用之前的线程并且为线程添加了name方便定位问题&＃xff0c;防止重复创建太多的匿名线程&＃xff0c;

/*** Author: fanyafeng* Data: 2020/5/6 17:51* Email: fanyafeng&＃64;live.cn* Description: 使用java自带的任务服务*/
interface ProcessEngine : Serializable {companion object {internal var singleExecutorInner &＃61; Executors.newSingleThreadExecutor(object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内部单线程池-" &＃43; atomic.getAndIncrement())}})internal val SINGLE_THREAD_EXECUTOR_INNER: ProcessEngine &＃61;object : ProcessEngine {override fun getExecutorService(): ExecutorService {return if (!singleExecutorInner.isShutdown) {singleExecutorInner} else {singleExecutorInner &＃61; Executors.newSingleThreadExecutor(object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内部单线程池-" &＃43; atomic.getAndIncrement())}})singleExecutorInner}}override fun shutdown() {singleExecutorInner.shutdown()}}/*** 单线程处理器* 处理任务为串行处理*/private var singleExecutor &＃61; Executors.newSingleThreadExecutor(object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置单线程池-" &＃43; atomic.getAndIncrement())}})val SINGLE_THREAD_EXECUTOR: ProcessEngine &＃61;object : ProcessEngine {override fun getExecutorService(): ExecutorService {return if (!singleExecutor.isShutdown) {singleExecutor} else {singleExecutor &＃61; Executors.newSingleThreadExecutor(object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置单线程池-" &＃43; atomic.getAndIncrement())}})singleExecutor}}override fun shutdown() {singleExecutor.shutdown()}}/*** 自带6个线程的处理器* 不用纠结个数为什么这么定义&＃xff0c;纯属个人喜欢的数字* 处理任务为并行处理&＃xff0c;并且顺序是打乱的*/private var maxExecutor &＃61; Executors.newFixedThreadPool(Thread.MAX_PRIORITY, object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置最大线程池-" &＃43; atomic.getAndIncrement())}})val MULTI_THREAD_EXECUTOR_MAX: ProcessEngine &＃61;object : ProcessEngine {override fun getExecutorService(): ExecutorService {return if (!maxExecutor.isShutdown) {maxExecutor} else {maxExecutor &＃61; Executors.newFixedThreadPool(Thread.MAX_PRIORITY, object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置最大线程池-" &＃43; atomic.getAndIncrement())}})maxExecutor}}override fun shutdown() {maxExecutor.shutdown()}}private var normalExecutor &＃61; Executors.newFixedThreadPool(Thread.NORM_PRIORITY, object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置一般线程池-" &＃43; atomic.getAndIncrement())}})val MULTI_THREAD_EXECUTOR_NORMAL: ProcessEngine &＃61;object : ProcessEngine {override fun getExecutorService(): ExecutorService {return if (!normalExecutor.isShutdown) {normalExecutor} else {normalExecutor &＃61; Executors.newFixedThreadPool(Thread.NORM_PRIORITY, object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置一般线程池-" &＃43; atomic.getAndIncrement())}})normalExecutor}}override fun shutdown() {normalExecutor.shutdown()}}private var minExecutor &＃61; Executors.newFixedThreadPool(Thread.MIN_PRIORITY, object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置最小线程池-" &＃43; atomic.getAndIncrement())}})val MULTI_THREAD_EXECUTOR_MIN: ProcessEngine &＃61;object : ProcessEngine {override fun getExecutorService(): ExecutorService {return if (!minExecutor.isShutdown) {minExecutor} else {minExecutor &＃61; Executors.newFixedThreadPool(Thread.MIN_PRIORITY, object :ThreadFactory {val atomic &＃61; AtomicInteger(1)override fun newThread(r: Runnable): Thread {return Thread(r, "ripple-task-内置最小线程池-" &＃43; atomic.getAndIncrement())}})minExecutor}}override fun shutdown() {minExecutor.shutdown()}}}/*** 获取任务处理任务** 1、线程池&＃xff1a; 提供一个线程队列&＃xff0c;队列中保存着所有等待状态的线程。避免了创建与销毁的额外开销&＃xff0c;提高了响应的速度。** [java.util.concurrent.Executor]* [java.util.concurrent.ExecutorService]* [java.util.concurrent.ThreadPoolExecutor]* [java.util.concurrent.ScheduledExecutorService]* [java.util.concurrent.ScheduledThreadPoolExecutor]** 2、线程池的体系结构&＃xff1a;* java.util.concurrent.Executor 负责线程的使用和调度的根接口* |--ExecutorService 子接口&＃xff1a; 线程池的主要接口* |--ThreadPoolExecutor 线程池的实现类* |--ScheduledExecutorService 子接口&＃xff1a; 负责线程的调度* |--ScheduledThreadPoolExecutor : 继承ThreadPoolExecutor&＃xff0c;实现了ScheduledExecutorService** [java.util.concurrent.Executor]** 3、工具类 &＃xff1a; Executors* ExecutorService newFixedThreadPool() : 创建固定大小的线程池* ExecutorService newCachedThreadPool() : 缓存线程池&＃xff0c;线程池的数量不固定&＃xff0c;可以根据需求自动的更改数量。* ExecutorService newSingleThreadExecutor() : 创建单个线程池。 线程池中只有一个线程** ScheduledExecutorService newScheduledThreadPool() : 创建固定大小的线程&＃xff0c;可以延迟或定时的执行任务*/fun getExecutorService(): ExecutorService/*** 停止任务*/fun shutdown()}


5.3 核心任务处理

这里就包含了任务处理器以及任务回调了&＃xff0c;因为使用者想要的就是把任务处理完成以及结果的回调不论成功或者失败。
更新后的task添加了泛型支持&＃xff0c;需要外部传入类型&＃xff0c;增加了通用性

/*** Author: fanyafeng* Data: 2020/6/3 19:39* Email: fanyafeng&＃64;live.cn* Description:*/
interface ProcessTask {/*** 所有单个任务回调&＃xff0c;任务回调包含以下所有回调&＃xff0c;但是为了简化使用* 除去单项任务完成回调必须重写其余进行了空实现** 单项任务开始回调* [com.ripple.task.callback.OnItemStart]** 单项任务进行回调* [com.ripple.task.callback.OnItemDoing]** 单项任务打断回调* [com.ripple.task.callback.OnItemInterrupted]** 单项任务失败回调* [com.ripple.task.callback.OnItemFailed]** 单项任务成功回调* [com.ripple.task.callback.OnItemSuccess]** 单项任务完成回调* [com.ripple.task.callback.OnItemFinish]*/fun getItemResult(): OnItemResult>?/*** 所有任务回调&＃xff0c;基本同上除去具体回调** 所有任务开始回调* [com.ripple.task.callback.OnStart]** 所有任务进行回调* [com.ripple.task.callback.OnDoing]** 所有任务失败回调* [com.ripple.task.callback.OnFailed]** 所有任务成功回调* [com.ripple.task.callback.OnSuccess]** 所有任务完成结束回调* [com.ripple.task.callback.OnFinish]*/fun getAllResult(): OnAllResult>>?/*** 获取任务处理器引擎* [com.ripple.task.engine.ProcessEngine]*/fun getProcessEngine(): ProcessEngine
}


5.4 以上任务处理的必须部分都完成&＃xff0c;剩下的就是将任务model交付ProcessTask处理即可

因为库的主要作用是为了方便大家使用&＃xff0c;这里进行默认实现&＃xff0c;因为代码量略大不在这里贴了&＃xff0c;如果感觉默认的使用不能满足大家的使用&＃xff0c;大家可以自己在实现相关接口之后进行自己的impl&＃xff0c;这里贴一下库的使用代码&＃xff0c;所有回调的结果都有&＃xff0c;需要哪个回调实现哪个回调方法即可。

5.4.1 以下为java方式调用

主要是kotlin使用比较简洁&＃xff0c;但是兼容了java的调用&＃xff1a;
更新后以string为例&＃xff1a;

val task &＃61; ProcessTaskImpl()
task.onAllResult &＃61; object : OnAllResult>> {override fun onFinish(finishResult: List>?,unFinishResult: List>?) {TODO("所有任务完成回调")}}
task.handleTaskList(listOf(Task1("abdaafda"),Task2("不会输出"),Task3("abdaafda又变大写了")
))


5.4.2 以下为kotlin方式调用

kotlin调用的话就稍微简单一点

handleTaskList(listOf(Task1("abdaafda"),Task2("不会输出"),Task3("abdaafda又变大写了"))
) {onSuccess { successResult ->}onFailed { failedResult ->}onFinish { finishResult, unFinishResult ->println("结果回调" &＃43; finishResult.toString())println(unFinishResult.toString())}
}