Combine框架中自定义通用Subscriber替代内置sink方法时的报错问题求助

嘿，我完全理解你想自己实现一个替代内置sink的方法来搞懂Combine底层逻辑的想法，遇到这个报错确实挺挠头的，我来帮你分析下问题所在并给出修复方案。

首先，咱们先拆解下你遇到的错误：Instance method 'subscribe' requires that 'CustomSubscriber<Self.Output, Self.Failure>' conform to 'Subject'。这个报错的核心原因有两个：

• 你调用的subscribe方法被编译器误匹配成了接受Subject类型的重载，而非接受Subscriber的版本；
• 你的customSink方法返回值要求是AnyCancellable，但直接返回subscribe(...)得到的是Subscription类型，类型不匹配导致编译器的推断混乱。

另外，你的CustomSubscriber还有个小问题：在receive(subscription:)方法里只保存了订阅对象，但没有主动请求数据，这样即使订阅成功，你也收不到任何输入值哦。

接下来咱们一步步修复代码：

第一步：完善CustomSubscriber的实现 #

我们需要在收到订阅时主动请求数据，同时保持原有逻辑：

class CustomSubscriber<Input, Failure>: Subscriber where Failure: Error {
    typealias Input = Input
    typealias Failure = Failure
    
    // 建议命名更贴合Combine的习惯，当然你也可以保留原来的命名
    private let receiveValue: (Input) -> Void
    private let receiveCompletion: (Subscribers.Completion<Failure>) -> Void
    private var subscription: Subscription?
    
    init(receiveValue: @escaping (Input) -> Void, receiveCompletion: @escaping (Subscribers.Completion<Failure>) -> Void) {
        self.receiveValue = receiveValue
        self.receiveCompletion = receiveCompletion
    }
    
    func receive(subscription: any Subscription) {
        self.subscription = subscription
        // 主动请求所有数据，和内置sink的默认行为一致
        subscription.request(.unlimited)
    }
    
    func receive(_ input: Input) -> Subscribers.Demand {
        receiveValue(input)
        // 返回.none表示不需要更多数据，也可以根据需求返回其他Demand值
        return .none
    }
    
    func receive(completion: Subscribers.Completion<Failure>) {
        receiveCompletion(completion)
        subscription = nil // 完成后清空订阅，避免内存泄漏
    }
}

第二步：修正Publisher的customSink扩展 #

这里要确保调用正确的subscribe重载，并且把Subscription包装成AnyCancellable：

extension Publisher {
    func customSink(
        receiveValue: @escaping (Output) -> Void,
        receiveCompletion: @escaping (Subscribers.Completion<Failure>) -> Void
    ) -> AnyCancellable {
        let subscriber = CustomSubscriber<Output, Failure>(
            receiveValue: receiveValue,
            receiveCompletion: receiveCompletion
        )
        // 明确调用接受Subscriber的subscribe方法
        self.subscribe(subscriber)
        // 将订阅包装成AnyCancellable，支持自动取消
        return AnyCancellable {
            subscriber.subscription?.cancel()
        }
    }
}

第三步：测试你的自定义方法 #

现在你就可以像内置sink一样使用customSink了：

let cancellable = [1, 2, 3].publisher
    .customSink(
        receiveValue: { print("收到值：\($0)") },
        receiveCompletion: { print("完成状态：\($0)") }
    )

这样运行后，你就能看到预期的输出：

收到值：1
收到值：2
收到值：3
完成状态：finished

总结下关键修复点：

1. 给CustomSubscriber添加subscription.request(.unlimited)，主动触发数据发送；
2. 在customSink里手动创建Subscriber实例，明确调用正确的subscribe重载；
3. 将Subscription包装成AnyCancellable，保证和内置sink的API行为一致。

内容来源于stack exchange