1. 安装
1. 安装1. 安装
# 安装 typescript， rxjs 包
npm install -D typescript @types/node
npm install rxjs

# 安装 typescript， rxjs 包
npm install -D typescript @types/node
npm install rxjs
2. 使用2. 使用2. 使用2.1 使用 from 来从数组生成源RxJS 有许多创建源的方法，如 from, fromEvent..., 这里使用 from做个例子
import {from} from 'rxjs'

// 从数组生成可订阅对象
// obser 的对象类型为 Observable
let obser = from([1,2,3,4,5])
// 消费对象
// next 当管道中有值流动，就会出发next
// 当发生错误是会触发 error
// 当循环完数组时，会调用complete
obse.subscribe({

next(item) {

console.log(item)

},

error(err) {

console.log(err)

},

complete() {

console.log("Done")

}
}
import {from} from 'rxjs'

// 从数组生成可订阅对象
// obser 的对象类型为 Observable
let obser = from([1,2,3,4,5])
// 消费对象
// next 当管道中有值流动，就会出发next
// 当发生错误是会触发 error
// 当循环完数组时，会调用complete
obse.subscribe({

next(item) {

console.log(item)

},

error(err) {

console.log(err)

},

complete() {

console.log("Done")

}
}2.2 自定义生成源

import {Observable, from} from 'rxjs'

// 新建一个可订阅对象
let obser = new Observable(productor => {

// 自定义数据流

productor.next("hello")

productor.next("world")

setTimeout(()=>{

productor.next("After 1 Sec")

productor.complete()

}, 1000)
})

// 消费
obse.subscribe({

next(item) {

console.log(item)

},

error(err) {

console.log(err)

},

complete() {

console.log("Done")

}
}


import {Observable, from} from 'rxjs'

// 新建一个可订阅对象
let obser = new Observable(productor => {

// 自定义数据流

productor.next("hello")

productor.next("world")

setTimeout(()=>{

productor.next("After 1 Sec")

productor.complete()

}, 1000)
})

// 消费
obse.subscribe({

next(item) {

console.log(item)

},

error(err) {

console.log(err)

},

complete() {

console.log("Done")

}
}

3. 实际例子，使用 Redis3. 实际例子，使用 Redis3. 实际例子，使用 Redis3.1 安装 Redis

npm install redis @types/redis

npm install redis @types/redis
3.2 使用

import {RedisClient} from "redis"
import {Observable} from "rxjs"
// 连接redis, redisclient 的构造类型是”ClientOpt接口“，只要是符合其字段定义的对象就可以
let reids = new RedisClient({

host:"localhost",

port:6379
})
// redis的操作都是异步操作，通过回调使用，很容易陷入毁掉地狱
redis.set("name", "tom", (err, res)=>{

// 需要在set成功后同步的操作

if(!err) {

redis.get("name", (err,res)=> {

console.log(res)

})

}
})

// 可以使用 Promise 封装上面的方法，但我们这里使用 rxjs的事件流
let redisObser = new Observable(productor => {
// 这个列子没啥实际用途，只做演示

redis.set("name", "jack", (err, res) => {

productor.next(res)

productor.complete()

})
})
// 消费代码
obse.subscribe({

next(key) {

// 读取或其他需保证在 set 后的操作

redis.get(key, (err, res) => {

console.log(res)

})

},

error(err) {

console.log(err)

},

complete() {

console.log("Query Finish")

}
}


import {RedisClient} from "redis"
import {Observable} from "rxjs"
// 连接redis, redisclient 的构造类型是”ClientOpt接口“，只要是符合其字段定义的对象就可以
let reids = new RedisClient({

host:"localhost",

port:6379
})
// redis的操作都是异步操作，通过回调使用，很容易陷入毁掉地狱
redis.set("name", "tom", (err, res)=>{

// 需要在set成功后同步的操作

if(!err) {

redis.get("name", (err,res)=> {

console.log(res)

})

}
})

// 可以使用 Promise 封装上面的方法，但我们这里使用 rxjs的事件流
let redisObser = new Observable(productor => {
// 这个列子没啥实际用途，只做演示

redis.set("name", "jack", (err, res) => {

productor.next(res)

productor.complete()

})
})
// 消费代码
obse.subscribe({

next(key) {

// 读取或其他需保证在 set 后的操作

redis.get(key, (err, res) => {

console.log(res)

})

},

error(err) {

console.log(err)

},

complete() {

console.log("Query Finish")

}
}