前端Rollup+RxJs响应式编程实践

响应式编程

以下概念介绍，基于 node v10.16.3 版本进行。不满足条件的，请移步 google or 度娘。

概念介绍

const a = 1,b = 2;
const c = a+ b;

响应式编程，当a,b再次变化时，c会再次做出修改。

背景

• RxJS是一个用于使用Observable进行响应式编程的库
• 可简化编写异步或基于回调的代码

何为Observable?

• js 函数定义

  functioin hello(){
      return 'hello'
  }
  hello(); //此处才能获取到'hello'

• js 生成器函数

 function* getValue(){
     yield 1;
     yield 2;
 }
 const ite = getValue();
 ite.next();//{value: 1, done: false}
 ite.next();//{value: 2, done: false}

• js promise

 const prom = new Promise((resolve,reject) => {
   if(isok){
       resolve('ok');
   }else{
       reject(new Error('error'));
   }
 })
prom.then(res => {
   console.log('res....',res);
}).catch(err => {
    console.log("error ....");
})

• js observable

 const data$ = new Observable((observer) => {
      observer.next(1);
      observer.next(2);
      observer.next(3);
      setTimeout(() => {
        observer.next(4);
        observer.complete();
      }, 1000);
 });
console.log('just before subscribe');
data$.subscribe({
  next(x) { console.log('got value ' + x); },
  error(err) { console.error('something wrong occurred: ' + err); },
  complete() { console.log('done'); }
});
console.log('just after subscribe');
//输出如下：
just before subscribe
got value 1
got value 2
got value 3
just after subscribe
//异步返回
got value 4
done

• 从上，我们可以看出
  • js function、js 生成器函数是一类，调用的地方决定何时使用
  • promise、observable一类，定义的地方产生数据，使用的地方不知道何时产生数据
  • observabel类似promise，但可同步返回，也可以异步返回，还可以return多个值。

本地环境搭建

• node
• npm script
• rollup

npm install --global rollup

• rxjs

npm install rxjs

• es6 语法，利用tree-shaking ，缩减代码

使用

hello-world 之123

import { of } from 'rxjs';
import { map } from 'rxjs/operators';
of(1,2,3).subscribe(res => {
    console.log(res);
});

rollup打包→ bundle.js → node bundle.js，连续输出123

rollup main.js --file bundle.js --format cjs

'use strict';

var rxjs = require('rxjs');
require('rxjs/operators');

rxjs.of(1,2,3).subscribe(res => {
    console.log(res);
});

优化rollup配置

创建rollup.config.js

export default {
  input: 'main.js',
  output: {
    file: 'bundle.js',
    format: 'cjs'
  }
};

rollup -c rollup.config.js 

npm script支持

package.json

 {
 "name": "rxjs-demo",
  "version": "1.0.0",
  "description": "",
  "main": "test.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1",
    "build":"rollup -c", // 默认配置 rollup --config  rollup.config.js
    "start":"npm run build && node bundle.js"   // && 是继发，& 是并发
  },
  "keywords": [],
  "author": "",
  "license": "ISC",
  "devDependencies": {
    "rxjs": "^6.5.3"
  }
}

最终的运行如下：

npm start 

Rxjs读取package.json字段version

// install plugin
npm install --save-dev rollup-plugin-json

// rollup.config.js
import json from 'rollup-plugin-json';
export default {
	input: 'main.js',
	output: {
		file: 'bundle.js',
		format: 'cjs'
	},
	plugins:[json()]
};

// main.js
import { of } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

of(1,2,3,version).subscribe(res => {
    console.log(res);
});

pipe管道在输出前修改数据

import { of } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

of(1,2,3,version).pipe(map(item => item + ' through pipe use map'))
.subscribe(res => {
    console.log(res);
});

后续的内容，就不在介绍rollup的配置，直接介绍rxjs相关代码实践

Observable代码实践

subscribe订阅，得到数据

console.log("before observable....");
const publisher$ = new  Observable(suber => {
    console.log("in observable....");
    suber.next(1);
    suber.next(2);
    suber.next(3);
    suber.next(version);
});
console.log("after observable....");
publisher$.pipe(map(item => item + ' through pipe use map')).subscribe(res => {
    console.log(res);
});
publisher$.subscribe(res => {
    console.log("subscribe second..." + res);
});

observable状态终止

import { of,Observable } from 'rxjs';
import { map } from 'rxjs/operators';

const publiser$ = new Observable(suber => {
    suber.next(1);
    suber.next(2);
    suber.next(3);
    // suber.complete();
    suber.error(); //error,complete两者择一，一旦发生，后面的不会再返回
    suber.next(5);
});
publiser$.subscribe({
    next:res => {
        console.log(`next ...${res}`); 
    },
    complete:() => {
        console.log("complete done....");
    },
    error: () => {
        console.log("error occur....");
    }
});
//输出
next ...1
next ...2
next ...3
error occur....

next中异常，不会进入error

import { of,Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';


const publiser$ = new Observable(suber => {
    suber.next(1);
    suber.next(2);
    suber.next(3);
    suber.next(version);
    // suber.complete();
    suber.error();
});
publiser$.subscribe({
    next:res => {
        console.log(`next ...${res}`);  # 此处发生异常，会抛出，不会进入error.
    },
    complete:() => {
        console.log("complete done...."); 
    },
    error: () => {
        console.log("error occur....");# error的发生由于publiser$处，触发的。
    }
});

停止订阅，释放资源，但状态没有终止

import { of,Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

const publiser$ = new Observable(suber => {
    let count = 0;
    const timer = setInterval(() => {
       suber.next(count);
			 count++;
    },400);
    //suber.next(1);
    // suber.complete();
    // suber.error();
    return {
        unsubscribe:() => {
            console.log("stop push data....");
            clearInterval(timer);
        }
    }
});
const unsuber = publiser$.subscribe({
    next:res => {
        console.log(`next ...${res}`);
    },
    complete:() => {
        console.log("complete done....");
    },
    error: () => {
        console.log("error occur....");
    }
});

setTimeout(() => {
    unsuber.unsubscribe();
},2000);

hot subscribe - 与时间先后有关，订阅内容差

import { of,Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

let count = 0;
console.log("before observable....");
const publisher$ = new  Observable(suber => {
    console.log("in observable....");
    setInterval(() => {
        count++;
        suber.next(count);
    }, 100);
});
console.log("after observable....");
publisher$.pipe(map(item => 'first ...'+ item)).subscribe(res => {
    console.log(res);
});

setTimeout(() => {
    const publisher2$ = new Observable(suber => {
        setInterval(() => {
            suber.next(count);
        }, 100);
    });
    publisher2$.subscribe(res => {
        console.log("second ..." + res);
    }); 
}, 1000);

cold subscribe -- 时间先后无关，内容无差异

import { of,Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';


console.log("before observable....");
const publisher$ = new  Observable(suber => {
    console.log("in observable....");
    setInterval(() => {
        let count = 0;
        count++;
        suber.next(count);
    }, 100);
});
console.log("after observable....");
publisher$.pipe(map(item => 'first ...'+ item)).subscribe(res => {
    console.log(res);
});

setTimeout(() => {
    const publisher2$ = new Observable(suber => {
        setInterval(() => {
            let count = 0;
            count++;
            suber.next(count);
        }, 100);
    });
    publisher2$.subscribe(res => {
        console.log("second ..." + res);
    }); 
}, 1000);

Operator- rxjs 内置操作符

分类：

• Pipeable Operators
  • filter(...)
  • mergeMap(...)
  • map()
  • ...
• Creation Operators
  • of
  • fromEvent
  • ajax
  • bindCallback
  • bindNodeCallback
  • defer
  • empty
  • from
  • fromEvent
  • ...

Observable都要求subscribe返回unsubscribe，释放资源

import { of,Observable, interval } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

const publiser$ = interval(400);
const unsuber = publiser$.subscribe({
    next:res => {
        console.log(`next ...${res}`);
    },
    complete:() => {
        console.log("complete done....");
    },
    error: () => {
        console.log("error occur....");
    }
});

setTimeout(() => {
    unsuber.unsubscribe();
},2000);

更多请参考

subject

即可以next,又可以subscribe的特殊的Observable- SubJect

import { of,Observable, interval,Subject } from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

const sub = new Subject();
sub.pipe(map(res => 'first ...'+res)).subscribe((res) => {
    console.log("sub one...",res);
});
sub.subscribe({
    next:(res) => {
        console.log(`second...${res}`);
    }
});
sub.next(1);
sub.next(2);

import { of,Observable, interval,Subject ,from} from 'rxjs';
import { map } from 'rxjs/operators';
import { version } from './package.json';

const sub = new Subject();
sub.pipe(map(res => 'first ...'+res)).subscribe((res) => {
    console.log("sub one...",res);
});
sub.subscribe({
    next:(res) => {
        console.log(`second...${res}`);
    }
});
// sub.next(1);
// sub.next(2);
const publisher$ = from([1,2,3]);
publisher$.subscribe(sub);

内置BehaviorSubject - 返回最近的一次value

import { BehaviorSubject } from 'rxjs';
const subject = new BehaviorSubject(0); // 0 is the initial value
 
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
 
subject.next(1);
subject.next(2);
 
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
 
subject.next(3);
 
// Logs
// observerA: 0
// observerA: 1
// observerA: 2
// observerB: 2
// observerA: 3
// observerB: 3

内置ReplaySubject- 返回指定个数的最近值

import { ReplaySubject } from 'rxjs';
const subject = new ReplaySubject(3); // 返回最近3条 for new subscribers
 
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
 
subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);
 
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
 
subject.next(5);
 
// Logs:
// observerA: 1
// observerA: 2
// observerA: 3
// observerA: 4
// observerB: 2
// observerB: 3
// observerB: 4
// observerA: 5
// observerB: 5

内置AsyncSubject - 结束才返回

import { AsyncSubject } from 'rxjs';
const subject = new AsyncSubject();
 
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
 
subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);
 
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
 
subject.next(5);
subject.complete();
 
// Logs:
// observerA: 5
// observerB: 5

参考文献

RxJS