Синхронная постановка в очередь Асинхронные операции

Question

У меня есть список / массив элементов, которые отображаются в UITableView.Каждый элемент содержит в себе конечный автомат, который выполняет некоторую асинхронную задачу с использованием механизма фьючерсов и обещаний, и его состояние отображается на соответствующем UITableViewCell.В настоящее время для одной модели это работает отлично.

Однако мне нужно сделать это как последовательность операций.Например, в массиве может быть 15 моделей, однако в определенный момент я смогу запустить только 3, после того, как одна из моделей завершится или произойдет сбой, я должен вручную запустить 4-ую модель, чтобы инициировать ее задачу.Примечание. Я не могу инициировать работу всех 15 моделей и просто жду обратного вызова, поскольку оно ограничено аппаратным обеспечением, и сразу приводит к сбою в этом сценарии.

Чтобы быть точным, если вышеприведенное не ясно, ниже приведены два примера: МойВид задачи в точности совпадает с функцией Обновить все в разделе Обновления на вкладке App Store в приложении iPhone.Если у вас есть 20 обновлений приложений и нажмите кнопку «Обновить все», это означает, что 17 приложений находятся в состоянии ожидания и загружают обновления только для 3 приложений в любой момент.После завершения обновления приложения оно переходит к следующему.Это точная копия моей постановки проблемы, но с небольшим поворотом.

Поворот: Мои операции связаны с аппаратным обеспечением через Bluetooth.Подумайте, у вас есть 20 носимых устройств, которые вы хотите настроить с записью некоторых данных через Bluetooth.Аппаратное ограничение: вы можете подключать до 3-4 устройств одновременно.Следовательно, как только устройство / периферийное устройство преуспевает или терпит неудачу с операцией, я должен попытаться соединить 4 один и так n так постепенно, пока все не будет сделано.Также есть функция повтора, которая ставит в очередь неудачную очередь.

Моя проблема в том, как мне структурировать, чтобы поддерживать это и отслеживать.У меня есть общее представление о параллелизме, но я не очень много работал над этим.Мое текущее чувство - использовать Очередь и счетчики, обернутые в классе Менеджера, для мониторинга состояний.Хотелось бы помочь, как подойти к этому.Также мне не нужен код, просто концептуальное решение для структуры данных.

Answer 1

Я думаю, что у вас есть хороший повод для немного реактивного.Я собрал небольшой пример с использованием ReactiveKit, чтобы вы могли взглянуть.Реактивный набор довольно прост и более чем достаточно для этого случая.Вы также можете использовать любую другую реактивную библиотеку.Я надеюсь, что это помогает.

ReactiveKit: https://github.com/DeclarativeHub/ReactiveKit Bond: https://github.com/DeclarativeHub/Bond

Вы можете запустить приведенный ниже код в рабочей области после установки зависимостей реактивного набора:

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