RXSwift - 链接 operator publish、m

默认情况下不同的订阅者这不共享 Observable，不同的订阅者订阅的，是自己的 Observable 。如果想共享 Observable 就需要以下的 connectable operator。

Publish 发布共享 Observable

        let  interval = Observable<Int>.interval(DispatchTimeInterval.seconds(1), scheduler: MainScheduler.instance).publish()
        
        _ = interval.subscribe(onNext: { (num) in
            print("Subscriber1: \(num)")
        })
        
        _ = interval.connect()
        
        DispatchQueue.global().asyncAfter(deadline: DispatchTime.now() + 2) {
            _ = interval.subscribe(onNext: { (num) in
                print("Subscriber2: \(num)")
            })
        }

执行过程

interval
-----------------0-------1--------2-------3-------4-------5-------6--------------
 
Subscriber1
-----------------0-------1--------2-------3-------4-------5-------6--------------
 
Subscriber2
----------------------------------2--------3------4-------5-------6-------------
 
                                            ....publish()
     
----------------0---------1-----(2,2)---(3,4)----(4,4)----(5,5)----(6,6)----------

note

• 一、在 next(0) 的时候 Subscriber 1 开始监听 。

• 二、因为 Subscriber 2 延时 两秒，所以在 next(1) Subscriber 2 没有监听到，只有Subscriber 1 监听。

• 三、在第三秒开始 Subscriber 2 也加入监听，所以结果都是 next(2)。（因为 Publish 共享了 observable 所以 结果都是 next(2)。不如没有用 Publish 共享，Subscriber 1 监听的是 next(2)，而 Subscriber 2 监听的是 next(0)

multicast 共享并监管 observable

        let supervisor = PublishSubject<Int>()
        
        let  interval = Observable<Int>.interval(DispatchTimeInterval.seconds(1), scheduler: MainScheduler.instance).multicast(supervisor)
        
        _ = supervisor.subscribe(onNext: { (num) in
            print("supervisor: \(num)")
        })
        
        _ = interval.subscribe(onNext: { (num) in
            print("     Subscriber1: \(num)")
        })
        
        _ = interval.connect()
        
        let current = DispatchTime.now()
        
        DispatchQueue.global().asyncAfter(deadline: current + 2) {
            _ = interval.subscribe(onNext: { (num) in
                print("     Subscriber2: \(num)")
            })
        }
        
        DispatchQueue.global().asyncAfter(deadline: current + 4) {
            _ = interval.subscribe(onNext: { (num) in
                print("     Subscriber3: \(num)")
            })
        }

执行过程

interval
-----------------0-------------1--------------2-------------3-------------4-----------------
     
supervisor
-----------------0-------------1--------------2-------------3-------------4-----------------
     
 
Subscriber1
-----------------0-------------1--------------2-------------3-------------4-----------------
 
Subscriber2
----------------------------------------------2--------------3-------------4-----------------
     
Subscriber2
--------------------------------------------------------------------------4-----------------
 
                              .....multicast(supervisor)
     
----------------(0,0)--------(1,1)----------(2,2,2)--------(3,3,3)-------(4,4,4,4)--------------

note

• supervisor 存在于每组监听里，虽然在第二秒和第四秒 ，分别新加入了两个监听者。但是他们监听的observable 都是同一个。