如何在Combine的AnyPublisher中使用switchToLatest()取消旧网络请求

问题根源 #

你现在的用法完全搞错了switchToLatest()的适用场景：这个方法是用来处理发布者的发布者（比如Publisher<Publisher<T, E>, E>），而你当前代码里delay之后的发布者是普通的AnyPublisher<[PickupDeliveryPlaces], SSServiceError>，根本不是嵌套的发布者，所以调用switchToLatest()会报错。而且你现在每次调用fetchPlacesPredictions都会创建新的订阅，根本做不到取消旧请求。

正确实现方式 #

要实现“新输入触发新请求时自动取消旧请求”，你需要把输入的字符串做成一个可观察的Subject，然后通过它来触发网络请求，再用switchToLatest()切换到最新的请求发布者。具体步骤如下：

• 添加输入Subject：定义一个用来接收搜索输入的CurrentValueSubject，跟踪用户的输入变化
• 替换delay为debounce：更适合搜索场景，在用户停止输入一段时间后再触发请求，避免频繁请求
• 用switchToLatest()切换请求：通过flatMap把每个输入转换成对应的网络请求发布者，确保只保留最新的请求，旧请求自动取消

修改后的完整代码 #

// 新增：用来接收用户输入的Subject
private let inputSubject = CurrentValueSubject<String, Never>("")
var pickupDeliveryPlaces: CurrentValueSubject<[PickupDeliveryPlaces], Never> = .init([])
private var disposeBag = Set<AnyCancellable>()

// 初始化时设置订阅，只需要订阅一次
private func setupPlacePredictionSubscription() {
    inputSubject
        // 等待用户停止输入0.8秒后再触发请求
        .debounce(for: .seconds(0.8), scheduler: RunLoop.main)
        // 过滤空输入，避免无效请求
        .filter { !$0.isEmpty }
        // 把每个输入转换成对应的网络请求发布者
        .flatMap { [weak self] input -> AnyPublisher<[PickupDeliveryPlaces], SSServiceError> in
            guard let self = self else {
                return Empty(completeImmediately: true).eraseToAnyPublisher()
            }
            return self.deliveryService.fetchPickupDeliveryPlacesCombine(input: input)
                .compactMap { $0 }
                .map { $0.places ?? [] }
                .eraseToAnyPublisher()
        }
        // 切换到最新的请求，旧请求自动取消
        .switchToLatest()
        .sink(receiveCompletion: { completion in
            switch completion {
            case .failure(let error):
                print("Something went wrong: \(error)")
                Debug.log("Error while fetching addresses: \(error)")
                self.error.send(SSServiceError(fuzzMessage: "Something went wrong, give it another fry… um, try!"))
            case .finished:
                print("Received Completion")
            }
        }, receiveValue: { [weak self] places in
            print(places, "---places")
            self?.pickupDeliveryPlaces.send(places)
        })
        .store(in: &disposeBag)
}

// 现在只需要给inputSubject发送新输入即可
private func fetchPlacesPredictions(input: String) {
    inputSubject.send(input)
}

// 你的网络请求方法保持不变
func fetchPickupDeliveryPlacesCombine(input: String, shouldRefresh: Bool = false, debug: Bool = false) -> AnyPublisher<PickupDeliveryPlacesResponse?, SSServiceError> {
    let parameters: [String: Any] = [
        "input": input,
        "filter": String(true)
    ]

    let endPoint = SSNextBaseService.Endpoints.getPlaces
    let request = nextBaseService.getRequest(endPoint, parameters: parameters)
    return service.fetchObservableCombine(request: request, serviceType: .fuzz, shouldRefresh: shouldRefresh, debug: debug)
        .map { ($0 as ResultWrapper<PickupDeliveryPlacesResponse?>).result}
        .eraseToAnyPublisher()
}

关键说明 #

• inputSubject：作为输入的唯一数据源，所有输入变化都通过它发送，确保只有一条订阅链处理请求
• debounce：比delay更适合搜索场景，忽略短时间内的连续输入，只在用户停止输入0.8秒后触发请求，减少无效请求
• switchToLatest()：这里flatMap返回的是嵌套的发布者，调用该方法会自动取消之前未完成的请求，只处理最新的那个

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