一个简易的并行任务调度器

news2024/9/19 9:37:39

介绍

当你需要在程序中处理大量同类型的耗时任务，并且需要监听每个任务的工作状态和任务的编号，能及时响应并处理错误，而且需要所有任务都执行完毕自动通知时，那么这个任务调度器就非常适合你！

它封装了 Java 的 ThreadPoolExecutor 和 ScheduledExecutorService，提供了一种简便的方式来管理并行任务的执行，并在所有任务完成后自动关闭执行器。

之前写了一个串行任务调度器，这篇也是根据这个改编而来：一个用Kotlin编写简易的串行任务调度器

使用方法

1.初始化：

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

    override fun afterExecute(task: ParallelTaskExecutor.NamedRunnable, exception: Exception?) {
        println("完成任务：${task.name}，异常：$exception")
    }
}
val taskExecutor = ParallelTaskExecutor(taskListener, 1)

2.提交任务：

taskExecutor.submit("加载数据") {
    // 加载数据的代码
}
taskExecutor.submit("处理数据") {
    // 处理数据的代码
}
//或者
for (i in 0..15) {
            taskExecutor.submit("task:$i") {
                Thread.sleep(3000)
            }
        }
...

3.优雅关闭：

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

完整代码

/**
 * ParallelTaskExecutor 是一个用于管理和并行执行任务的类。
 * 它确保所有任务都被执行，并提供一个机制在所有任务完成时通知。
 */
class ParallelTaskExecutor(private val listener: TaskListener? = null, private val timeout: Long = 5) {

    // 用于存储任务的队列
    private val taskQueue = ConcurrentLinkedQueue<NamedRunnable>()

    // 标记当前是否有任务正在处理
    private val isTaskRunning = AtomicBoolean(false)

    // 用于并行运行任务的执行器服务
    private var executorService: ExecutorService? = null

    // 用于安排执行器服务关闭的调度执行器服务
    private val scheduler: ScheduledExecutorService = Executors.newScheduledThreadPool(1)

    // 用于安排执行器服务关闭的 Future
    private var scheduledShutdownFuture: ScheduledFuture<*>? = null

    // 计数器，用于跟踪运行中的任务数量
    private val runningTasksCounter = AtomicInteger(0)

    /**
     * 提交一个任务到执行器。
     * @param name 任务的名称。
     * @param task 要执行的任务。
     */
    @Synchronized
    fun submit(name: String, task: Runnable) {
        ensureExecutorService() // 确保执行器服务已启动
        taskQueue.offer(NamedRunnable(name, task)) // 将任务添加到队列中

        // 取消之前安排的任何关闭操作
        scheduledShutdownFuture?.cancel(false)

        // 如果当前没有运行的任务，开始处理任务
        if (isTaskRunning.compareAndSet(false, true)) {
            processTasks()
        }
    }

    /**
     * 处理队列中的任务。
     */
    private fun processTasks() {
        try {
            while (taskQueue.isNotEmpty()) {
                // 从队列中获取下一个任务
                val nextTask = taskQueue.poll()

                // 递增运行任务计数器
                runningTasksCounter.incrementAndGet()

                // 将任务提交到执行器服务
                executorService?.execute {
                    listener?.beforeExecute(nextTask)
                    var exception: Exception? = null
                    try {
                        nextTask.run()
                    } catch (e: Exception) {
                        e.printStackTrace()
                        exception = e
                    }
                    listener?.afterExecute(nextTask, exception)

                    // 任务执行完毕后递减计数器，并检查是否所有任务都已完成
                    if (runningTasksCounter.decrementAndGet() == 0) {
                        scheduleShutdown()
                    }
                }
            }
        } finally {
            // 重置 isTaskRunning 标志
            isTaskRunning.set(false)
        }
    }

    /**
     * 确保执行器服务已初始化并在运行中。
     */
    private fun ensureExecutorService() {
        if (executorService == null || executorService!!.isShutdown) {
            executorService = Executors.newFixedThreadPool(20)
        }
    }

    /**
     * 如果所有任务都已完成，安排关闭执行器服务。
     */
    private fun scheduleShutdown() {
        // 如果任务队列为空且没有运行中的任务，则安排关闭
        if (taskQueue.isEmpty() && runningTasksCounter.get() == 0) {
            scheduledShutdownFuture = scheduler.schedule({
                executorService?.shutdown()
                executorService = null
                listener?.onShutdown()
            }, timeout, TimeUnit.SECONDS)
        }
    }

    /**
     * 任务监听器接口，提供任务执行事件的回调。
     */
    interface TaskListener {
        fun beforeExecute(task: NamedRunnable)
        fun afterExecute(task: NamedRunnable, exception: Exception?)
        fun onShutdown()
    }

    /**
     * NamedRunnable 是一个对 Runnable 任务的包装，包含一个用于标识的名称。
     */
    class NamedRunnable(val name: String, private val task: Runnable) : Runnable {
        override fun run() {
            task.run()
        }
    }
}