【angular教程240111】08异步数据流编程与angular ：promise，Rxjs6.x
异步与 Rxjs6.x异步数据流编程-Angular Rxjs快速入门
一、 Rxjs介绍
二、 Promise和RxJS 处理异步对比
三、 Rxjs unsubscribe取消订阅
四、 Rxjs 订阅后多次执行
五、 Angualr6.x之前使用Rxjs的工具函数map filter
六、 Angualr6.x 以后 Rxjs6.x的变化以及使用
七、 Rxjs延迟执行
八、 Rxjs把Dom事件转化成流
异步编程是一种编程范式，它允许程序在等待某个长时间运行的任务（如访问网络、文件系统操作、计时器等）完成时继续执行其他任务，而不是停下来等待任务完成。在JavaScript和Angular中，异步编程特别重要，因为它帮助避免阻塞UI渲染和用户交互，从而提高应用的响应性和性能。
这是最基础的异步编程技术。通过将一个函数（回调函数）作为参数传递给另一个函数，在特定的操作完成时执行。
function doAsyncTask(cb) {
setTimeout(() => {
console.log('Async Task Calling Callback');
cb();
}, 1000);
}
doAsyncTask(() => console.log('Callback Called'));
这种模式允许订阅一个事件，并在事件发生时执行一些操作。在Angular中，可以使用EventEmitter和@Output装饰器来实现组件间（如上一章节的父子组件间）的事件通信。
import { Component, EventEmitter, Output } from '@angular/core';
@Component({
selector: 'app-event-emitter',
template: '<button (click)="onClick()">Click me!</button>',
})
export class EventEmitterComponent {
@Output() clicked = new EventEmitter<void>();
onClick() {
this.clicked.emit();
}
}
Promise是ES6中引入的，用于表示一个异步操作的最终完成（或失败）及其结果值。
function doAsyncTask() {
return new Promise((resolve, reject) => {
setTimeout(() => {
console.log('Async Task Complete');
resolve();
}, 1000);
});
}
doAsyncTask().then(() => console.log('Task Done!'));
RxJS是Angular推荐的异步编程模型，它提供了一个强大的Observable类型。这是一个能够发射多个值的流，并且可以用各种强大的操作符来处理这些值。
import { of } from 'rxjs';
import { delay } from 'rxjs/operators';
of('Hello World').pipe(delay(1000)).subscribe(console.log);
在Angular中，RxJS是处理事件和异步操作的首选方式，因为它提供了更高的灵活性、可组合性和强大的操作符集合，更适用于复杂的应用程序。它被广泛应用在Angular的HTTP服务、表单事件和状态管理等场景中。
RxJS是Reactive Extensions For JavaScript的缩写，它提供了一种模型来处理时间序列上的事件。RxJS使用了Observables和一系列操作符来处理异步事件。
Angular的HttpClient服务返回的Observable对象是处理HTTP异步请求的例子。这些Observable可以被订阅，以在数据到达时更新UI。
import { HttpClient } from '@angular/common/http';
import { Observable } from 'rxjs';
export class MyService {
constructor(private http: HttpClient) {}
getData(): Observable<MyDataType> {
return this.http.get<MyDataType>('/api/data');
}
}
在组件中使用服务：
export class MyComponent implements OnInit {
constructor(private myService: MyService) {}
ngOnInit() {
this.myService.getData().subscribe(data => {
console.log(data);
// 更新UI
});
}
}
这里，getData方法发起HTTP请求并返回Observable。组件订阅这个Observable，并在数据到达时响应。由于使用了HttpClient，Angular会自动处理变更检测，所以不需要手动触发。
Promise: 一次性的异步操作，无法取消，不支持重试。
RxJS: 可以被取消，支持重试、过滤、转换等，可以表示多个值。
“语法糖”（Syntactic Sugar）是一个编程术语，指的是在编程语言中添加的某种语法，这种语法对语言的功能并没有影响，但是可以使代码更易读、更易写。换句话说，语法糖就是让人们编写代码时更加方便的语法。
这种语法让开发者用更简洁、更清晰的代码来实现相同的功能，通常隐藏了一些实现的复杂性。语法糖让代码更接近自然语言，减少了编程的冗余，使得代码更加简洁。
例如，在JavaScript中，async/await 是一个常见的语法糖。它们让用同步的方式写出异步的代码，其背后实际上是基于Promise的。
不使用语法糖的代码示例（使用Promise）:
function fetchData() {
return getData() // 返回一个Promise
.then(data => {
return 'Done with data: ' + data;
});
}
使用async/await语法糖的代码示例:
async function fetchData() {
const data = await getData(); // 等待Promise解决
return 'Done with data: ' + data;
}
在这个例子中，async/await 使得异步代码看起来和同步代码非常相似，从而使得代码更容易理解和维护。尽管async/await 看起来像是同步执行，但它们实际上还是异步的，它们只是让异步代码的书写和阅读变得更容易。
取消订阅是RxJS中管理Observable订阅的过程，用来防止内存泄漏。
import { Subscription } from 'rxjs';
const subscription: Subscription = observable.subscribe(
value => console.log(value),
error => console.error(error),
() => console.log('Completed!')
);
// 取消订阅
subscription.unsubscribe();
Observables可以发送多个值，并且可以订阅这些值。
import { Observable } from 'rxjs';
const observable = new Observable(subscriber => {
subscriber.next(1);
subscriber.next(2);
setTimeout(() => {
subscriber.next(3);
subscriber.complete();
}, 1000);
});
observable.subscribe(value => console.log(value));
import { Observable } from 'rxjs';
import { map, filter } from 'rxjs/operators';
const observable = new Observable((subscriber) => {
subscriber.next(1);
subscriber.next(2);
subscriber.next(3);
}).pipe(
filter((value) => value > 1),
map(value => value * value)
);
observable.subscribe(value => console.log(value));
RxJS 6.x 有很大的变化，比如引入了管道（pipeable）操作符。
import { of } from 'rxjs';
import { map } from 'rxjs/operators';
const source$ = of(1, 2, 3, 4, 5);
const result$ = source$.pipe(map(val => val * 10));
result$.subscribe(console.log);
使用delay操作符可以延迟Observable发出的值。
import { of } from 'rxjs';
import { delay } from 'rxjs/operators';
of('Hello World!').pipe(delay(1000)).subscribe(console.log);
RxJS可以把DOM事件转化为Observable流。
import { fromEvent } from 'rxjs';
const button = document.querySelector('button');
const myObservable = fromEvent(button, 'click');
myObservable.subscribe(event => console.log(event));
以上是各个主题的简要介绍和代码示例。在实际的Angular应用程序中，会把这些代码片段整合到服务或组件中去管理数据流和事件。
异步‘
假设你想拿一个数据但是这个数据给到你需要时间，怎么让程序不会直接在还没拿到时候就报错或者跳过去
服务 request.service.ts
import { Injectable } from '@angular/core';
import { timeout } from 'rxjs';
@Injectable({
providedIn: 'root',
})
export class RequestService {
constructor() {}
// 执行顺序1 同步方法
getData() {
return '这是一个服务数据';
}
// 2 模拟异步方法
// 1
/* getCallbackData() {
// 执行顺序2
setTimeout(() => {
// 执行顺序4
var data = '张三';
return data;
}, 1000);
// 执行顺序3
//会报错，因为 异步数据无法在外部直接获取 因此要写回调函数
} */
getCallbackData(cb: any) {
/* ---------------cb回调函数
home组件中 // (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数
传给cd 然后 在本服务中 调用
*/
setTimeout(() => {
var username = '张三';
// return username;
/* cd 即 function (data)=>{
console.log(data);
}
*/
cb(username);
}, 1000);
}
}
子组件 home上使用服务
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
ngOnInit() {
//1同步方法
let data = this.myRequest.getData();
console.log(data);
// 2异步方法----回调函数
/*
let calllbackData = this.myRequest.getCallbackData();
console.log(calllbackData);
//会报错undedined，因为 异步数据无法在外部直接获取 因此要写回调函数 (data:any)=>{} 在回调函数的方法体内
*/
let calllbackData = this.myRequest.getCallbackData((data: any) => {
console.log(data);
});
// (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数 传给cd 然后 在服务中 调用
}
}
服务 request.service.ts
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
ngOnInit() {
//1同步方法
let data = this.myRequest.getData();
console.log(data);
// 2异步方法----回调函数
/*
let calllbackData = this.myRequest.getCallbackData();
console.log(calllbackData);
//会报错undedined，因为 异步数据无法在外部直接获取 因此要写回调函数 (data:any)=>{} 在回调函数的方法体内
*/
let calllbackData = this.myRequest.getCallbackData((data: any) => {
console.log(data);
});
// (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数 传给cd 然后 在服务中 调用
//3 promise 获取异步数据
var promiseData: any = this.myRequest.getPromiseData();
promiseData.then((data: any) => {
console.log(data);
});
}
}
子组件 home
import { Injectable } from '@angular/core';
import { timeout } from 'rxjs';
@Injectable({
providedIn: 'root',
})
export class RequestService {
constructor() {}
// 执行顺序1 同步方法
getData() {
return '这是一个服务数据';
}
// 2 模拟异步方法
// 1
/* getCallbackData() {
// 执行顺序2
setTimeout(() => {
// 执行顺序4
var data = '张三';
return data;
}, 1000);
// 执行顺序3
//会报错，因为 异步数据无法在外部直接获取 因此要写回调函数
} */
getCallbackData(cb: any) {
/* ---------------cb回调函数
home组件中 // (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数
传给cd 然后 在本服务中 调用
*/
setTimeout(() => {
var username = '张三';
// return username;
cb(username);
}, 1000);
}
getPromiseData() {
/* ---------------不用cb回调函数
home组件中 // (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数
传给cd 然后 在本服务中 调用
*/
// promise()方法 并在里面传入一个function () =>{ } 方法 ,() =>{ }方法里面有两个参数 resolve 和 rejecte 表示 成功和失败后的回调函数
return new Promise((resolve, rejecte) => {
setTimeout(() => {
var username = '张三---promiise';
//1同步return username;
//2 回调函数 cb(username);
//3
resolve(username);
}, 1000);
});
}
}
function resolve(username: string) {
throw new Error('Function not implemented.');
}
服务 request.service.ts
import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';
@Injectable({
providedIn: 'root',
})
export class RequestService {
constructor() {}
// 执行顺序1 同步方法
getData() {
return '这是一个服务数据';
}
// 2 模拟异步方法
// 1
/* getCallbackData() {
// 执行顺序2
setTimeout(() => {
// 执行顺序4
var data = '张三';
return data;
}, 1000);
// 执行顺序3
//会报错，因为 异步数据无法在外部直接获取 因此要写回调函数
} */
getCallbackData(cb: any) {
/* ---------------cb回调函数
home组件中 // (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数
传给cd 然后 在本服务中 调用
*/
setTimeout(() => {
var username = '张三';
// return username;
cb(username);
}, 1000);
}
getPromiseData() {
/* ---------------不用cb回调函数
home组件中 // (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数
传给cd 然后 在本服务中 调用
*/
// promise()方法 并在里面传入一个function () =>{ } 方法 ,() =>{ }方法里面有两个参数 resolve 和 rejecte 表示 成功和失败后的回调函数
return new Promise((resolve, rejecte) => {
setTimeout(() => {
var username = '张三---promiise';
//1同步return username;
//2 回调函数 cb(username);
//3 传入待处理的数据 username
resolve(username);
}, 3000);
});
}
// 4
getRxjsData() {
/* 通过 Observable，来return返回一个处理的对象 ，
在 Observable的( )内传入一个异步的方法 ( )=> ,
异步的方法 ( )=> 的( )内，来传参数，一般可以传入任意的参数，但此参数 一般叫observer ，
处理异步的数据
*/ return new Observable((observer) => {
// setTimeout 一个定时器
setTimeout(() => {
var username = '张三---rxjs';
// resolve(username);
observer.next(username);
// 观测到数据后，传入数据，待返回使用 这就封装好了一个异步的rxjs方法
// 然后再要使用的地方 subscribe订阅 就
}, 3000);
});
}
}
子组件home
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
ngOnInit() {
//1同步方法
let data = this.myRequest.getData();
console.log(data);
// 2异步方法----回调函数
/*
let calllbackData = this.myRequest.getCallbackData();
console.log(calllbackData);
//会报错undedined，因为 异步数据无法在外部直接获取 因此要写回调函数 (data:any)=>{} 在回调函数的方法体内
*/
let calllbackData = this.myRequest.getCallbackData((data: any) => {
console.log(data);
});
// (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数 传给cd 然后 在服务中 调用
//3 promise 获取异步数据
var promiseData: any = this.myRequest.getPromiseData();
promiseData.then((data: any) => {
console.log(data);
});
// 4 rxjs获取异步方法里面的返回的数据
/* 要在外部获取 使用 封装好的 getRxjsData异步方法里面返回的数据，即 observable方法观察到的observer这个观察对象
获取异步方法里面的返回的数据 ，返回的observable的这个对象的rxjsData的subscribe
的使用一个变量rajxData来接收
然后使用 订阅 subscribe 里面的rxjs的来打印出来此数据 */
/* promise 与 rxjs对比
1 返回数据 ---promise 方法 resolve（username） rxjs里用 observe.next（username）
2 获取异步方法里面的数据 ---promise 方法 promiseData.then rxjs用 rsjxData.subscribe（data）
3 rxjs更强大
*/
var rxjsData = this.myRequest.getRxjsData();
rxjsData.subscribe((data) => {
console.log(data);
});
}
}
上述3 服务 request.service.ts中，异步方法rxjs的数据是设置的 3秒钟之后获得，
子组件home 在使用此rxjs时，可以设置定时器，通过判断数据获取的时间，来订阅、取消订阅。
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
ngOnInit() {
//1同步方法
let data = this.myRequest.getData();
console.log(data);
// 2异步方法----回调函数
/*
let calllbackData = this.myRequest.getCallbackData();
console.log(calllbackData);
//会报错undedined，因为 异步数据无法在外部直接获取 因此要写回调函数 (data:any)=>{} 在回调函数的方法体内
*/
let calllbackData = this.myRequest.getCallbackData((data: any) => {
console.log(data);
});
// (data: any) => { console.log(calllbackData) }整个回调函数 都做为 参数 传给cd 然后 在服务中 调用
//3 promise 获取异步数据
var promiseData: any = this.myRequest.getPromiseData();
promiseData.then((data: any) => {
console.log(data);
});
// 4 rxjs获取异步方法里面的返回的数据
/* 要在外部获取 使用 封装好的 getRxjsData异步方法里面返回的数据，即 observable方法观察到的observer这个观察对象
获取异步方法里面的返回的数据 ，返回的observable的这个对象的rxjsData的subscribe
的使用一个变量rajxData来接收
然后使用 订阅 subscribe 里面的rxjs的来打印出来此数据 */
/* promise 与 rxjs对比
1 返回数据 ---promise 方法 resolve（username） rxjs里用 observe.next（username）
2 获取异步方法里面的数据 ---promise 方法 promiseData.then rxjs用 rsjxData.subscribe（data）
3 rxjs更强大
*/
/* 5 功能 测试，因此注释
// -var rxjsData = this.myRequest.getRxjsData();
- // rxjsData.subscribe((data) => {
- // console.log(data);
- // });
代码
*/
// var rxjsData = this.myRequest.getRxjsData();
// rxjsData.subscribe((data) => {
// console.log(data);
// });
// 5 外部调用时，设置中途撤回，如一秒钟未获得数据，就撤回刚才的操作
var streem = this.myRequest.getRxjsData();
// 把rxjs的obserable返回的对象 赋值给 stream 异步数据流
var d = streem.subscribe((data) => {
// streem.subscribe(方法 来获取异步数据流里面的数据 (data) =>
console.log(data);
// 将数据 (data)进行处理 => { console.log(data); }); 此处是一个 console.log 控制台你打印 处理
});
//然后将返回的东西 赋值给一个变量 d ，接下来通过一个定时器 timeout来设置超时取消
setTimeout(() => {
d.unsubscribe();
// 1 取消订阅
/*
注意语法错误：取消订阅的部分误写为 d.unsubscribe;这实际上并没有调用 unsubscribe 函数。
正确的方式是 d.unsubscribe();。
缺少括号意味着函数没有被调用，因此订阅没有被取消。
*/
}, 1000);
// 一秒钟未获得数据，就撤回刚才的操作
/*
// 2 设置 第四秒 意味着 订阅数据
// }, 4000); // 将时间设置为大于Observable发送数据的时间
*/
}
}
request.service.ts服务内封装方法
// 5 多次执行
getRxjsIntervalData() {
let count = 0;
return new Observable<any>((observer) => {
setInterval(() => {
count++;
var username = '张三--Rxjs-Interval' + count;
observer.next(username);
// observer.error('数据')
}, 1000);
});
}
子组件 home使用
// 这是一个同步方法
ngOnInit(): void {
// 6 多次执行
// 这是一个同步方法
//7、rxjs执行多次
var streemInterval = this.myRequest.getRxjsIntervalData();
streemInterval.subscribe((data: any) => {
console.log(data);
});**
}
假设有一个观察对象（Observable），它发出一系列数字，想对这些数字进行处理。
使用 map: 假设想将每个数字乘以 2。
import { of } from 'rxjs';
import { map } from 'rxjs/operators';
// 创建一个发出数字 1, 2, 3 的 Observable
const numbers$ = of(1, 2, 3);
// 使用 map 操作符将每个数字乘以 2
const doubledNumbers$ = numbers$.pipe(
map(number => number * 2)
);
// 订阅并打印结果
doubledNumbers$.subscribe(console.log); // 输出将是 2, 4, 6
这里，map(number => number * 2) 会将每个发出的值乘以 2。
使用 filter: 假设只对大于 1 的数字感兴趣。
import { of } from 'rxjs';
import { filter } from 'rxjs/operators';
// 创建一个发出数字 1, 2, 3 的 Observable
const numbers$ = of(1, 2, 3);
// 使用 filter 操作符选择大于 1 的数字
const filteredNumbers$ = numbers$.pipe(
filter(number => number > 1)
);
// 订阅并打印结果
filteredNumbers$.subscribe(console.log); // 输出将是 2, 3
在这个例子中，filter(number => number > 1) 只会通过那些大于 1 的值。
这些例子展示了如何使用 map 和 filter 操作符来转换和筛选 Observable 发出的值。通过这种方式，可以很方便地处理和操作异步数据流。
服务
request.service.ts
import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';
@Injectable({
providedIn: 'root',
})
export class RequestService {
constructor() {}
getRxjsIntervalNum() {
let count = 0;
return new Observable<any>((observer) => {
setInterval(() => {
count++;
observer.next(count);
// observer.error('数据')
}, 1000);
});
}
}
子组件ts
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
import { map, filter } from 'rxjs/operators';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
// 这是一个同步方法
ngOnInit(): void {
var streemNum = this.myRequest.getRxjsIntervalNum();
streemNum
.pipe(
filter((value: any): any => {
// filter(value:any)要指定类型，否则会报错
if (value % 2 == 0) {
return true;
}
})
)
.subscribe((data: any) => {
console.log(data);
});
}
}
控制台运行结果
[webpack-dev-server] Server started: Hot Module Replacement disabled, Live Reloading enabled, Progress disabled, Overlay enabled.
core.mjs:26656 Angular is running in development mode.
home.component.ts:25 2
home.component.ts:25 4
home.component.ts:25 6
home.component.ts:25 8
home.component.ts:25 10
home.component.ts:25 12
home.component.ts:25 14
home.component.ts:25 16
…
子组件home
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
import { map, filter } from 'rxjs/operators';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
// 这是一个同步方法
ngOnInit(): void {
// map
var streemNum = this.myRequest.getRxjsIntervalNum();
streemNum
.pipe(
map((value) => {
return value * value;
})
)
.subscribe((data) => {
console.log(data);
});
/* // filter
var streemNum = this.myRequest.getRxjsIntervalNum();
streemNum
.pipe(
filter((value: any): any => {
// filter(value:any)要指定类型，否则会报错
if (value % 2 == 0) {
return true;
}
})
)
.subscribe((data: any) => {
console.log(data);
});
// */
}
}
控制台运行结果
[webpack-dev-server] Server started: Hot Module Replacement disabled, Live Reloading enabled, Progress disabled, Overlay enabled.
core.mjs:26656 Angular is running in development mode.
home.component.ts:25 1
home.component.ts:25 4
home.component.ts:25 9
home.component.ts:25 16
home.component.ts:25 25
home.component.ts:25 36
。。。
服务不变
子组件 filter & map结合使用
import { Component, OnInit } from '@angular/core';
import { RequestService } from 'src/app/services/request.service';
import { map, filter } from 'rxjs/operators';
@Component({
selector: 'app-home',
templateUrl: './home.component.html',
styleUrls: ['./home.component.scss'],
})
export class HomeComponent implements OnInit {
constructor(public myRequest: RequestService) {}
// 这是一个同步方法
// 这是一个同步方法
ngOnInit(): void {
var streemNum = this.myRequest.getRxjsIntervalNum();
streemNum
.pipe(
filter((value:any):any => {
if (value % 2 == 0) {
return true;
}
}),
map((value) => {
return value * value;
})
)
.subscribe((data) => {
console.log(data);
});
}
}
控制台运行结果
[webpack-dev-server] Server started: Hot Module Replacement disabled, Live Reloading enabled, Progress disabled, Overlay enabled.
core.mjs:26656 Angular is running in development mode.
home.component.ts:28 4
home.component.ts:28 16
home.component.ts:28 36
home.component.ts:28 64
home.component.ts:28 100
home.component.ts:28 144
home.component.ts:28 196