observable_observable 延迟执行 - CSDN
精华内容
参与话题
  • Observable入门

    千次阅读 2018-05-24 09:35:16
    1. 什么是Observable和 Observer?1.1 ObservableObservable是能向观察它的人(或外界)在一定时间段内推送值的实体。Observable实体可能会一直推送下去,也可能在某时间点结束,也可能在某时间点抛出error故...

    1.  什么是Observable和 Observer?

    1.1 Observable

    Observable是能向观察它的人(或外界)在一定时间段内推送值的实体。

    Observable实体可能会一直推送下去,也可能在某时间点结束,也可能在某时间点抛出error

    故Observable可能的行为为:

    向observer推送值,告诉observer结束,告诉observer有异常(error)

     

    例子:

    Observale 实体:

    第1步向观察它的实体(observer)推送值1

    第2步向观察它的实体(observer)推送值2

    第3步向观察它的实体(observer)推送值3

    然后 等待10分钟

    第4步向observer推送值4

    第5步告诉observer结束了

     

    实现为编程语言为:

    observer.next(1);

    observer.next(2);

    observer.next(3);

    setTimeout(()=>{

               observer.next(4);

               observer.complete();

    }, 1000*60*10);

     

    1.2 Observer

    是观察Observable的行为后,对其行为进行响应的实体。

    做出的响应有三类:

    对于推送值的响应

    对于结束的响应

    对于observable抛出的error的响应

     

    2.  http请求的Observable的例子

    谁是observable?

    HTTP的response是一个observable:它对观察它的实体在timeout时间内

    推送值或抛出异常

     

    谁是observer?

    就是得到response后,做处理。

    对推送值做处理 或者是 对error做处理

     

    3.  如何生成Observable

    3.1 自己把observable的动作序列编程实现,比如第一个例子

    3.2 已有的返回observable的服务

    比如http的get

     

    3.3 rxjs库里,提供构造简单的observable实体的方法。

    一个简单的observable是一个接一个的推送值,供推送n次

    Rxjs提供的方法有:

    Observable.of(x1,x2,…,xn);

    或者

    Observable.from([x1,x2,…,xn]);

     

    4.  Observable的算符

    【Old Observable】-- operator –> 【New Observable】

     

    Map filter reduce pipe

    4.1 map最简单

    Observablex.map(x => 2*x) 则之前推送为10时,经过算符,推送20

    4.2 filter也不难

    Observablex.filter(x => x%2 === 0) 则推送是偶数时,接着推送;推送是奇数时,则不推送

     

    4.3 todo


    5.  理解Angular中对参数的操作

    ngOnInit(){

    this.hero$ =this.route.paramMap.pipe( switchMap((params:ParamMap)=>this.service.getHero(params.get('id'))));

    }

    注意:ParamMap类型不是Observable; this.route.paramMap返回的是Observable<ParamMap>

    根本目的是 想做一个map,使得输入的每一个 params得到对应的 Hero。

    即 params:ParamMap => Hero

    但是,因为this.service.getHero 返回的是 Observable<Hero>

    所以 需要把返回的flat化。

    所以 当 函数的输入输出类型是  ClassA => ClassB 时,使用map

    当 函数的输入输出类型是 ClassA=> Observable<ClassB>时,使用 switchMap


    展开全文
  • Observable详解

    2019-04-07 09:35:28
    浏览新版,请访问 RxJS Observable 在介绍 Observable 之前,我们要先了解两个设计模式: Observer Pattern - (观察者模式) Iterator Pattern - (迭代器模式) 这两个模式是 Observable 的基础,下面我们先来介绍...
        
    浏览新版,请访问 RxJS Observable

    在介绍 Observable 之前,我们要先了解两个设计模式:

    • Observer Pattern - (观察者模式)
    • Iterator Pattern - (迭代器模式)

    这两个模式是 Observable 的基础,下面我们先来介绍一下 Observer Pattern。

    Observer Pattern

    观察者模式定义

    观察者模式软件设计模式的一种。在此种模式中,一个目标对象管理所有相依于它的观察者对象,并且在它本身的状态改变时主动发出通知。这通常透过呼叫各观察者所提供的方法来实现。此种模式通常被用来实时事件处理系统。 — 维基百科

    观察者模式又叫发布订阅模式(Publish/Subscribe),它定义了一种一对多的关系,让多个观察者对象同时监听某一个主题对象,这个主题对象的状态发生变化时就会通知所有的观察者对象,使得它们能够自动更新自己。

    我们可以使用日常生活中,期刊订阅的例子来形象地解释一下上面的概念。期刊订阅包含两个主要的角色:期刊出版方和订阅者,他们之间的关系如下:

    • 期刊出版方 - 负责期刊的出版和发行工作
    • 订阅者 - 只需执行订阅操作,新版的期刊发布后,就会主动收到通知,如果取消订阅,以后就不会再收到通知

    在观察者模式中也有两个主要角色:Subject (主题) 和 Observer (观察者) 。它们分别对应例子中的期刊出版方和订阅者。接下来我们来看张图,从而加深对上面概念的理解。

    图片描述

    观察者模式优缺点

    观察者模式的优点:

    • 支持简单的广播通信,自动通知所有已经订阅过的对象
    • 目标对象与观察者之间的抽象耦合关系能够单独扩展以及重用

    观察者模式的缺点:

    • 如果一个被观察者对象有很多的直接和间接的观察者的话,将所有的观察者都通知到会花费很多时间
    • 如果在观察者和观察目标之间有循环依赖的话,观察目标会触发它们之间进行循环调用,可能导致系统崩溃

    观察者模式的应用

    在前端领域,观察者模式被广泛地使用。最常见的例子就是为 DOM 对象添加事件监听,具体示例如下:

    <button id="btn">确认</button>
    
    function clickHandler(event) {
        console.log('用户已点击确认按钮!');
    }
    document.getElementById("btn").addEventListener('click', clickHandler);

    上面代码中,我们通过 addEventListener API 监听 button 对象上的点击事件,当用户点击按钮时,会自动执行我们的 clickHandler 函数。

    观察者模式实战

    Subject 类定义:

    class Subject {
        
        constructor() {
            this.observerCollection = [];
        }
        
        registerObserver(observer) {
            this.observerCollection.push(observer);
        }
        
        unregisterObserver(observer) {
            let index = this.observerCollection.indexOf(observer);
            if(index >= 0) this.observerCollection.splice(index, 1);
        }
        
        notifyObservers() {
            this.observerCollection.forEach((observer)=>observer.notify());
        }
    }

    Observer 类定义:

    class Observer {
        
        constructor(name) {
            this.name = name;
        }
        
        notify() {
            console.log(`${this.name} has been notified.`);
        }
    }

    使用示例:

    let subject = new Subject(); // 创建主题对象
    
    let observer1 = new Observer('semlinker'); // 创建观察者A - 'semlinker'
    let observer2 = new Observer('lolo'); // 创建观察者B - 'lolo'
    
    subject.registerObserver(observer1); // 注册观察者A
    subject.registerObserver(observer2); // 注册观察者B
     
    subject.notifyObservers(); // 通知观察者
    
    subject.unregisterObserver(observer1); // 移除观察者A
    
    subject.notifyObservers(); // 验证是否成功移除

    以上代码成功运行后控制台的输出结果:

    semlinker has been notified. # 输出一次
    2(unknown) lolo has been notified. # 输出两次

    需要注意的是,在观察者模式中,通常情况下调用注册观察者后,会返回一个函数,用于移除监听,有兴趣的读者,可以自己尝试一下。(备注:在 Angular 1.x 中调用 $scope.$on() 方法后,就会返回一个函数,用于移除监听)

    Iterator Pattern

    迭代器模式定义

    迭代器(Iterator)模式,又叫做游标(Cursor)模式。它提供一种方法顺序访问一个聚合对象中的各个元素,而又不需要暴露该对象的内部表示。迭代器模式可以把迭代的过程从业务逻辑中分离出来,在使用迭代器模式之后,即使不关心对象的内部构造,也可以按顺序访问其中的每个元素。

    迭代器模式的优缺点

    迭代器模式的优点:

    • 简化了遍历方式,对于对象集合的遍历,还是比较麻烦的,对于数组或者有序列表,我们尚可以通过游标取得,但用户需要在对集合了解的前提下,自行遍历对象,但是对于 hash 表来说,用户遍历起来就比较麻烦。而引入迭代器方法后,用户用起来就简单的多了。
    • 封装性良好,用户只需要得到迭代器就可以遍历,而不用去关心遍历算法。

    迭代器模式的缺点:

    • 遍历过程是一个单向且不可逆的遍历

    ECMAScript 迭代器

    在 ECMAScript 中 Iterator 最早其实是要采用类似 Python 的 Iterator 规范,就是 Iterator 在没有元素之后,执行 next 会直接抛出错误;但后来经过一段时间讨论后,决定采更 functional 的做法,改成在取得最后一个元素之后执行 next 永远都回传 { done: true, value: undefined }

    一个迭代器对象 ,知道如何每次访问集合中的一项, 并记录它的当前在序列中所在的位置。在 JavaScript 中迭代器是一个对象,它提供了一个 next() 方法,返回序列中的下一项。这个方法返回包含 donevalue 两个属性的对象。对象的取值如下:

    • 在最后一个元素前:{ done: false, value: elementValue }
    • 在最后一个元素后:{ done: true, value: undefined }

    详细信息可以参考 - 可迭代协议和迭代器协议

    ES 5 迭代器

    接下来我们来创建一个 makeIterator 函数,该函数的参数类型是数组,当调用该函数后,返回一个包含 next() 方法的 Iterator 对象, 其中 next() 方法是用来获取容器对象中下一个元素。具体示例如下:

    function makeIterator(array){
        var nextIndex = 0;
        
        return {
           next: function(){
               return nextIndex < array.length ?
                   {value: array[nextIndex++], done: false} :
                   {done: true};
           }
        }
    }

    一旦初始化, next() 方法可以用来依次访问可迭代对象中的元素:

    var it = makeIterator(['yo', 'ya']);
    console.log(it.next().value); // 'yo'
    console.log(it.next().value); // 'ya'
    console.log(it.next().done);  // true

    ES 6 迭代器

    在 ES 6 中我们可以通过 Symbol.iterator 来创建可迭代对象的内部迭代器,具体示例如下:

    let arr = ['a', 'b', 'c'];
    let iter = arr[Symbol.iterator]();

    调用 next() 方法来获取数组中的元素:

    > iter.next()
    { value: 'a', done: false }
    > iter.next()
    { value: 'b', done: false }
    > iter.next()
    { value: 'c', done: false }
    > iter.next()
    { value: undefined, done: true }

    ES 6 中可迭代的对象:

    • Arrays
    • Strings
    • Maps
    • Sets
    • DOM data structures (work in progress)

    Observable

    RxJS 是基于观察者模式和迭代器模式以函数式编程思维来实现的。RxJS 中含有两个基本概念:Observables 与 Observer。Observables 作为被观察者,是一个值或事件的流集合;而 Observer 则作为观察者,根据 Observables 进行处理。

    Observables 与 Observer 之间的订阅发布关系(观察者模式) 如下:

    • 订阅:Observer 通过 Observable 提供的 subscribe() 方法订阅 Observable。
    • 发布:Observable 通过回调 next 方法向 Observer 发布事件。

    Proposal Observable

    自定义 Observable

    如果你想真正了解 Observable,最好的方式就是自己写一个。其实 Observable 就是一个函数,它接受一个 Observer 作为参数然后返回另一个函数。

    它的基本特征:

    • 是一个函数
    • 接受一个 Observer 对象 (包含 next、error、complete 方法的对象) 作为参数
    • 返回一个 unsubscribe 函数,用于取消订阅

    它的作用:

    作为生产者与观察者之间的桥梁,并返回一种方法来解除生产者与观察者之间的联系,其中观察者用于处理时间序列上数据流。接下来我们来看一下 Observable 的基础实现:

    DataSource - 数据源

    class DataSource {
      constructor() {
        let i = 0;
        this._id = setInterval(() => this.emit(i++), 200); // 创建定时器
      }
      
      emit(n) {
        const limit = 10;  // 设置数据上限值
        if (this.ondata) {
          this.ondata(n);
        }
        if (n === limit) {
          if (this.oncomplete) {
            this.oncomplete();
          }
          this.destroy();
        }
      }
      
      destroy() { // 清除定时器
        clearInterval(this._id);
      }
    }

    myObservable

    function myObservable(observer) {
        let datasource = new DataSource(); // 创建数据源
        datasource.ondata = (e) => observer.next(e); // 处理数据流
        datasource.onerror = (err) => observer.error(err); // 处理异常
        datasource.oncomplete = () => observer.complete(); // 处理数据流终止
        return () => { // 返回一个函数用于,销毁数据源
            datasource.destroy();
        };
    }

    使用示例:

    const unsub = myObservable({
      next(x) { console.log(x); },
      error(err) { console.error(err); },
      complete() { console.log('done')}
    });
    
    /**
    * 移除注释,可以测试取消订阅
    */
    // setTimeout(unsub, 500); 

    具体运行结果,可以查看线上示例

    SafeObserver - 更好的 Observer

    上面的示例中,我们使用一个包含了 next、error、complete 方法的普通 JavaScript 对象来定义观察者。一个普通的 JavaScript 对象只是一个开始,在 RxJS 5 里面,为开发者提供了一些保障机制,来保证一个更安全的观察者。以下是一些比较重要的原则:

    • 传入的 Observer 对象可以不实现所有规定的方法 (next、error、complete 方法)
    • complete 或者 error 触发之后再调用 next 方法是没用的
    • 调用 unsubscribe 方法后,任何方法都不能再被调用了
    • completeerror 触发后,unsubscribe 也会自动调用
    • nextcompleteerror 出现异常时,unsubscribe 也会自动调用以保证资源不会浪费
    • nextcompleteerror是可选的。按需处理即可,不必全部处理

    为了完成上述目标,我们得把传入的匿名 Observer 对象封装在一个 SafeObserver 里以提供上述保障。SafeObserver 的具体实现如下:

    class SafeObserver {
      constructor(destination) {
        this.destination = destination;
      }
      
      next(value) {
        // 尚未取消订阅,且包含next方法
        if (!this.isUnsubscribed && this.destination.next) {
          try {
            this.destination.next(value);
          } catch (err) {
            // 出现异常时,取消订阅释放资源,再抛出异常
            this.unsubscribe();
            throw err;
          }
        }
      }
      
      error(err) {
        // 尚未取消订阅,且包含error方法
        if (!this.isUnsubscribed && this.destination.error) {
          try {
            this.destination.error(err);
          } catch (e2) {
            // 出现异常时,取消订阅释放资源,再抛出异常
            this.unsubscribe();
            throw e2;
          }
          this.unsubscribe();
        }
      }
    
      complete() {
        // 尚未取消订阅,且包含complete方法
        if (!this.isUnsubscribed && this.destination.complete) {
          try {
            this.destination.complete();
          } catch (err) {
            // 出现异常时,取消订阅释放资源,再抛出异常
            this.unsubscribe();
            throw err;
          }
          this.unsubscribe();
        }
      }
      
      unsubscribe() { // 用于取消订阅
        this.isUnsubscribed = true;
        if (this.unsub) {
          this.unsub();
        }
      }
    }

    myObservable - 使用 SafeObserver

    function myObservable(observer) {
      const safeObserver = new SafeObserver(observer); // 创建SafeObserver对象
      const datasource = new DataSource(); // 创建数据源
      datasource.ondata = (e) => safeObserver.next(e);
      datasource.onerror = (err) => safeObserver.error(err);
      datasource.oncomplete = () => safeObserver.complete();
    
      safeObserver.unsub = () => { // 为SafeObserver对象添加unsub方法
        datasource.destroy();
      };
      // 绑定this上下文,并返回unsubscribe方法
      return safeObserver.unsubscribe.bind(safeObserver); 
    }

    使用示例:

    const unsub = myObservable({
      next(x) { console.log(x); },
      error(err) { console.error(err); },
      complete() { console.log('done')}
    });

    具体运行结果,可以查看线上示例

    Operators - 也是函数

    Operator 是一个函数,它接收一个 Observable 对象,然后返回一个新的 Observable 对象。当我们订阅新返回的 Observable 对象时,它内部会自动订阅前一个 Observable 对象。接下来我们来实现常用的 map 操作符:

    Observable 实现:

    class Observable {
      constructor(_subscribe) {
        this._subscribe = _subscribe;
      }
      
      subscribe(observer) {
        const safeObserver = new SafeObserver(observer);
        safeObserver.unsub = this._subscribe(safeObserver);
        return safeObserver.unsubscribe.bind(safeObserver);
      }
    }

    map 操作符实现:

    function map(source, project) {
      return new Observable((observer) => {
        const mapObserver = {
          next: (x) => observer.next(project(x)),
          error: (err) => observer.error(err),
          complete: () => observer.complete()
        };
        return source.subscribe(mapObserver);
      });
    }

    具体运行结果,可以查看线上示例

    改进 Observable - 支持 Operator 链式调用

    如果把 Operator 都写成如上那种独立的函数,我们链式代码会逐渐变丑:

    map(map(myObservable, (x) => x + 1), (x) => x + 2);

    对于上面的代码,想象一下有 5、6 个嵌套着的 Operator,再加上更多、更复杂的参数,基本上就没法儿看了。

    你也可以试下 Texas Toland 提议的简单版管道实现,合并压缩一个数组的Operator并生成一个最终的Observable,不过这意味着要写更复杂的 Operator,上代码:JSBin。其实写完后你会发现,代码也不怎么漂亮:

    pipe(myObservable, map(x => x + 1), map(x => x + 2));

    理想情况下,我们想将代码用更自然的方式链起来:

    myObservable.map(x => x + 1).map(x => x + 2);

    幸运的是,我们已经有了这样一个 Observable 类,我们可以基于 prototype 在不增加复杂度的情况下支持多 Operators 的链式结构,下面我们采用prototype方式再次实现一下 Observable

    Observable.prototype.map = function (project) {
        return new Observable((observer) => {
            const mapObserver = {
                next: (x) => observer.next(project(x)),
                error: (err) => observer.error(err),
                complete: () => observer.complete()
            };
            return this.subscribe(mapObserver);
        });
    };

    现在我们终于有了一个还不错的实现。这样实现还有其他好处,例如:可以写子类继承 Observable 类,然后在子类中重写某些内容以优化程序。

    接下来我们来总结一下该部分的内容:Observable 就是函数,它接受 Observer 作为参数,又返回一个函数。如果你也写了一个函数,接收一个 Observer 作为参数,又返回一个函数,那么,它是异步的、还是同步的 ?其实都不是,它就只是一个函数。任何函数的行为都依赖于它的具体实现,所以当你处理一个 Observable 时,就把它当成一个普通函数,里面没有什么黑魔法。当你要构建 Operator 链时,你需要做的其实就是生成一个函数将一堆 Observers 链接在一起,然后让真正的数据依次穿过它们。

    Rx.Observable.create

    var observable = Rx.Observable
        .create(function(observer) {
            observer.next('Semlinker'); // RxJS 4.x 以前的版本用 onNext
            observer.next('Lolo');
        });
        
    // 订阅这个 Observable    
    observable.subscribe(function(value) {
        console.log(value);
    });

    以上代码运行后,控制台会依次输出 'Semlinker' 和 'Lolo' 两个字符串。

    需要注意的是,很多人认为 RxJS 中的所有操作都是异步的,但其实这个观念是错的。RxJS 的核心特性是它的异步处理能力,但它也是可以用来处理同步的行为。具体示例如下:

    var observable = Rx.Observable
        .create(function(observer) {
            observer.next('Semlinker'); // RxJS 4.x 以前的版本用 onNext
            observer.next('Lolo');
        });
        
    console.log('start');
    observable.subscribe(function(value) {
        console.log(value);
    });
    console.log('end');

    以上代码运行后,控制台的输出结果:

    start
    Semlinker
    Lolo
    end

    当然我们也可以用它处理异步行为:

    var observable = Rx.Observable
        .create(function(observer) {
            observer.next('Semlinker'); // RxJS 4.x 以前的版本用 onNext
            observer.next('Lolo');
            
            setTimeout(() => {
                observer.next('RxJS Observable');
            }, 300);
        })
        
    console.log('start');
    observable.subscribe(function(value) {
        console.log(value);
    });
    console.log('end');

    以上代码运行后,控制台的输出结果:

    start
    Semlinker
    Lolo
    end
    RxJS Observable

    从以上例子中,我们可以得出一个结论 - Observable 可以应用于同步和异步的场合。

    Observable - Creation Operator

    RxJS 中提供了很多操作符,用于创建 Observable 对象,常用的操作符如下:

    • create
    • of
    • from
    • fromEvent
    • fromPromise
    • empty
    • never
    • throw
    • interval
    • timer

    上面的例子中,我们已经使用过了 create 操作符,接下来我们来看一下其它的操作符:

    of

    var source = Rx.Observable.of('Semlinker', 'Lolo');
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log(error);
        }
    });

    以上代码运行后,控制台的输出结果:

    Semlinker
    Lolo
    complete!

    from

    var arr = [1, 2, 3];
    var source = Rx.Observable.from(arr); // 也支持字符串,如 "Angular 2 修仙之路"
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log(error);
        }
    });

    以上代码运行后,控制台的输出结果:

    1
    2
    3
    complete!

    fromEvent

    Rx.Observable.fromEvent(document.querySelector('button'), 'click');

    fromPromise

    var source = Rx.Observable
      .fromPromise(new Promise((resolve, reject) => {
        setTimeout(() => {
          resolve('Hello RxJS!');
        },3000)
    }));
      
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log(error);
        }
    });

    以上代码运行后,控制台的输出结果:

    Hello RxJS!
    complete!

    empty

    var source = Rx.Observable.empty();
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log(error);
        }
    });

    以上代码运行后,控制台的输出结果:

    complete!

    empty 操作符返回一个空的 Observable 对象,如果我们订阅该对象,它会立即返回 complete 信息。

    never

    var source = Rx.Observable.never();
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log(error);
        }
    });

    never 操作符会返回一个无穷的 Observable,当我们订阅它后,什么事情都不会发生,它是一个一直存在却什么都不做的 Observable 对象。

    throw

    var source = Rx.Observable.throw('Oop!');
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log('Throw Error: ' + error);
        }
    });

    以上代码运行后,控制台的输出结果:

    Throw Error: Oop!

    throw 操作如,只做一件事就是抛出异常。

    interval

    var source = Rx.Observable.interval(1000);
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log('Throw Error: ' + error);
        }
    });

    以上代码运行后,控制台的输出结果:

    0
    1
    2
    ...

    interval 操作符支持一个数值类型的参数,用于表示定时的间隔。上面代码表示每隔 1s,会输出一个递增的值,初始值从 0 开始。

    timer

    var source = Rx.Observable.timer(1000, 5000);
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log('Throw Error: ' + error);
        }
    });

    以上代码运行后,控制台的输出结果:

    0 # 1s后
    1 # 5s后
    2 # 5s后
    ...

    timer 操作符支持两个参数,第一个参数用于设定发送第一个值需等待的时间,第二个参数表示第一次发送后,发送其它值的间隔时间。此外,timer 操作符也可以只传递一个参数,具体如下:

    var source = Rx.Observable.timer(1000);
    
    source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log('Throw Error: ' + error);
        }
    });

    以上代码运行后,控制台的输出结果:

    0
    complete!

    Subscription

    有些时候对于一些 Observable 对象 (如通过 interval、timer 操作符创建的对象),当我们不需要的时候,要释放相关的资源,以避免资源浪费。针对这种情况,我们可以调用 Subscription 对象的 unsubscribe 方法来释放资源。具体示例如下:

    var source = Rx.Observable.timer(1000, 1000);
    
    // 取得subscription对象
    var subscription = source.subscribe({
        next: function(value) {
            console.log(value);
        },
        complete: function() {
            console.log('complete!');
        },
        error: function(error) {
            console.log('Throw Error: ' + error);
        }
    });
    
    setTimeout(() => {
        subscription.unsubscribe();
    }, 5000);

    RxJS - Observer

    Observer (观察者) 是一个包含三个方法的对象,每当 Observable 触发事件时,便会自动调用观察者的对应方法。

    Observer 接口定义

    interface Observer<T> {
      closed?: boolean; // 标识是否已经取消对Observable对象的订阅
      next: (value: T) => void;
      error: (err: any) => void;
      complete: () => void;
    }

    Observer 中的三个方法的作用:

    • next - 每当 Observable 发送新值的时候,next 方法会被调用
    • error - 当 Observable 内发生错误时,error 方法就会被调用
    • complete - 当 Observable 数据终止后,complete 方法会被调用。在调用 complete 方法之后,next 方法就不会再次被调用

    接下来我们来看个具体示例:

    var observable = Rx.Observable
        .create(function(observer) {
                observer.next('Semlinker');
                observer.next('Lolo');
                observer.complete();
                observer.next('not work');
        });
        
    // 创建一个观察者
    var observer = {
        next: function(value) {
            console.log(value);
        },
        error: function(error) {
            console.log(error);
        },
        complete: function() {
            console.log('complete');
        }
    }
    
    // 订阅已创建的observable对象
    observable.subscribe(observer);

    以上代码运行后,控制台的输出结果:

    Semlinker
    Lolo
    complete

    上面的例子中,我们可以看出,complete 方法执行后,next 就会失效,所以不会输出 not work

    另外观察者可以不用同时包含 next、complete、error 三种方法,它可以只包含一个 next 方法,具体如下:

    var observer = {
        next: function(value) {
            console.log(value);
        }
    };

    有时候 Observable 可能是一个无限的序列,例如 click 事件,对于这种场景,complete 方法就永远不会被调用。

    我们也可以在调用 Observable 对象的 subscribe 方法时,依次传入 next、error、complete 三个函数,来创建观察者:

    observable.subscribe(
        value => { console.log(value); },
        error => { console.log('Error: ', error); },
        () => { console.log('complete'); }
    );

    Pull vs Push

    Pull 和 Push 是数据生产者和数据的消费者两种不同的交流方式。

    什么是Pull?

    在 "拉" 体系中,数据的消费者决定何时从数据生产者那里获取数据,而生产者自身并不会意识到什么时候数据将会被发送给消费者。

    每一个 JavaScript 函数都是一个 "拉" 体系,函数是数据的生产者,调用函数的代码通过 ''拉出" 一个单一的返回值来消费该数据。

    const add = (a, b) => a + b;
    let sum = add(3, 4);

    ES6介绍了 iterator迭代器Generator生成器 — 另一种 "拉" 体系,调用 iterator.next() 的代码是消费者,可从中拉取多个值

    什么是Push?

    在 "推" 体系中,数据的生产者决定何时发送数据给消费者,消费者不会在接收数据之前意识到它将要接收这个数据。

    Promise(承诺) 是当今 JS 中最常见的 "推" 体系,一个Promise (数据的生产者)发送一个 resolved value (成功状态的值)来执行一个回调(数据消费者),但是不同于函数的地方的是:Promise 决定着何时数据才被推送至这个回调函数。

    RxJS 引入了 Observables (可观察对象),一个全新的 "推" 体系。一个可观察对象是一个产生多值的生产者,当产生新数据的时候,会主动 "推送给" Observer (观察者)。

    生产者 消费者
    pull拉 被请求的时候产生数据 决定何时请求数据
    push推 按自己的节奏生产数据 对接收的数据进行处理

    接下来我们来看张图,从而加深对上面概念的理解:

    图片描述

    Observable vs Promise

    Observable(可观察对象)是基于推送(Push)运行时执行(lazy)的多值集合。

    MagicQ 单值 多值
    拉取(Pull) 函数 遍历器
    推送(Push) Promise Observable
    • Promise

      • 返回单个值
      • 不可取消的
    • Observable

      • 随着时间的推移发出多个值
      • 可以取消的
      • 支持 map、filter、reduce 等操作符
      • 延迟执行,当订阅的时候才会开始执行

    延迟计算 & 渐进式取值

    延迟计算

    所有的 Observable 对象一定会等到订阅后,才开始执行,如果没有订阅就不会执行。

    var source = Rx.Observable.from([1,2,3,4,5]);
    var example = source.map(x => x + 1);

    上面的示例中,因为 example 对象还未被订阅,所以不会进行运算。这跟数组不一样,具体如下:

    var source = [1,2,3,4,5];
    var example = source.map(x => x + 1); 

    以上代码运行后,example 中就包含已运算后的值。

    渐进式取值

    数组中的操作符如:filter、map 每次都会完整执行并返回一个新的数组,才会继续下一步运算。具体示例如下:

    var source = [1,2,3,4,5];
    var example = source
                    .filter(x => x % 2 === 0) // [2, 4]
                      .map(x => x + 1) // [3, 5]

    关于数组中的 mapfilter 的详细信息,可以参考 - RxJS Functional Programming

    为了更好地理解数组操作符的运算过程,我们可以参考下图:

    图片描述
    查看原图

    虽然 Observable 运算符每次都会返回一个新的 Observable 对象,但每个元素都是渐进式获取的,且每个元素都会经过操作符链的运算后才输出,而不会像数组那样,每个阶段都得完整运算。具体示例如下:

    var source = Rx.Observable.from([1,2,3,4,5]);
    var example = source
                  .filter(x => x % 2 === 0)
                  .map(x => x + 1)
    
    example.subscribe(console.log);

    以上代码的执行过程如下:

    • source 发出 1,执行 filter 过滤操作,返回 false,该值被过滤掉
    • source 发出 2,执行 filter 过滤操作,返回 true,该值被保留,接着执行 map 操作,值被处理成 3,最后通过 console.log 输出
    • source 发出 3,执行 filter 过滤操作,返回 false,该值被过滤掉
    • source 发出 4,执行 filter 过滤操作,返回 true,该值被保留,接着执行 map 操作,值被处理成 5,最后通过 console.log 输出
    • source 发出 5,执行 filter 过滤操作,返回 false,该值被过滤掉

    为了更好地理解 Observable 操作符的运算过程,我们可以参考下图:

    图片描述
    查看原图

    学习资源

    参考资源


    展开全文
  • 2019独角兽企业重金招聘Python工程师标准>>> ...

    1.简介RxJava

    一般我们进行耗时任务,如网络、数据库查询、复杂计算等等,我们都回开启一个线程,然后通过接口回调,获取我们的结果。 但随着我们业务逻辑的越来越复杂,我们就会陷入一个回调地狱,回调里面还有回调,在日后我们维护代码来说简直是噩梦。

    RxJava的出现正式为了解决这个问题而生的,它支持链式调用!

    关键字:异步链式调用观察者模式

    这篇文章主要来记录Observable基本用法

    2.create

    final List<String> list = Arrays.asList(new String[]    
                                        {"one","two","three"});
    
    Observable observable = Observable.create(new 
                            OnSubscribe<List<String>>() {
                @Override
                public void call(Subscriber<? super 
                                    List<String>> subscriber) {
                    subscriber.onNext(list);
                    subscriber.onCompleted();
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    可以发现,我们发射的是以整个List<String> 我们可以发射一个一个对象吗?

    当然可以:

            Observable observable = Observable.create(new 
                OnSubscribe<String>() {
                @Override
                public void call(Subscriber<? super String> 
                subscriber) {
                    for (String str:list) {
                        subscriber.onNext(str);
                    }
                    subscriber.onCompleted();
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    这样看起来好像还不是很优雅!有没有办法刚优雅呢? 那我们来看看from这个方法

    3.from

    Observable.from(list).subscribe(new Observer<String>() {
    
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }
    
                @Override
                public void onError(Throwable arg0) {
    
                }
    
                @Override
                public void onNext(String string) {
                    System.out.println(string);
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    结果是:

    one
    two
    three
    onCompleted
    
    • 1
    • 2
    • 3
    • 4
    • 5

    符合我们的预期!

    4.just

    如果我只想发射list中的第二和第三位可以吗?

    当然可以,我们可以借助just方法:

            Observable.just(list.get(1), list.get(2)).subscribe(new Observer<String>() {
    
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }
    
                @Override
                public void onError(Throwable arg0) {
    
                }
    
                @Override
                public void onNext(String string) {
                    System.out.println(string);
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    结果:

    two
    three
    onCompleted
    
    • 1
    • 2
    • 3
    • 4

    符合我们的预期!

    备注:just方法可以接受1-10个参数

    5.repeat

    如果我们想将整个list重复发射两次或者三次呢?

    Observable.from(list).repeat(2).subscribe(new 
            Observer<String>() {
    
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }
    
                @Override
                public void onError(Throwable arg0) {
    
                }
    
                @Override
                public void onNext(String string) {
                    System.out.println(string);
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18

    结果:

    one
    two
    three
    one
    two
    three
    onCompleted
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    符合我们预期

    备注:repeat可以不传参,效果是:无限循环

    6.range

    从X按顺序输出Y位数字?

    Observable.range(88,10).subscribe(new Observer<Integer>() {
    
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }
    
                @Override
                public void onError(Throwable arg0) {
    
                }
    
                @Override
                public void onNext(Integer arg0) {
                    System.out.println(arg0+"");
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    onCompleted
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    7.interval(测试失败)

    间隔时间发射:

            Observable.interval(3, TimeUnit.SECONDS).subscribe(new Subscriber<Long>() {
    
                @Override
                public void onCompleted() {
                    System.out.println("onCompleted");
                }
    
                @Override
                public void onError(Throwable arg0) {
    
                }
    
                @Override
                public void onNext(Long arg0) {
                    System.out.println(arg0+"");
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    很奇怪,这个方法并没有起作用。

    8.timer(测试失败)

    延迟发射:

    Observable.timer(1, TimeUnit.SECONDS).subscribe(new Observer<Long>() {
    
                @Override
                public void onCompleted() {
                    // TODO Auto-generated method stub
    
                }
    
                @Override
                public void onError(Throwable arg0) {
                    // TODO Auto-generated method stub
    
                }
    
                @Override
                public void onNext(Long arg0) {
                    System.out.println(arg0+"");
                }
            });
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    这个方法和interval一样,也是不能测试。

    它还有一个三个参数的方法timer(3,3,TimeUnit.SECONDS) 延迟3秒之后,每隔3秒发射一次

    转载于:https://my.oschina.net/carbenson/blog/901785

    展开全文
  • Observable

    2019-06-11 19:06:37
    ②订阅 ObservableObservable 事件监听 ④Observable 的Dispose ①可被监听的序列-概念 ❶Observable理解 Observable<T> 类用于描述元素异步产生的序列,既可观察序列,是 Rx 框架的基础 作用是:异步...

    ①可被监听的序列-概念

    ❶Observable理解

    • Observable<T> 类用于描述元素异步产生的序列,既可观察序列,是 Rx 框架的基础

    • 作用是:异步地产生一系列的 Event(事件),即一个 Observable<T>对象会随着时间推移不定期地发出 event(element : T)

    • Event 还可以携带数据,它的泛型<T> 就是用来指定这个 Event 携带的数据的类型

    • 每一个Observable的实例都是一个序列

    • Observable序列相比于Swift序列的关键优势点在于它能够异步地接收元素

    • 有可观察序列,还需要一个 Observer(订阅者)来订阅它,这个订阅者才能收到 Observable<T>不时发出的 Event

    Observable(ObservableType) 等效于Sequence
    observableType.subscribe(_:) 方法等效于Sequence.makeIterator()
    ObservableType.subscribe(_:) 接收一个观察者ObserverType参数,它将被订阅自动接收由可观察到的序列事件和元素,而不是在返回的生成器上手动调用next()
    • 如果一个Observable发出一个next事件(Event.next(Element)),它还能够继续发出更多的事件
    • 如果一个Observable发出一个error事件(Event.error(ErrorType))或者一个completed事件(Event.completed),那么这个Observable序列就不能给订阅者发送其他的事件了

    ❷创建序列

    框架已经创建好了许多常用的序列。例如:button的点击,textField的当前文本,switch的开关状态,slider的当前数值等等

    一些自定义的序列是需要自己创建的

    • 创建序列最直接的方法就是调用 Observable.create

    创建一个 [0, 1, ... 4] 的序列

    • 调用 Observable.create,在构建函数里面描述元素的产生过程

    • observer.onNext(0)就代表产生了一个元素,他的值是 0

    • 又产生 4个元素分别是 1, 2, ... 4

    • 最后,用 observer.onCompleted() 表示元素已经全部产生,没有更多元素了

    let numbers: Observable<Int> = Observable.create { observer -> Disposable in
          observer.onNext(0)  
          observer.onNext(1)
          observer.onNext(2)
          observer.onNext(3)
          observer.onNext(4)
          observer.onCompleted()
         return Disposeable.create()
      }
    复制代码

    可以用这种方式来封装功能组件,例如:闭包回调

    • 在闭包回调中,如果任务失败,就调用 observer.onError(error!)
    • 如果获取到目标元素,就调用 observer.onNext(jsonObject)
    • 由于这个序列只有一个元素,所以在成功获取到元素后,就直接调用 observer.onCompleted() 来表示任务结束
    • 最后 Disposables.create { task.cancel() } 说明如果数据绑定被清除(订阅被取消)的话,就取消网络请求。
    typealias JSON = Any
    let json: Observable<JSON> = Observable.create { (observer) -> Disposable in
        let task = URLSession.shared.dataTask(with: ...) { data, _, error in
            guard error == nil else {
                observer.onError(error!)
                return
            }
            guard let data = data,
                let jsonObject = try? JSONSerialization.jsonObject(with: data, options: .mutableLeaves)
                else {
                observer.onError(DataError.cantParseJSON)
                return
            }
            observer.onNext(jsonObject)
            observer.onCompleted()
        }
        task.resume()
        return Disposables.create { task.cancel() }
    }
    复制代码

    以上就将传统的闭包回调转换成序列。然后用 subscribe 方法来响应这个请求的结果

    json.subscribe(
        onNext: { json in
            print("取得 json 成功: \(json)")}, 
        onError: { error in
            print("取得 json 失败 Error: \(error.localizedDescription)")},
        onCompleted: {
            print("取得 json 任务成功完成")}
        )
        .disposed(by: disposeBag)
    复制代码

    ❸Event

    事件 Event 的定义

    • Event 就是一个枚举,一个 Observable 是可以发出 3 种不同类型的 Event 事件
    public enum Event<Element>{
        case next(Element)
        case error(Swift.Error)
        case completed
    }
    复制代码

    next

    • next 事件就是那个可以携带数据 的事件,可以说它就是一个“最正常”的事件
    1--2--3--->
    复制代码

    error

    • error 事件表示一个错误,它可以携带具体的错误内容,一旦 Observable 发出了 error event,则这个 Observable 就等于终止了,以后它再也不会发出 event 事件了
    -------1-----2------ x  
    复制代码

    completed

    • completed:completed 事件表示 Observable 发出的事件正常地结束了,跟 error 一样,一旦 Observable 发出了 completed event,则这个 Observable 就等于终止了,以后它再也不会发出 event 事件了
    ------1 ------2--------3-------|
    复制代码

    ❹Observable 与 Sequence比较

    • 可以把每一个 Observable 的实例想象成于一个 Swift 中的 Sequence

    • 即一个 Observable(ObservableType)相当于一个序列 Sequence(SequenceType)

    • ObservableType.subscribe(_:) 方法其实就相当于 SequenceType.generate()

    区别

    • Swift 中的 SequenceType 是同步的循环,而 Observable 是异步的
    • Observable 对象会在有任何 Event 时候,自动将 Event 作为一个参数通过 ObservableType.subscribe(_:) 发出,并不需要使用 next 方法

    ②订阅 Observable

    • 有了Observable,还要用 subscribe() 方法来订阅它,接收它发出的 Event

    订阅方法1

    ❶使用 subscribe() 订阅一个 Observable 对象

    • 该方法的 block 的回调参数就是被发出的 event 事件,将其直接打印出来
    let observable = Observable.of("A","B","C")
    observable.subscribe { event in
                         print(event)}
    //print
    next(A)
    next(B)
    next(C)
    completed
    复制代码

    ❷如果获取到事件里的数据,通过 event.element

    let observable = Observable.of("A","B","C")
    observable.subscribe{ event in
                        print(eventt.element)}????????????
    //print
    Optional("A")
    Optional("B")
    Optional("C")	q
    nil
    复制代码

    订阅方法2

    RxSwift还提供另一个subscribe 方法,它可以把 event 进行分类

    • 通过不同的 block 回调处理不同类型的 event(其中 onDisposed 表示订阅行为被 dispose 后的回调)
    • 同时会把 event 携带的数据直接解包出来作为参数,方便我们使用
    let observable = Observable("A","B","C")
    observable.subscribe(
        onNext:{ element in
                  print(element) },
        onError:{error in
                print(error)},
        onCompleted:{
            print("completed")},
        onDisposed:{
            print("disposed")
        }
    )
    //print
    A
    B
    C
    completed
    disposed
    复制代码
    • subscribe() 方法的 onNext、onError、onCompleted 和 onDisposed 这四个回调 block 参数都是有默认值的,即它们都是可选的。我们可以只处理 onNext 而不管其他的情况
    let observable = Observable.of("A","B","C")
    observable.subscribe(
       onNext:{element in
              print(element)}
    )
    //print
    A
    B
    C
    复制代码

    ③Observable 事件监听

    监听事件的生命周期

    doOn 介绍

    使用 doOn 方法来监听事件的生命周期,它会在每一次事件发送前被调用

    同时它和 subscribe 一样,可以通过不同的 block 回调处理不同类型的 event

    • do(onNext:) 方法就是在 subscribe(onNext:) 前调用
    • 而 do(onCompleted:) 方法则会在 subscribe(onCompleted:) 前面调用

    栗子

    let observable = Observable.of("A","B","C")
    observable.do(
                onNext: { element in 
                        print("Intercepted Next:", element)},
                onError: {error in
                           print("Intercepted Error:", error)},
                onCompleted: {
                     print("Intercepted Completed")
                },
                onDispose: {
                    print("Intercepted Disposed")
                }
                )
               .subscribe(
               onNext: {element in
                       print(element)},
                   onError: {error in
                            print(error)},
                   onCompleted:{
                       print("completed")},
                   onDisposed:{
                       print("disposed")
                   }
               )
    复制代码

    ④Observable 的Dispose

    ❶Observable 从创建到终结流程

    一个 Observable 序列被创建出来后它不会马上就开始被激活从而发出 Event,而是要等到它被某个人订阅了才会激活它 而Observable序列激活之后要一直等到它发出了 .error 或者 .completedevent后,它才被终结

    ❷dispose() 方法

    除了 dispose() 方法之外,更经常用 DisposeBag 的对象来管理多个订阅行为的销毁

    • 以把一个DisposeBag对象看成一个垃圾袋,把用过的订阅行为都放进去,感觉有点像OC释放池
    • 而这个DisposeBag就会在自己快要dealloc的时,对它里面的所有订阅行为都调用 dispose() 方法
    let disposeBag = DisposeBag()
    //第1个Observable,和订阅
    let observable1 = Observable.of("A","B","C")
    observable1.subscribe{ event in
                         print(event)}
               .disposed(by: disposeBag)
    //第2个Observable,和订阅
    let observable2 = Obserable.of(1,2,3)
    observable2.subscribe{event in
                         print(event)}
               .disposed(by: disposeBag)
    复制代码

    转载于:https://juejin.im/post/5be80ac4e51d457844615d24

    展开全文
  • 实例讲解:Observable

    千次阅读 2014-03-28 02:02:40
    基础概念二:Observable与computed 纵观KO的所有应用场景,基本上这2个属性至少会用到一个。个人认为这是KO最常使用的东西。他们用法如下: Observable(监控属性):监控自身属性的变化,向外部发送通知。外部...
  • Observable 在开始讲服务之前,我们先来看一下一个新东西——Observable(可观察对象),是属于RxJS库里面的一个对象,可以用来处理异步事件,例如HTTP请求(实际上,在Angular中,所有的HTTP请求返回的都是Obse...
  • observable()方法

    千次阅读 2017-09-28 17:18:58
    observable()方法,用于设置监控属性,这样就可以监控viewModule中的属性值的变化,从而就可以动态的改变某个元素中的值 viewModule对象的形式有两种,一种可以是var obj={};的形式,另一种是通过new 实例化的对象,...
  • Observable与Observer

    万次阅读 2010-07-20 20:12:00
    在Java中通过Observable类和Observer接口实现了观察者模式。一个Observer对象监视着一个Observable对象的变化,当Observable对象发生变化时,Observer得到通知,就可以进行相应的工作。例如在文档/视图结构中...
  • observable

    2017-08-09 15:47:57
    1.observable.partition将输入根据函数分为两部分,前者为符合函数条件的,后者为不符合条件的 例如: let [customerSource, tenantSource] = this.formGroup.valueChanges.partition(type => type !== "hudheud...
  • 在Rxjava中Observable 有 Hot 和 Cold之分.  Hot Observable:  Hot Observable无论有没有观察者进行订阅事件始终都会发生. 当Hot Observable 有多个订阅者时(即多个订阅者进行订阅时), Hot Observable与订阅者...
  • RxJava 合并组合两个(或多个)Observable数据源

    万次阅读 多人点赞 2017-02-07 12:48:05
    关键词:合并 Observable前言在RxJava中, 同时处理多个Observables是很常见的一种操作。下面我们简单分析下几个组合类的操作符。Merge在异步的世界经常会创建这样的场景,我们有多个来源但是只想有一个结果:多输入...
  • Angular2 Observable和RxJS

    千次阅读 2018-08-07 23:08:29
    看到Observable和RxJS就感觉很亲切,因为之前做Android开发的时候接触过RxJava。Observable和RxJS的相关知识最好的文档还是官方文档 https://www.angular.cn/guide/observables 强烈推荐大家看官方文档。 一、可...
  • Observable ###基础 概述 一个观察者订阅(Observer)一个可观察对象(Observable)。观察者对观察对象发射的数据或数据序列做出响应 普通方法调用流程 调用某方法 用变量保存方法保存结果 使用变量新值继续操作 ...
  • observable 中的fliter 方法 传入一个Predicate对象 实现方法并返回一个boolean值。作用是 在interval 循环的时候 进行判断 如果返回结果为true 则执行onnNext 方法 否则继续循环;take方法需要传入一个int值作用...
  • RxJava 驯服数据流之 hot & cold Observable

    千次阅读 2016-07-06 15:10:29
    Observable 数据流有两种类型:hot 和 cold。这两种类型有很大的不同。本节介绍他们的区别,以及作为 Rx 开发者应该如何正确的使用他们。Cold observables只有当有订阅者订阅的时候, Cold Observable 才开始执行...
  • 官方解释: ...observable.deep: 所有 observable 都使用的默认的装饰器。它可以把任何指定的、非原始数据类型的、非 observable 的值转换成 ...observable.ref: 禁用自动的 observable 转换,只是创建一个 observabl...
  • ES6 rxjs Observable

    千次阅读 2018-04-12 11:33:00
    一、拉取(Pull) 、 推送(Push)介绍 Observable 之前,要先了解两个设计模式:Observer Pattern - (观察者模式)Iterator Pattern - (迭代器模式)这两个模式是 Observable 的基础。Observable(可观察对象)是基于推...
  • 【java】Observer和Observable详解

    万次阅读 多人点赞 2017-09-25 19:20:25
    1.1 观察者模式是oo设计中经常用到的模式之一,大家在解决实际需求时,观察者模式往往都会用到,而javase中已经提供了Observer接口和Observable类让你简单快速的实现观察者模式,因此有必要去了解Observer和...
  • public void invalidated(Observable observable); } 什么是对象无效? 被观察对象正在发生改变,或绑定了其他对象 可观察对象 Observable public interface Observable { void addListener(In...
  • 转载 ... 同样是做异步,为什么人们用它,而不用现成的 AsyncTask / Handler / XXX / ...... RxJava 好在哪?就好在简洁,好在那把什么复杂逻辑都能穿成一条线的简洁。
1 2 3 4 5 ... 20
收藏数 52,097
精华内容 20,838
关键字:

observable