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

亦凉 2023-09-28 09:54 80阅读 0赞

## 前言 ##

日常开发中，我们时常会听到什么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上的运行时长

![edf5e630556c41138b0f50326b2ac7f1.png][]

## 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

![c0748737f432e12f134910210ab13e21.png][]

## 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

![fcaf81f4b24c089411ded1d5640c8c49.png][]

## 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，这篇文章到这里就结束了。希望这篇文章能给大家在**线程的使用**和**线程池的设计**上带来帮助。

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

[edf5e630556c41138b0f50326b2ac7f1.png]: https://img-blog.csdnimg.cn/img_convert/edf5e630556c41138b0f50326b2ac7f1.png
[c0748737f432e12f134910210ab13e21.png]: https://img-blog.csdnimg.cn/img_convert/c0748737f432e12f134910210ab13e21.png
[fcaf81f4b24c089411ded1d5640c8c49.png]: https://img-blog.csdnimg.cn/img_convert/fcaf81f4b24c089411ded1d5640c8c49.png