深入理解RxSwift

黑客技术

点击右侧关注，了解黑客的世界！

Java 开发进阶

点击右侧关注，掌握进阶之路！

Python开发

点击右侧关注，探讨技术话题！

作者丨嘻嘻z

来源丨掘金

链接：

https://juejin.im/post/5d93541ae51d45782632e378

简介

本篇重点在于深入RxSwift的部分常用特性，所以希望读者在了解RxSwift官方的基本讲解与Demo之后再进行阅读。

RxSwift版本为5.0.0以上。

Dispose

在写代码的时候，我们经常会在很多情况下创建一个Observable

_ = Observable<String>.create { observerOfString -> Disposable in
    observerOfString.on(.next("😬"))
    observerOfString.on(.completed)
    return Disposables.create()
}

又或者是在监听 Observable ：

let disposeBag = DisposeBag()

Observable<Int>.empty()
    .subscribe { event in
        print(event)
    }
    .disposed(by: disposeBag)

这里总是看到是 Disposables 和 DisposeBag ，那么它们到底是什么。

Disposables

Disposables是一个简单的结构体

public struct Disposables {
    private init() {}
}

通过extension生成create方法, 创建四种不同的Disposable。

Disposable定义：

  
 
   
  
  
 
  
 
   
  public protocol Disposable {
    /// Dispose resource.
    func dispose()
}

Cancelable , 在 Disposable 的基础上添加了 isDisposed 属性，便于追踪 Disposable 的状态:

  
 
   
  
  
 
  
 
   
  public protocol Cancelable : Disposable {
    /// Was resource disposed.
    var isDisposed: Bool { get }
}

NopDisposable , 在disposal时什么都不做。

AnonymousDisposable , 创建时传入一个 public typealias DisposeAction = () -> Void 类型的闭包，在disposal时调用。

BinaryDisposable , 在创建时传入两个 Disposable 类型的参数，在disposal时调用。

CompositeDisposable , 在创建时传入多个 Disposable 类型的参数，在disposal时调用。

除了这四种可以通过快速创建的，还有其他类型的 Disposable

BooleanDisposable ，主要用于追踪disposal的状态，初始化时传入Bool类型的参数，表示是否已经被Dispose。

SubscriptionDisposable ，主要在Subject中使用。

RefCountDisposable ，初始化时传入一个 Disposable 类型的参数，如同命名，内存采用了引用计数的管理方法，调用retain方式时计数+1，调用release时计数-1，引用计数不能小于0，当等于0时，如果没有调用过dispose方法（其实是将内部的 _primaryDisposed 属性标记为 true ），也不会触发dispose方法，同样如果引用计数不为0，调用dispose方法也不会触发内部的 Disposable 的dispose。

ScheduledDisposable ，初始化时传入一个 Disposable 类型的参数与一个 ImmediateSchedulerType ，指定传入的 Disposable 在某个线程上执行dispose。

SerialDisposable : 可以手动替换内部的 Disposable ，如果在 _isDisposed 为false的状态，替换后会自动触发之前的 Disposable 的dispose方法，如果为true，则直接触发替换的 Disposable 的dispose方法。

SingleAssignmentDisposable ：对内部的 Disposable 只能设定一次，若设定多次会报错，在没有设定的情况下触发dispose方法也会报错。

DisposeBag

DisposeBag 如同名字，是一个 Bag 类型的数据结构，里面存放 Disposable 的数据。

每次看到Disposable类型调用 disposed(by: disposeBag) ，其实是将Disposable放在DisposeBag中进行管理，当DisposeBag进行dispose时，对其中管理的Disposable分别dispose，但是DisposeBag不能主动调用，只能在deinit时自动释放，所以想要进行dispose操作，只能将disposeBag重新赋值，比如：

  
 
   
  
  
 
  
 
   
  disposeBag = DisposeBag()

Subject 简介

Subject是在RX的某些实现中可用的桥接或代理。

通过继承了 ObserverType ，可以成为一个观察者，可以通过 on(_ event: Event<Element>) 发送事件。

通过继承了 Observable ，可以成为一个被观察者，可以通过 subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element 推送事件。

Subject都是通过 Broadcasts 的方式传播事件。

冷热信号

在RAC中通过Signal与SignalProducer区分冷热信号的概念，RxSwift其实也有冷热信号。

冷信号是指在被观察之后才会推送事件。

热型号是指就算没有被观察，也能推送事件。

可以理解为一个主动一个被动。

这是一个比较抽象的例子：

  
 
   
  
  
 
  
 
   
  let subject = Subject()
subject.onNext("1")
subject.subcribeNext { value in
    print(value)
}
subject.onNext("2")
如果是冷信号，这时候会打印1 2，热信号则只会打印2.

各种Subject

在Rx中有四种Subject。分别是 BehaviorSubject ， PublishSubject ， ReplaySubject ， AsyncSubject 。

下列图标中，圆代表next信号，竖线表示complete，x表示error。

PublishSubject 就是一个热信号的Subject。

BehaviorSubject 是一个有初始值，缓存数量为1的Subject。在5.0版本之前的RxSwift中，有一个叫做 Variable 的属性，在5.0之后，因为相同的特性，完全被 BehaviorSubject 取代。

ReplaySubject 是一个可以自定义缓存数量的Subject。当设定缓存数量为0时，几乎可以当做 PublishSubject 使用，缓存数量为1时，可以当做一个没有初始值的 BehaviorSubject 使用。

AsyncSubject ，只会在接收到Complete事件后，才会Subscribe最后一个信号，如果没有在onComplete之前没有onNext事件，或者触发了onError，则不会触发Subscribe。

Demo

简单的按钮处理，当用户名长度大于4位且密码长度大于6位时，登录按钮才能点击:

  
 
   
  
  
 
  
 
   
  private var disposeBag = DisposeBag()
private let nameSubject = BehaviorSubject<String>(value: "")
private let passwordSubject = BehaviorSubject<String>(value: "")
private let loginSubject = BehaviorSubject<Bool>(value: false)
private let nameTextField = UITextField()
private let passwordTextField = UITextField()
private let loginButton = UIButton()
......
//中间内容省略
......
nameTextField.rx.text.orEmpty.bind(to: nameSubject).disposed(by: disposeBag)
passwordTextField.rx.text.orEmpty.bind(to: passwordSubject).disposed(by: disposeBag)
Observable<Bool>.combineLatest(nameSubject, passwordSubject) { (name, password) -> Bool in
    return name.count > 4 && password.count > 6
}.bind(to: loginButton.rx.isEnabled).disposed(by: disposeBag)

通过某个实时持续的交互，每秒刷新UI:

private let socketSubject = ReplaySubject<String>.create(bufferSize: 1)
......
DoSomething { 
    socketSubject.onNext(value)
}

Observable<Int>
    .interval(.seconds(1), scheduler: MainScheduler.instance)
    .withLatestFrom(socketSubject)
    .distinctUntilChanged()
    .subscribe(onNext: { (value) in
        print(value)
    }).disposed(by: disposeBag)

思考

如果 ReplaySubject 的bufferSize为1，是否与 BehaviorSubject 相同？

如果 ReplaySubject 的bufferSize为0，是否与 PublishSubject 相同？

Schedulers

Schedulers是RxSwift中的调度机制。

主要运算符只有两个 observeOn 以及 subscribeOn 。

  
 
   
  
  
 
  
 
   
  sequence1
  .observeOn(backgroundScheduler)
  .map { n in
      print("This is performed on the background scheduler")
  }
  .observeOn(MainScheduler.instance)
  .map { n in
      print("This is performed on the main scheduler")
  }
  .subscribeOn(subscribeScheduler)
  .subscribe { _ in
      print("This is performed on the subscribeScheduler")
    }

通过 observeOn 与 subscribeOn 可以控制处理信号的线程，并且可以支持多次切换。

各种Scheduler

CurrentThreadScheduler （串行）。指代当前线程，若没有指定Schuduler，则默认使用。

MainScheduler （串行）。主线程，通常使用进行UI操作。在subscribeOn时更应该用做过优化的 ConcurrentMainScheduler 。

SerialDispatchQueueScheduler （串行）。串行线程，主线程也是一种串行线程。

ConcurrentDispatchQueueScheduler （并发）。在初始化时也可以传入串行dispatch queue，也不会有任何问题。适用于需要在后台的工作。

OperationQueueScheduler （并发）。是 NSOperationQueue 的一种抽象示例。适用与单个任务量大的任务并行处理的情况，同时希望设定 maxConcurrentOperationCount 。

思考

为什么说在subscribeOn时更应该用做过优化的 ConcurrentMainScheduler ？

在阅读文档时，经常看到在observeOn时，如果指定了使用 SerialDispatchQueueScheduler ，会有所优化，具体是怎么优化的？

对于第二个问题，在阅读代码时，发现在 observeOn 时做了单独的处理

  
 
   
  
  
 
  
 
   
  public func observeOn(_ scheduler: ImmediateSchedulerType)
    -> Observable<Element> {
        if let scheduler = scheduler as? SerialDispatchQueueScheduler {
            return ObserveOnSerialDispatchQueue(source: self.asObservable(), scheduler: scheduler)
        }
        else {
            return ObserveOn(source: self.asObservable(), scheduler: scheduler)
        }
}