资讯专栏INFORMATION COLUMN

RxJS Observable - 一个奇特的函数

omgdog / 607人阅读

摘要:形式上比普通函数直接返回值啰嗦一些。这里和的重要区别在于,模式下,可以决定什么时候返回值,以及返回几个值即调用回调函数的次数。

前言

RxJS 的 Observable 有点难理解,其实 RxJS 相关的概念都有点难理解。毕竟 RxJS 引入了响应式编程这种新的模式,会不习惯是正常的。不过总得去理解嘛,而认识新的事物时,如果能够参照一个合适的已知事物比对着,会比较容易理解吧。对于 Observable,类比 JS 中的函数,还是比较好的。

开始 封装

先来看一个普通函数调用的例子:

function foo() {
  console.log("process...")
}

foo()
// 输出:
// process...

很简单,函数 foo() 封装了一段逻辑(这里只是向控制台输出),然后通过调用函数,函数执行内部的逻辑。

再来看 RxJS Observable 的一个例子:

var foo = Rx.Observable.create(() => {
  console.log("process...")
})

foo.subscribe()
// 输出:
// process...

上例中,通过 Rx.Observable.create() 来创建 Observable 对象,将同样将一段代码逻辑封装到 Observable 对象 foo 中,然后通过 foo.subscribe() 来执行封装的代码逻辑。

对于普通函数和 Observable 对象,封装的代码逻辑在每次调用时都会重新执行一次。从这一点来看,Observable 能够和普通函数一样实现封装代码进行复用。

返回值

函数调用后可以有返回值:

function foo() {
  console.log("process...")
  return 42
}

console.log(foo())
// 输出:
// process...
// 42

Observable 执行后也会产生值,不过和函数直接返回的方式不同,要通过回调函数方式获取:

var foo = Rx.Observable.create((observer) => {
  console.log("process...")
  observer.next(42)
})

foo.subscribe(value => console.log(value))
// 输出:
// process...
// 42

Observable 对象内部是通过 observer.next(42) 这种方式返回值,而调用方则通过回调函数来接收返回的数据。形式上比普通函数直接返回值啰嗦一些。

从调用方的角度来看,两个过程分别是:

普通函数:调用 > 执行逻辑 > 返回数据

Observable:订阅(subscribe) > 执行逻辑 > 返回数据

从获取返回值方式来看,调用函数是一种直接获取数据的模式,从函数那里“拿”(pull)数据;而 Observable 订阅后,是要由 Observable 通过间接调用回调函数的方式,将数据“推”(push)给调用方。

这里 pull 和 push 的重要区别在于,push 模式下,Observable 可以决定什么时候返回值,以及返回几个值(即调用回调函数的次数)。

var foo = Rx.Observable.create((observer) => {
  console.log("process...")
  observer.next(1)
  setTimeout(() => observer.next(2), 1000)
})

console.log("before")
foo.subscribe(value => console.log(value))
console.log("after")
// 输出:
// before
// process...
// 1
// after
// 2

上面例子中,Observable 返回了两个值,第1个值同步返回,第2个值则是过了1秒后异步返回。

也就是说,从返回值来说,Observable 相比普通函数区别在于:

可以返回多个值

可以异步返回值

异常处理

函数执行可能出现异常情况,例如:

function foo() {
  console.log("process...")
  throw new Error("BUG!")
}

我们可以捕获到异常状态进行处理:

try {
   foo()
} catch(e) {
  console.log("error: " + e)
}

对于 Observable,也有错误处理的机制:

var foo = Rx.Observable.create((observer) => {
  console.log("process...")
  observer.error(new Error("BUG!"))
})

foo.subscribe(
  value => console.log(value),
  e => console.log("error: " + e)
)

Observable 的 subscribe() 方法支持传入额外的回调函数,用于处理异常情况。和函数执行类似,出现错误之后,Observable 就不再继续返回数据了。

subscribe() 方法还支持另一种形式传入回调函数:

foo.subscribe({
  next(value) { console.log(value) },
  error(e) { console.log("error: " + e) }
})

而这种形式下,传入的对象和 Observable 内部执行函数中的 observer 参数在形式上就比较一致了。

中止执行

Observable 内部的逻辑可以异步多个返回值,甚至返回无数个值:

var foo = Rx.Observable.create((observer) => {
  let i = 0
  setInterval(() => observer.next(i++), 1000)
})

foo.subscribe(i => console.log(i))
// 输出:
// 0
// 1
// 2
// ...

上面例子中,Observable 对象每隔 1 秒会返回一个值给调用方。即使调用方不再需要数据,仍旧会继续通过回调函数向调用推送数据。

RxJS 提供了中止 Observable 执行的机制:

var foo = Rx.Observable.create((observer) => {
  console.log("start")
  let i = 0
  let timer = setInterval(() => observer.next(i++), 1000)
  return () => {
    clearInterval(timer)
    console.log("end")
  }
})

var subscription = foo.subscribe(i => console.log(i))
setTimeout(() => subscription.unsubscribe(), 2500)
// 输出:
// start
// 0
// 1
// 2
// end

subscribe() 方法返回一个订阅对象(subscription),该对象上的 unsubscribe() 方法用于取消订阅,也就是中止 Observable 内部逻辑的执行,停止返回新的数据。

对于具体的 Observable 对象是如何中止执行,则要由 Observable 在执行后返回一个用于中止执行的函数,像上面例子中的这种方式。

Observable 执行结束后,会触发观察者的 complete 回调,所以可以这样:

foo.subscribe({
  next(value) { console.log(value) },
  complete() { console.log("completed") }
})

Observable 的观察者共有上面三种回调:

next:获得数据

error:处理异常

complete:执行结束

其中 next 可以被多次调用,error 和 complete 最多只有一个被调用一次(任意一个被调用后不再触发其他回调)。

数据转换

对于函数返回值,有时候我们要转换后再使用,例如:

function foo() {
  return 1
}

console.log(f00() * 2)
// 输出:
// 2

对于 Observable 返回的值,也会有类似的情况,不过通常采用下面的方式:

var foo = Rx.Observable.create((observer) => {
  let i = 0
  setInterval(() => observer.next(i++), 1000)
})

foo.map(i => i * 2).subscribe(i => console.log(i))
// 输出:
// 0
// 2
// 4
// ...

其实 foo.map() 返回了新的 Observable 对象,上面代码等价于:

var foo2 = foo.map(i => i * 2)
foo2.subscribe(i => console.log(i))

Observable 对象 foo2 被订阅时执行的内部逻辑可以简单视为:

function subscribe(observer) {
  let mapFn = v => v * 2
  foo.subscribe(v => {
    observer.next(mapFn(v))
  })
}

将这种对数据的处理和数组进行比较看看:

var array = [0, 1, 2, 3, 4, 5]
array.map(i => i * 2).forEach(i => console.log(i))

是不是有点像?

除了 map() 方法,Observable 还提供了多种转换方法,如 filter() 用于过滤数据,find() 值返回第一个满足条件的数据,reduce() 对数据进行累积处理,在执行结束后返回最终的数据。这些方法和数组方法功能是类似的,只不过是对异步返回的数据进行处理。还有一些转换方法更加强大,例如可以 debounceTime() 可以在时间维度上对数据进行拦截等等。

Observable 的转换方法,本质不过是创建了一个新的 Observable,新的 Observable 基于一定的逻辑对原 Observable 的返回值进行转换处理,然后再推送给观察者。

总结

Observable 就是一个奇怪的函数,它有和函数类似的东西,例如封装了一段逻辑,每次调用时都会重新执行逻辑,执行有返回数据等;也有更特殊的特性,例如数据是推送(push)的方式返回给调用方法,返回值可以是异步,可以返回多个值等。

不过将 Observable 视作特殊函数,至少对于理解 Observable 上是比较有帮助的。

Observable 也被视为 data stream(数据流),这是从 Observable 可以返回多个值的角度来看的,而数据转换则是基于当前数据流创建新的数据流,例如:

不过上图看到的只是数据,而将 Observable 视为特殊函数时,不应该忘了其内部逻辑,不然数据是怎么产生的呢。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/88703.html

相关文章

  • RxJS笔记

    摘要:深入浅出读书笔记遗留问题的与对应的实际场景,以及在编码中的体现部分测试你对时间的感觉按住我一秒钟然后松手你的时间毫秒实现重置避免直接触发事件,例如在处点击然后在处实现获取间隔时间你超过了的用户的使用主要用来加载静态资源,所 RxJS 《深入浅出RxJS》读书笔记 遗留问题 Observable的HOT与COLD对应的实际场景,以及在编码中的体现 chapter1 html部分 测...

    shinezejian 评论0 收藏0
  • RxJS基础教程

    摘要:是一个基于可观测数据流在异步编程应用中的库。正如官网所说,是基于观察者模式,迭代器模式和函数式编程。它具有时间与事件响应的概念。通知不再发送任何值。和通知可能只会在执行期间发生一次,并且只会执行其中的一个。 RxJS是一个基于可观测数据流在异步编程应用中的库。 ReactiveX is a combination of the best ideas fromthe Observer p...

    defcon 评论0 收藏0
  • RxJS 6有哪些新变化?

    摘要:有哪些新变化于年月日正式发布,为开发人员带来了一些令人兴奋的增补和改进。不要移除包,直到你将所有的链式操作修改为管道操作符。 RxJS 6有哪些新变化? RxJs 6于2018年4月24日正式发布,为开发人员带来了一些令人兴奋的增补和改进。Ben Lesh, rxJS核心开发成员,强调: RxJS 6在拥有更小API的同时,带来了更整洁的引入方式 提供一个npm包,该package可...

    LeanCloud 评论0 收藏0
  • 使用RxJS管理React应用状态实践分享

    摘要:随着前端应用的复杂度越来越高,如何管理应用的数据已经是一个不可回避的问题。应用的数据不是只有状态的,还有事件异步常量等等。出于以上两点原因,最终决定基于来设计一套管理应用的状态的解决方案。 随着前端应用的复杂度越来越高,如何管理应用的数据已经是一个不可回避的问题。当你面对的是业务场景复杂、需求变动频繁、各种应用数据互相关联依赖的大型前端应用时,你会如何去管理应用的状态数据呢? 我们认为...

    KunMinX 评论0 收藏0
  • 从观察者模式到迭代器模式系统讲解 RxJS Observable(一)

    摘要:是的缩写,起源于,是一个基于可观测数据流结合观察者模式和迭代器模式的一种异步编程的应用库。是基于观察者模式和迭代器模式以函数式编程思维来实现的。学习之前我们需要先了解观察者模式和迭代器模式,还要对流的概念有所认识。 RxJS 是 Reactive Extensions for JavaScript 的缩写,起源于 Reactive Extensions,是一个基于可观测数据流 Stre...

    notebin 评论0 收藏0

发表评论

0条评论

omgdog

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<