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

前言

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

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

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

image-20210521152444904

正文

最近团队里有基础技术的同学对项目中的线程池进行了重新设计,调整了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/sched

se.sum_exec_runtime CPU上的运行时长

img

img

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

img

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

img

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

最后,让我们一起加油吧,“打工人”!

遇事不要慌,加入萝卜头联盟大家庭QQ群(1061186476),互帮互助,提高自己!!还有更多好的赚钱项目等你~~~~

image-20210519160604903

萝卜头联盟,一个谈钱不伤感情的网站!
萝卜头网创联盟 » 如何区分IO密集型、CPU密集型任务?

发表评论

提供最优质的资源集合

立即查看 了解详情