同步排队 异步操作

Question

我有一个显示在 UITableView 上的项目列表/数组。每个项目都包含一个状态机，它使用期货和承诺机制执行一些异步任务，并且它的状态显示在相应的 UITableViewCell 上。目前对于一个模型来说它工作得很好。

但是，我需要将其作为批处理序列进行。例如，一个数组中可以有 15 个模型，但是在特定时刻我只能启动 3 个模型，一旦模型完成或失败，我应该手动触发第 4 个模型来启动它的任务。注意：我无法启动所有 15 个模型的操作，而只是等待回调，因为它受到硬件的限制，并且在那种情况下会立即失败。

如果上面不清楚，具体来说，下面是两个例子： 我的问题陈述与 iPhone App Store 应用中 Updates 选项卡下的 全部更新 功能完全匹配。如果您有 20 个应用程序更新并点击“全部更新”按钮，它会显示 17 个处于等待状态的应用程序，并且随时仅在 3 个应用程序上运行更新下载。应用程序更新完成后，将移至下一个。这是我的问题陈述的精确复制品，但有一点小改动。

Twist：我的操作是通过蓝牙进行的硬件相关操作。想想你有 20 个可穿戴设备，你想通过蓝牙写入一些数据来进行配置。硬件限制是，您一次最多可以连接 3-4 个设备。因此，一旦设备/外围设备操作成功或失败，我应该尝试连接 4 个，依此类推，直到所有操作都完成。还有重试功能，可以将失败的队列返回。

我的问题是我应该如何构建来维护和监控。我对并发有一个大致的了解，但是并没有做太多的工作。我目前的感觉是使用包裹在 Manager 类中的队列和计数器来监视状态。想要一些关于如何解决这个问题的帮助。我也不需要代码，只需要数据结构的概念解决方案。

Answer 1

我认为你的情况是一个很好的例子。我用 ReactiveKit 整理了一个简单的例子，你可以看看。反应套件非常简单，对于这种情况来说绰绰有余。您也可以使用任何其他反应式库。我希望它有所帮助。

ReactiveKit：https://github.com/DeclarativeHub/ReactiveKit 邦德：https://github.com/DeclarativeHub/Bond

安装 reactiveKit 依赖后，您可以在工作区中运行以下代码：

import UIKit
import Bond
import ReactiveKit

class ViewController: UIViewController {

    var jobHandler : JobHandler!
    var jobs = [Job(name: "One", state: nil), Job(name: "Two", state: nil), Job(name: "Three", state: nil), Job(name: "Four", state: nil), Job(name: "Five", state: nil)]

    override func viewDidLoad() {
        super.viewDidLoad()
        self.jobHandler = JobHandler()
        self.run()
    }

    func run() {
        // Initialize jobs with queue state
        _ = self.jobs.map({$0.state.value = .queue})
        self.jobHandler.jobs.insert(contentsOf: jobs, at: 0)
        self.jobHandler.queueJobs(limit: 2) // Limit of how many jobs you can start with
    }
}

// Job state, I added a few states just as test cases, change as required
public enum State {
    case queue, running, completed, fail
}

class Job : Equatable {

    // Initialize state as a Reactive property
    var state = Property<State?>(nil)
    var name : String!

    init(name: String, state: State?) {
        self.state.value = state
        self.name = name
    }

    // This runs the current job
    typealias jobCompletion = (State) -> Void
    func runJob (completion: @escaping jobCompletion) {
        DispatchQueue.main.asyncAfter(deadline: .now() + 2.0) {
            self.state.value = .completed
            completion(self.state.value ?? .fail)
            return
        }
        self.state.value = .running
        completion(.running)
    }

    // To find the index of current job
    static func == (lhs: Job, rhs: Job) -> Bool {
        return lhs.name == rhs.name
    }
}

class JobHandler {

    // The array of jobs in an observable form, so you can see event on the collection
    var jobs = MutableObservableArray<Job>([])
    // Completed jobs, you can add failed jobs as well so you can queue them again
    var completedJobs = [Job]()


    func queueJobs (limit: Int) {
        // Observe the events in the datasource
        _ = self.jobs.observeNext { (collection) in
            let jobsToRun = collection.collection.filter({$0.state.value == .queue})
            self.startJob(jobs: Array(jobsToRun.prefix(limit)))
        }.dispose()
    }

    func startJob (jobs: [Job?]) {
        // Starts a job thrown by the datasource event
        jobs.forEach { (job) in
            guard let job = job else { return }
            job.runJob { (state) in
                switch state {
                case .completed:
                    if !self.jobs.collection.isEmpty {
                        guard let index = self.jobs.collection.indexes(ofItemsEqualTo: job).first else { return }
                        print("Completed " + job.name)
                        self.jobs.remove(at: index)
                        self.completedJobs.append(job)
                        self.queueJobs(limit: 1)
                    }
                case .queue:
                    print("Queue")
                case .running:
                    print("Running " + job.name)
                case .fail:
                    print("Fail")
                }
            }
        }
    }
}

extension Array where Element: Equatable {
    func indexes(ofItemsEqualTo item: Element) -> [Int]  {
        return enumerated().compactMap { $0.element == item ? $0.offset : nil }
    }
} 

Answer 2

使用OperationQueue，对于每种操作，您都有Operation 的子类。通过操作，您可以添加对其他操作的依赖以完成。因此，例如，如果您想等到第 3 次操作完成后再启动第 4、第 5 和第 6 次操作，您只需将第 3 次操作添加为它们的依赖项。

编辑：因此，要将操作组合在一起，您可以为其创建一个单独的类。我在下面添加了一个代码示例。 add(dependency: OperationGroup) 函数告诉其他组在原始组中执行完操作后立即开始执行操作。

//Make a subclass for each kind of operation.
class BluetoothOperation: Operation
{
    let number: Int

    init(number: Int)
    {
        self.number = number
    }

    override func main() 
    {
        print("Executed bluetooth operation number \(number)")
    }
}

class OperationGroup
{
    var operationCounter: Int = 0
    var operations: [Operation]
    let operationQueue: OperationQueue = OperationQueue()

    init(operations: [Operation])
    {
        self.operations = operations
    }

    func executeAllOperations()
    {
        operationQueue.addOperations(operations, waitUntilFinished: true)
    }

    //This in essence is popping the "Stack" of operations you have.
    func pop() -> Operation?
    {
        guard operationCounter < operations.count else { return nil }

        let operation = operations[operationCounter]

        operationCounter += 1

        return operation
    }

    func add(dependency: OperationGroup)
    {
        dependency.operations.forEach(
        {
            $0.completionBlock =
            {
                if let op = self.pop()
                {
                    dependency.operationQueue.addOperation(op)
                }
            }
        })
    }
}


let firstOperationGroup = OperationGroup(operations: [BluetoothOperation(number: 1), BluetoothOperation(number: 2), BluetoothOperation(number: 3)])
let secondOperationGroup = OperationGroup(operations: [BluetoothOperation(number: 4), BluetoothOperation(number: 5), BluetoothOperation(number: 6)])

secondOperationGroup.add(dependency: firstOperationGroup)

firstOperationGroup.executeAllOperations()