如何区分IO密集型、CPU密集型任务？

前言

日常开发中，我们时常会听到什么IO密集型、CPU密集型任务...

那么这里提一个问题：大家知道什么样的任务或者代码会被认定为IO/CPU密集？又是用什么样的标准来认定IO/CPU密集？

如果你没有明确的答案，那么就随着这篇文章一起来聊一聊吧。

正文

最近团队里有基础技术的同学对项目中的线程池进行了重新设计，调整了IO线程池等线程池的优化。因此借助这个机会也就了解了一波开篇的那些问题。

一、宏观概念区分

这一部分经验丰富的同学都很熟悉。比如：

1.1、IO密集型任务

一般来说：文件读写、DB读写、网络请求等

1.2、CPU密集型任务

一般来说：计算型代码、Bitmap转换、Gson转换等

二、用代码区分

上一part都是咱们凭借经验划分的，这一part咱们就来用正经的指标来划分任务。

先看有哪些数据指标可以用来进行评估（以下方法以系统日志为准，加之开发经验为辅）：

1. wallTime

任务的整体运行时长（包括了running + runnable + sleep等所有时长）。获取方案：
run() { long start = System.currentTimeMillis(); // 业务代码 long wallTime =
System.currentTimeMillis() - start; }
2. cpuTime

cputime是任务真正在cpu上跑的时长，即为running时长

获取方案1：
run() { long start = SystemClock.currentThreadTimeMillis(); // 业务代码 long
cpuTime = SystemClock.currentThreadTimeMillis() - start; }
获取方案2：
/proc/pid/task/tid/schedse.sum_exec_runtime CPU上的运行时长

3. iowait time/count

指线程的iowait耗时。获取方案：
/proc/pid/task/tid/sched se.statistics.iowait_sum IO等待累计时间
se.statistics.iowait_count IO等待累计次数
具体日志位置同上

4. runnable time

线程runnabel被调度的时长。获取方案：
/proc/pid/task/tid/sched se.statistics.wait_sum 就绪队列等待累计时间
具体日志位置同上

5. sleep time

线程阻塞时长（包括Interruptible-sleep和Uninterruptible-sleep和iowait的时长）。获取方案：
/proc/pid/task/tid/sched se.statistics.sum_sleep_runtime 阻塞累计时间
具体日志位置同上

6. utime/stime

utime是线程在用户态运行时长，stime是线程在内核态运行时长。获取方案：
/proc/pid/task/tid/stat 第14个字段是utime，第15个字段是stime

7. rchar/wchar

wchar是write和pwrite函数写入的byte数。获取方案：
/proc/pid/task/tid/io rchar: ...wchar: ...
（没找到合适的日志，暂不讨论此情况）基于读写char数，我们可以将IO细分成读IO密集型和写IO密集型。

8. page_fault

缺页中断次数，分为major/minor fault。获取方案：
/proc/pid/task/tid/stat 第10个字段是minor_fault，第12个字段是major_fault

9. ctx_switches

线程在用户/内核态的切换次数，分为voluntary和involuntary两种切换。获取方案：
/proc/pid/task/tid/sched nr_switches 总共切换次数 nr_voluntary_switches 自愿切换次数
nr_involuntary_switches 非自愿切换次数
日志位置同上

10. percpuload

平均每个cpu的执行时长。获取方案：
/proc/pid/task/tid/sched avg_per_cpu
日志位置同上

有了上述这些指标，我们就可以开始我们的任务确定了。

以下内容，大家可以自行测试加深印象。

2.1、IO密集型任务

比如这段代码：
val br = BufferedReader(FileReader("xxxx"), 1024) try { while (br.readLine()
!= null) { } } finally { if (br != null) { br.close() } }
基于上述部分3. iowait time/count，我们可以在对应的日志文件中看出这段代码有明显的iowait。

2.2、CPU密集型任务

比如这段代码：
var n = 0.0 for (i in 0..9999999) { n = Math.cos(i.toDouble() )}
基于上述部分6. utime/stime的内容，看一看出这段代码utime会占比非常高，且几乎没有stime，此外没有io相关的耗时。

三、这玩意有啥用？

说白了，我们一切的优化手段都是为了服务于业务。对于业务开发来说：

为了不占用主线程 -> 所以启一个新线程 -> 频繁的new线程又会带来大量的开销 -> 所以使用线程池进行复用 ->
而不合理的线程池设计又会带来线程使用低效，甚至新加入的任务只能等待 -> 优化线程池

举个最简单的例子：线程池中放了最大允许俩个线程并行，那么假设运行中的俩个都是长IO的任务。那么新来的任务就只能等，哪怕它并不是特别耗时...

因此这玩意有啥用，还不是为更好的线程池设计做指导思想，更好的提升线程运行效率，降低业务上不必要的等待。

这里提供一些可供参考的工具方法和线程池设计：

3.1、判断任务类型

这里贴一些核心的思路，毕竟全部方案数据公司的代码，我也不方便全部贴出来：
class TaskInfo { var cpuTimeStamp = 0.0 var timeStamp = 0.0 var iowaitTime =
0.0 var sleepTime = 0.0 var runnableTime = 0.0 var totalSwitches = 0.0 var
voluntarySwitches = 0.0} object TaskInfoUtils { private const val
SUM_SLEEP_RUNTIME = "se.statistics.sum_sleep_runtime" private const val
WAIT_SUM = "se.statistics.wait_sum" private const val IOWAIT_SUM =
"se.statistics.iowait_sum" private const val NR_SWITCHES = "nr_switches "
private const val NR_VOLUNTARY_SWITCHES = "nr_voluntary_switches" private var
schedPath = ThreadLocal<String>() fun buildCurTaskInfo(): TaskInfo { val
threadInfo = TaskInfo() threadInfo.timeStamp =
System.currentTimeMillis().toDouble() threadInfo.cpuTimeStamp =
SystemClock.currentThreadTimeMillis().toDouble() if (schedPath.get() == null) {
schedPath.set("/proc/${android.os.Process.myPid()}/task/${getTid()}/sched") }
BufferedReader(FileReader(schedPath.get()), READ_BUFFER_SIZE).use { br ->
br.readLines().forEach { line -> when { line.startsWith(SUM_SLEEP_RUNTIME) ->
threadInfo.sleepTime = line.split(":")[1].toDouble() line.startsWith(WAIT_SUM)
-> threadInfo.runnableTime = line.split(":")[1].toDouble()
line.startsWith(IOWAIT_SUM) -> threadInfo.iowaitTime =
line.split(":")[1].toDouble() line.startsWith(NR_SWITCHES) ->
threadInfo.totalSwitches = line.split(":")[1].toDouble()
line.startsWith(NR_VOLUNTARY_SWITCHES) -> threadInfo.voluntarySwitches =
line.split(":")[1].toDouble() } } } return threadInfo }} object TaskBoundJudge
{ private const val CPU_CPUTIME_INTERVAL = 0.8 private const val
CPU_SWITCHES_INTERVAL = 0.1 private const val CPU_IOWAIT_INTERVAL = 0.01
private const val CPU_SLEEP_INTERVAL = 0.02 private const val
CPU_CPUTIME_WEIGHTS = 0.1 private const val CPU_SWITCHES_WEIGHTS = 0.35 private
const val CPU_IOWAIT_WEIGHTS = 0.15 private const val CPU_SLEEP_WEIGHTS = 0.40
private const val IO_CPUTIME_INTERVAL = 0.5 private const val
IO_SWITCHES_INTERVAL = 0.4 private const val IO_IOWAIT_INTERVAL = 0.1 private
const val IO_SLEEP_INTERVAL = 0.15 private const val IO_CPUTIME_WEIGHTS = 0.1
private const val IO_SWITCHES_WEIGHTS = 0.35 private const val
IO_IOWAIT_WEIGHTS = 0.35 private const val IO_SLEEP_WEIGHTS = 0.2 fun
isCpuTask(start: TaskInfo?, end: TaskInfo?): Boolean { if (start == null || end
== null) { return false } val wallTime = end.timeStamp - start.timeStamp val
cpuTime = end.cpuTimeStamp - start.cpuTimeStamp val runnableTime =
end.runnableTime - start.runnableTime val totalSwitches = end.totalSwitches -
start.totalSwitches val voluntarySwitches = end.voluntarySwitches -
start.voluntarySwitches val iowaitTime = end.iowaitTime - start.iowaitTime val
sleepTime = end.sleepTime - start.sleepTime var result = 0.0 if (cpuTime /
(wallTime - runnableTime) > CPU_CPUTIME_INTERVAL) { result +=
CPU_CPUTIME_WEIGHTS } if (voluntarySwitches / totalSwitches <
CPU_SWITCHES_INTERVAL) { result += CPU_SWITCHES_WEIGHTS } if (iowaitTime /
sleepTime < CPU_IOWAIT_INTERVAL) { result += CPU_IOWAIT_WEIGHTS } if (sleepTime
/ (wallTime - runnableTime) < CPU_SLEEP_INTERVAL) { result += CPU_SLEEP_WEIGHTS
} return result > 0.5 } fun isIOTask(start: TaskInfo?, end: TaskInfo?): Boolean
{ if (start == null || end == null) { return false } val wallTime =
end.timeStamp - start.timeStamp val cpuTime = end.cpuTimeStamp -
start.cpuTimeStamp val runnableTime = end.runnableTime - start.runnableTime val
totalSwitches = end.totalSwitches - start.totalSwitches val voluntarySwitches =
end.voluntarySwitches - start.voluntarySwitches val iowaitTime = end.iowaitTime
- start.iowaitTime val sleepTime = end.sleepTime - start.sleepTime var result =
0.0 if (cpuTime / (wallTime - runnableTime) < IO_CPUTIME_INTERVAL) { result +=
IO_CPUTIME_WEIGHTS } if (voluntarySwitches / totalSwitches >
IO_SWITCHES_INTERVAL) { result += IO_SWITCHES_WEIGHTS } if (iowaitTime /
sleepTime > IO_IOWAIT_INTERVAL) { result += IO_IOWAIT_WEIGHTS } if (sleepTime /
(wallTime - runnableTime) > IO_SLEEP_INTERVAL) { result += IO_SLEEP_WEIGHTS }
return result > 0.5 } }

当我们想对某个方法进行计算是CPU还是IO。可以在这个方法的开始、结束调用 TaskInfoUtils.buildCurTaskInfo()；然后调用 TaskBoundJudge.isCpuTask(start,end)， TaskBoundJudge.isIOTask(start,end)即可。

3.2、线程池

IO密集型参考线程池：
public static final ExecutorService IO_EXECUTOR = new ThreadPoolExecutor( 2,
128, 15, TimeUnit.SECONDS, new SynchronousQueue<>(), new
CustomThreadFactory("MDove-IO", CustomThreadPriority.NORMAL), AbortPolicy() //
根据业务情况，自行定义拒绝实现。比如上报监控平台 );
CPU密集型参考线程池：
public static final int CPU_COUNT =
Runtime.getRuntime().availableProcessors(); public static final int
MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1; private static final int
CPU_CORE_POOL_SIZE = Math.max(Math.min(MAXIMUM_POOL_SIZE, 4),
Math.min(CPU_COUNT + 1, 9)); public static final ExecutorService CPU_EXECUTOR =
new ThreadPoolExecutor( CPU_CORE_POOL_SIZE, CPU_COUNT * 2 + 1, 30,
TimeUnit.SECONDS, new LinkedBlockingQueue<>(256), new
SSThreadFactory("MDove-CPU", CustomThreadPriority.NORMAL), AbortPolicy() //
根据业务情况，自行定义拒绝实现。比如上报监控平台 );
上述线程池中设计的额外代码：
class CustomThreadFactory : ThreadFactory { var name: String private set
private var priority = CustomThreadPriority.NORMAL constructor(name: String,
priority: CustomThreadPriority) { this.name = name this.priority = priority }
override fun newThread(r: Runnable): Thread { val name = name + "-" +
sCount.incrementAndGet() return object : Thread(r, name) { override fun run() {
if (priority == CustomThreadPriority.LOW) {
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND) } else if
(priority == CustomThreadPriority.HIGH) {
Process.setThreadPriority(Process.THREAD_PRIORITY_DISPLAY) } super.run() } } }
companion object { private val sCount = AtomicInteger(0) } } enum class
CustomThreadPriority { LOW, NORMAL, HIGH, IMMEDIATE }
尾声

OK，这篇文章到这里就结束了。希望这篇文章能给大家在线程的使用和线程池的设计上带来帮助。

最后，让我们一起加油吧，“打工人”！

• « 上一篇：阿里巴巴Java开发手册（详尽PDF版）.
• » 下一篇：Java 期末总结