一个用Kotlin编写简易的串行任务调度器

引言

由于项目中有处理大量后台任务并且串行执行的需求，特意写了一个简易的任务调度器，方便监控每个任务执行和异常情况，任务之间互不影响。正如上所述，Kotlin中的TaskScheduler类提供了一个强大的解决方案，用于使用ScheduledExecutorService异步地排队和执行任务。

---

使用方法

1.初始化：

val taskListener = object : TaskScheduler.TaskListener {
    override fun beforeExecute(task: TaskScheduler.NamedRunnable) {
        println("开始任务：${task.name}")
    }

    override fun afterExecute(task: TaskScheduler.NamedRunnable, exception: Exception?) {
        println("完成任务：${task.name}，异常：$exception")
    }
}
val scheduler = TaskScheduler(taskListener, 5)
1
2
3
4
5
6
7
8
9
10

2.提交任务：

scheduler.submit("加载数据") {
    // 加载数据的代码
}
scheduler.submit("处理数据") {
    // 处理数据的代码
}
1
2
3
4
5
6

3.优雅关闭：

当所有任务完成后，调度器将在指定的超时后自动关闭，确保不浪费资源。

---

完整代码

import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicBoolean

class TaskScheduler(private val listener: TaskListener? = null, private val timeout: Long = 5) {
    private val taskQueue = ConcurrentLinkedQueue<NamedRunnable>()
    private val isTaskRunning = AtomicBoolean(false)
    private var executorService: ScheduledExecutorService? = null

    @Synchronized
    fun submit(name: String, task: Runnable) {
        ensureExecutorService()
        taskQueue.offer(NamedRunnable(name, task))
        if (isTaskRunning.compareAndSet(false, true)) {
            executorService?.submit { processTasks() }
        }
    }

    private fun processTasks() {
        try {
            while (taskQueue.isNotEmpty()) {
                val nextTask = taskQueue.poll()
                listener?.beforeExecute(nextTask)
                var exception: Exception? = null
                try {
                    nextTask.run()
                } catch (e: Exception) {
                    exception = e
                }
                listener?.afterExecute(nextTask, exception)
            }
        } finally {
            isTaskRunning.set(false)
            scheduleShutdown()
        }
    }

    private fun ensureExecutorService() {
        if (executorService == null || executorService!!.isShutdown) {
            executorService = Executors.newSingleThreadScheduledExecutor()
            println("ensureExecutorService newSingleThreadScheduledExecutor")
        }
    }

    private fun scheduleShutdown() {
        executorService?.schedule({
            if (taskQueue.isEmpty() && isTaskRunning.compareAndSet(false, true)) {
                executorService?.shutdown()
                executorService = null
                println("scheduleShutdown shutdown")
            } else {
                isTaskRunning.set(false)  // 确保新任务可以触发执行器重启
            }
        }, timeout, TimeUnit.SECONDS)
    }

    interface TaskListener {
        fun beforeExecute(task: NamedRunnable)
        fun afterExecute(task: NamedRunnable, exception: Exception?)
    }

    class NamedRunnable(val name: String, private val task: Runnable) : Runnable {
        override fun run() {
            task.run()
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

最后

简要概括下优缺点：

• 资源自动管理，超时自动释放资源
• 任务命名，更清晰的了解每个任务执行情况
• 线程安全，不用担心多线程添加任务导致顺序紊乱

优点：

• 灵活性：允许动态添加任务，并根据任务负载需要创建或关闭执行器，从而管理执行器的生命周期。

缺点：

• 单线程限制：当前实现使用单线程执行器，这意味着任务是顺序执行的，而不是并行执行。这可能是CPU密集型任务的瓶颈。