基于Swift Concurrency的任务调度器实现优化咨询

你的实现已经满足了基本的并发控制需求，但确实可以从简化结构、内聚逻辑、增强灵活性这几个方向优化，下面是具体的思路和代码示例：

一、核心优化点 #

1. 移除冗余的CustomJob类 #

原CustomJob只是简单包装了一个闭包，完全可以直接用() async -> Void类型存储任务——既简化了结构，又能天然支持异步任务（原实现仅能处理同步闭包，优化后可兼容同步/异步两种任务类型）。

2. 让Actor全权负责调度逻辑 #

原实现把任务调度的逻辑拆分到JobQueue和JobScheduler两个类中，还需要外部递归调用run()触发任务启动。可以把任务的添加、启动、续跑逻辑全部收拢到Actor内部，减少跨Actor的调用次数，让逻辑更内聚。

3. 合并状态更新与任务续跑 #

原finishOneJob()仅做计数递减，需要外部再触发run()才能启动下一个任务。可以改成在任务完成后，由Actor自动尝试启动下一批任务，避免外部的递归触发逻辑。

二、优化后的代码示例 #

actor JobScheduler {
    private let maxConcurrentCount: Int
    private var runningCount = 0
    private var pendingJobs: [() async -> Void] = []
    
    init(maxConcurrentCount: Int) {
        self.maxConcurrentCount = maxConcurrentCount
    }
    
    func scheduleJob(_ job: @escaping () async -> Void) {
        pendingJobs.append(job)
        tryStartNextJobs()
    }
    
    private func tryStartNextJobs() {
        // 尽可能启动任务，直到达到并发上限或无待处理任务
        while runningCount < maxConcurrentCount, let job = pendingJobs.removeFirst() {
            runningCount += 1
            // 启动任务，完成后自动更新状态并尝试启动下一批
            Task {
                await job()
                runningCount -= 1
                tryStartNextJobs()
            }
        }
    }
}

同步任务适配 #

如果需要兼容原有的同步闭包，只需在调度时简单包装成异步即可：

extension JobScheduler {
    func scheduleSyncJob(_ job: @escaping () -> Void) {
        scheduleJob {
            job()
        }
    }
}

三、测试代码优化（用Swift Concurrency替代DispatchGroup） #

// MARK: - 测试
Task {
    let scheduler = JobScheduler(maxConcurrentCount: 2)
    
    await withTaskGroup(of: Void.self) { group in
        for jobId in 1...5 {
            group.addTask {
                print("Job \(jobId) scheduled")
                await scheduler.scheduleSyncJob {
                    print("Job \(jobId) starts")
                    sleep(2)
                    print("Job \(jobId) complete")
                }
            }
        }
    }
    
    print("Done")
}

四、其他可选实现思路 #

• 基于AsyncChannel的任务分发：如果需要更灵活的任务流转，可以使用Swift Async Algorithms库中的AsyncChannel，Actor负责往通道中发送任务，同时启动固定数量的消费者Task从通道取任务执行，这种方案适合需要动态调整并发数的场景。
• 利用TaskGroup手动控制并发：如果任务是批量提交的，可以直接用withThrowingTaskGroup配合计数器手动限制并发数，但这种方式对动态添加任务的支持不如Actor方案灵活。

内容的提问来源于stack exchange，提问作者Alex