RxSwift使用步骤：

创建序列
订阅信号
发送信号

示例代码

    let disposbag: DisposeBag = DisposeBag()
    // 创建序列
    let ob:Observable<Any> = Observable.create { (observer) -> Disposable in
        
        // 发送信号
        observer.onNext("text")
        observer.onCompleted()
        return Disposables.create()
    }
    
    override func touchesBegan(_ touches: Set<UITouch>, with event: UIEvent?) {
        
        // 订阅信号
        ob.subscribe(onNext: { (text) in
            print(text)
        }, onCompleted:{
            print("订阅完成")
        }).disposed(by: disposbag)
    }

执行结果

text
订阅完成

关注点：

创建序列做了什么
订阅信号做了什么
如何将发送信号

接下来进入RxSwift源码世界一探究竟！

创建序列

首先点击create方法进入源码实现

create

很遗憾，只有声明，没有实现······
没事，看注释。在项目里搜索create文件。

create

AnonymousObservable(subscribe)

综上：create方法创建并返回了一个AnonymousObservable对象，我们暂且称之为序列对象A，序列对象A保存了create方法传入的闭包。

订阅信号

点击subscribe方法，进入源码。
上面的序列对象A遵循了ObservableType$协议，所以进入第二个方法查看。

可以看出subscribe方法中创建了一个继承自AnonymousObserver类的观察者observer，我们暂且称它为观察者对象B。观察者对象B保存了EventHandler闭包，监听了传入的onNext,onError,onComplete,onDisposed四个闭包。

我们来看看这句代码

self.asObservable().subscribe(observer),

self其实就是序列对象A，在AnonymousObservable中并没有找到asObservable()实现,通过继承链，AnonymousObservable ->Producer-> Observable，在父类Observable中找到了实现，其实没做什么，只是返回了自己。

asObservable()

继续下一步：AnonymousObservable .subscribe(observer)。通过继承，AnonymousObservable ->Producer，在父类Producer中找到了subscribe方法，

subscribe(observer)

rxAbstractMethod()是个抽象方法，不在父类上实现，回到AnonymousObservable类：

AnonymousObservableSink(observer: observer, cancel: cancel)：

保存observer，即保存了onNext,onError,onComplete,onDisposed四个闭包。

sink.run(self)：

调用run，其中parent是AnonymousObservable的别名，就是创建序列的时候的AnonymousObservable。AnyObserver(self)即AnyObserver(AnonymousObservableSink)，self.observer = observer.on，即：self.observer = AnonymousObservableSink.on，这就将观察者对象B和序列对象A的on事件关联起来了。