Swift Combine，如何取消执行答案

【问题标题】：Swift Combine, how to cancel the executionSwift Combine，如何取消执行
【发布时间】：2021-06-06 19:56:10
【问题描述】：

我是 Combine 的新手，我试图通过解决“旧”问题来理解它

我的目标是使进程可取消，但即使我在printFizzbuzz() 方法之后（或带有排序延迟）调用了.cancel() 方法，代码仍在运行（大约 3 秒）直到完成

我在新的 Xcode 项目中试过下面的代码，还是一样

import Foundation
import Combine

enum PrinterError: Error {
    case indexError(String)
    case subscriptionError(String)
    
    var description: String {
        switch self {
        case .indexError(let descpription):
            return descpription
        case .subscriptionError(let description):
            return description
        }
    }
}

struct FizzbuzzPrinter {
    private var subscriptions = Set<AnyCancellable>()

    mutating func printFizzbuzz(fromIndex: Int, toIndex: Int, handler: @escaping (_ result: Result<Int,PrinterError>) -> Void) {
        guard toIndex > fromIndex else {
            handler(.failure(.indexError("toIndex must larger than fromIndex")))
            return
        }

        var currentIndex: Int = fromIndex
        
        Array<Int>(fromIndex ..< toIndex).publisher
            .handleEvents(receiveOutput: { index in
                currentIndex = index
            }, receiveCancel: {
                handler(.failure(.subscriptionError("cancaled at \(currentIndex)")))
            })
            .map { number -> String in
                switch (number.isMultiple(of: 3), number.isMultiple(of: 5) ) {
                case (true, true):
                    return "fizzbuzz at \(number)"
                case (true, false):
                    return "fizz at \(number)"
                case (false, true):
                    return "buzz at \(number)"
                case (false, false):
                    return String()
                }
            }
            .filter{ !$0.isEmpty }
            .sink { _ in
                handler(.success(currentIndex))
            } receiveValue: { print($0)}
            .store(in: &subscriptions)
        
    }
    
    mutating func cancelAll() {
        subscriptions.forEach{ $0.cancel()}
    }
}

var fizzBuzzPrinter = FizzbuzzPrinter()

DispatchQueue.main.async {
    fizzBuzzPrinter.printFizzbuzz(fromIndex: 1, toIndex: 60001) { result in
        switch result {
        case .failure(let printerError):
            print(printerError.description)
        case .success(let finishedIndex):
            print("finished at \(finishedIndex)")
        }
    }
}
DispatchQueue.main.async {
    fizzBuzzPrinter.cancelAll()
}

DispatchQueue.main.asyncAfter(deadline: .now() + 0.1) {
    fizzBuzzPrinter.cancelAll()
}

代码打印（最后 10 行）：

fizz at 59982
fizzbuzz at 59985
fizz at 59988
buzz at 59990
fizz at 59991
fizz at 59994
buzz at 59995
fizz at 59997
fizzbuzz at 60000
finished at 60000

我也试过用 .switchToLatest() 操作符，还是无法取消

struct FizzbuzzPrinter {
    private var subscriptions = Set<AnyCancellable>()
    
    private let publishers = PassthroughSubject<AnyPublisher<Int, Never>, Never>()
    private let finishPublisher = PassthroughSubject<Int,Never>()


    mutating func printFizzbuzz(fromIndex: Int, toIndex: Int, handler: @escaping (_ result: Result<Int,PrinterError>) -> Void) {
        guard toIndex > fromIndex else {
            handler(.failure(.indexError("toIndex must larger than fromIndex")))
            return
        }

        var currentIndex: Int = fromIndex
        
        publishers
            .switchToLatest()
            .handleEvents(receiveOutput: { index in
                currentIndex = index
            }, receiveCancel: {
                handler(.failure(.subscriptionError("cancaled at \(currentIndex)")))
            })
            .map { number -> String in
                switch (number.isMultiple(of: 3), number.isMultiple(of: 5) ) {
                case (true, true):
                    return "fizzbuzz at \(number)"
                case (true, false):
                    return "fizz at \(number)"
                case (false, true):
                    return "buzz at \(number)"
                case (false, false):
                    return String()
                }
            }
            .filter{ !$0.isEmpty }
            .sink { _ in
                handler(.success(currentIndex))
            } receiveValue: { print($0)}
            .store(in: &subscriptions)
        
        publishers.send(Array<Int>(fromIndex ..< toIndex)
                            .publisher
                            .eraseToAnyPublisher())
        
    }
    
    mutating func cancel() {
        publishers.send(finishPublisher.eraseToAnyPublisher())
        finishPublisher.send(completion: .finished)
    }
}

var fizzBuzzPrinter = FizzbuzzPrinter()

DispatchQueue.main.asyncAfter(deadline: .now() + 0.1) {
    fizzBuzzPrinter.cancel()
}
fizzBuzzPrinter.printFizzbuzz(fromIndex: 1, toIndex: 60001) { result in
    switch result {
    case .failure(let printerError):
        print(printerError.description)
    case .success(let finishedIndex):
        print("finished at \(finishedIndex)")
    }
}

我觉得我在某个地方犯了一个错误，但我想不通。

++++++++++++++++++++更新：+++++++++++++++++++

谢谢@matt 我将发布者更改为基于计时器的发布者，现在可以使用了

        Timer.publish(every: 0.1, on: .main, in: .common)
            .autoconnect()
            .scan(fromIndex) { current, _ in
                current + 1
            }
            .prefix(toIndex - fromIndex)

++++++++++++++++++++更新：2 +++++++++++++++++++++

基于@matt 评论

.subscribe(on: DispatchQueue(label: "serial queue"))

和@matt 链接：link 最后一节“施加背压”

            .flatMap(maxPublishers: .max(1)){ num in
                Just(num).delay(for: .seconds(0.01), scheduler: DispatchQueue.main)
            }

两者都有效

谢谢！！

++++++++++++++++++++更新：3 +++++++++++++++++++++

我修改了我以前的代码以成为更通用的代码并发现使用

.subscribe(on: DispatchQueue.global(qos: .background))

和

.receive(on: DispatchQueue.main)

最适合我，速度不受硬编码时间间隔限制，不会互相阻塞并且可以单独取消

我调用了下面的代码（不包括一些不相关的逻辑）

typealias CancelableIntTaskType = CancelableTask<AnyPublisher<Int, Never>>
var cancelableIntTasks = CancelableIntTaskType()
var taskIdArray = [UUID?]()

taskIdArray = [
    startFizzBuzzTask(50, -80, &cancelableIntTasks),
    startFizzBuzzTask(100, 200, &cancelableIntTasks),
    startFizzBuzzTask(600, 1000, &cancelableIntTasks),
    startFizzBuzzTask(2000, 2600, &cancelableIntTasks)
]

checkAllTasksStarted(taskIdArray)

DispatchQueue.main.asyncAfter(deadline: .now() + 0.0001) {
    if let id = taskIdArray.compactMap({$0}).first {
        cancelableIntTasks.cancelTaskWithID(id)
    }
}

DispatchQueue.main.asyncAfter(deadline: .now() + 0.0002) {
    cancelableIntTasks.cancelAll()
}

输出

*id 52F started
id:52F val: 100 progress 0.0%
id:52F val: 101 progress 1.0%
id:52F val: 102 progress 2.0%
id:52F val: 103 progress 3.0%
id:52F val: 104 progress 4.0%
id:52F val: 105 progress 5.0%
id:52F val: 106 progress 6.0%
id:52F val: 107 progress 8.0%
id:52F val: 108 progress 8.0%
id:52F val: 109 progress 9.0%
id:52F val: 110 progress 10.0%
id:52F val: 111 progress 11.0%
id:52F val: 112 progress 12.0%
*id 9EA started
id:52F val: 113 progress 13.0%
id:52F val: 114 progress 15.0%
id:52F val: 115 progress 15.0%
*id 97B started
id:52F val: 116 progress 16.0%
id:52F val: 117 progress 17.0%
id:52F val: 118 progress 18.0%
id:9EA val: 600 progress 0.0%
id:9EA val: 601 progress 1.0%
id:97B val: 2000 progress 0.0%
id:9EA val: 602 progress 1.0%
id:9EA val: 603 progress 1.0%
id:9EA val: 604 progress 1.0%
id:97B val: 2001 progress 1.0%
id:9EA val: 605 progress 2.0%
id:97B val: 2002 progress 1.0%
id:52F val: 119 progress 19.0%
id:9EA val: 606 progress 2.0%
4 tasks initiated 3 started
task at index 0 failed
id:97B val: 2003 progress 1.0%
id:52F val: 120 progress 20.0%
id:9EA val: 607 progress 2.0%
id:97B val: 2004 progress 1.0%
id:9EA val: 608 progress 2.0%
id:52F val: 121 progress 21.0%
id:97B val: 2005 progress 1.0%
id:9EA val: 609 progress 3.0%
id:97B val: 2006 progress 1.0%
id:9EA val: 610 progress 3.0%
id:97B val: 2007 progress 2.0%
id:9EA val: 611 progress 3.0%
id:97B val: 2008 progress 2.0%
id:9EA val: 612 progress 3.0%
id:97B val: 2009 progress 2.0%
id:9EA val: 613 progress 4.0%
id:97B val: 2010 progress 2.0%
id:9EA val: 614 progress 4.0%
id:97B val: 2011 progress 2.0%
id:52F val: 122 progress 22.0%
id:9EA val: 615 progress 4.0%
id:52F val: 123 progress 23.0%
id:9EA val: 616 progress 4.0%
id:52F val: 124 progress 24.0%
id:9EA val: 617 progress 5.0%
id:52F val: 125 progress 25.0%
id:97B val: 2012 progress 2.0%
id:9EA val: 618 progress 5.0%
**52F canceled at 125 progress: 25.0%
**97B canceled at 2012 progress: 2.0%
**9EA canceled at 619 progress: 5.0%
id:9EA val: 619 progress 5.0%
id:52F val: 126 progress 26.0%

++++++++++++++++++++++更新4+++++++++++++++++++

刚刚做了一个项目来学习取消/恢复任务的方法

https://github.com/hgtlzyc/PokemonDisplay

【问题讨论】：

嗨@matt 我正在尝试使 printFizzbuzz() 中的耗时过程可以随时取消（也可以用新任务替换）并打印出（在什么索引处）该过程被取消//我读过“取消订阅会释放以前通过附加订阅者分配的任何资源。”，所以我首先尝试取消 FizzbuzzPrinter 中的订阅以取消“fizzbuzz”进程，但它不起作用/然后我尝试使用 .switchToLatest( ) 操作员提前终止进程但仍然无法正常工作
问题是，它不仅耗时，而且阻塞。
@matt 我认为你是对的，我试过 ".subscribe(on: DispatchQueue(label: "serial queue")) " 它变成了可取消的

标签： ios swift xcode combine

【解决方案1】：

问题在于您的发布者过于人为地粗糙：它不是异步的。数组发布者只是一次发布它的所有值，所以你取消太晚了；你正在阻塞主线程。改用计时器发布者之类的东西，或者使用带有延迟和背压的平面图。

【讨论】：

我认为您是对的，我将其更改为基于计时器的计时器，它现在可以使用，目前我很难理解使用“具有延迟和背压的平面图”的方式，我会的稍后再阅读、研究和更新我的帖子// 谢谢
见apeth.com/UnderstandingCombine/operators/…