同步排队异步操作

Question

我有一个在UITableView上显示的项目的列表/数组。每个项目都包含一个状态机，它使用期货和承诺机制执行一些异步任务，状态在各自的UITableViewCell上显示。目前，对于一种模式，它的工作非常好。

但是，我需要作为一个批处理序列来完成这个任务。例如，数组中可以有15个模型，但是在特定的时刻，我只能启动3个模型，一旦模型完成或失败，我应该手动触发第四个模型来启动它的任务。注意:我不能启动所有15种型号的操作，只需等待回调，因为它受到硬件的限制，并且在该场景中立即失败。

具体而言，如果上面的内容不清楚，下面有两个示例:我的问题陈述与above应用程序中的选项卡中的更新功能完全匹配。如果你有20个应用程序更新，点击‘更新所有’按钮，它会显示17个应用程序处于等待状态，并且在任何时候只运行3个应用程序的更新下载。一旦应用程序更新完成，它就会移动到下一个应用程序。这是我的问题陈述的精确复制品，然而，有一点小的扭曲。

扭转:我的操作是与硬件相关的蓝牙操作。想想看，你有20个可穿戴设备，你想通过蓝牙编写一些数据来配置。硬件限制是，您可以一次连接到最多3-4个设备.因此，一旦设备/外围设备在操作中成功或失败，我应该尝试将4个设备连接起来，然后逐步连接，直到全部完成为止。还有retry函数，哪个队列返回失败的队列。

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

Answer 1

使用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()

Answer 2

我认为你的理由是有一点反应的。我使用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 }
    }
}