通过使用Http,我们调用一个方法来进行网络调用,并返回一个Http可观察对象:
getCustomer() {
return this.http.get('/someUrl').map(res => res.json());
}
如果我们获取这个可观察对象并向其添加多个订阅者:
let network$ = getCustomer();
let subscriber1 = network$.subscribe(...);
let subscriber2 = network$.subscribe(...);
我们要做的是确保这不会导致多个网络请求。
这似乎是一个不寻常的场景,但实际上很常见:例如,如果调用者订阅了可观察对象以显示错误消息,并使用异步管道将其传递给模板,那么我们已经有两个订阅者了。
在RxJs 5中正确的方法是什么?
也就是说,这似乎工作得很好:
getCustomer() {
return this.http.get('/someUrl').map(res => res.json()).share();
}
但是这是RxJs 5中惯用的方法吗,或者我们应该用别的方法来代替?
注意:根据Angular 5的新HttpClient,所有示例中的.map(res => res. JSON())部分现在都是无用的,因为现在默认假设JSON结果。
我把问题打了星号,但我会试着试一下。
//this will be the shared observable that
//anyone can subscribe to, get the value,
//but not cause an api request
let customer$ = new Rx.ReplaySubject(1);
getCustomer().subscribe(customer$);
//here's the first subscriber
customer$.subscribe(val => console.log('subscriber 1: ' + val));
//here's the second subscriber
setTimeout(() => {
customer$.subscribe(val => console.log('subscriber 2: ' + val));
}, 1000);
function getCustomer() {
return new Rx.Observable(observer => {
console.log('api request');
setTimeout(() => {
console.log('api response');
observer.next('customer object');
observer.complete();
}, 500);
});
}
这是证据:)
这里只有一个要点:getCustomer().subscribe(customer$)
我们不是订阅getCustomer()的api响应,我们是订阅一个ReplaySubject,它是可观察的,它也可以订阅一个不同的可观察对象,并且(这很重要)持有它最后发出的值并重新发布给它的任何(ReplaySubject的)订阅者。
更新:Ben Lesh说在5.2.0之后的下一个小版本中,你将能够调用shareplay()来真正地缓存。
以前……
首先,不要使用share()或publishReplay(1). refcount(),它们是相同的,它的问题是,它只在可观察对象处于活动状态时建立连接时共享,如果你在它完成连接后,它会再次创建一个新的可观察对象,翻译,而不是真正的缓存。
Birowski给出了正确的解决方案,即使用ReplaySubject。ReplaySubject将缓存你给它的值(bufferSize),在我们的例子中是1。它不会像share()一样在refCount为零时创建一个新的可观察对象,并且你建立了一个新的连接,这是缓存的正确行为。
这是一个可重用函数
export function cacheable<T>(o: Observable<T>): Observable<T> {
let replay = new ReplaySubject<T>(1);
o.subscribe(
x => replay.next(x),
x => replay.error(x),
() => replay.complete()
);
return replay.asObservable();
}
下面是如何使用它
import { Injectable } from '@angular/core';
import { Http } from '@angular/http';
import { Observable } from 'rxjs/Observable';
import { cacheable } from '../utils/rxjs-functions';
@Injectable()
export class SettingsService {
_cache: Observable<any>;
constructor(private _http: Http, ) { }
refresh = () => {
if (this._cache) {
return this._cache;
}
return this._cache = cacheable<any>(this._http.get('YOUR URL'));
}
}
下面是一个更高级版本的可缓存函数。这个函数允许有自己的查找表+提供自定义查找表的能力。这样的话,你就不用检查了。_cache就像上面的例子。还要注意的是,你传递的不是可观察对象作为第一个参数,而是一个返回可观察对象的函数,这是因为Angular的Http会立即执行,所以通过返回一个延迟执行的函数,如果它已经在缓存中,我们可以决定不调用它。
let cacheableCache: { [key: string]: Observable<any> } = {};
export function cacheable<T>(returnObservable: () => Observable<T>, key?: string, customCache?: { [key: string]: Observable<T> }): Observable<T> {
if (!!key && (customCache || cacheableCache)[key]) {
return (customCache || cacheableCache)[key] as Observable<T>;
}
let replay = new ReplaySubject<T>(1);
returnObservable().subscribe(
x => replay.next(x),
x => replay.error(x),
() => replay.complete()
);
let observable = replay.asObservable();
if (!!key) {
if (!!customCache) {
customCache[key] = observable;
} else {
cacheableCache[key] = observable;
}
}
return observable;
}
用法:
getData() => cacheable(this._http.get("YOUR URL"), "this is key for my cache")
这是.publishReplay (1) .refCount ();或.publishLast () .refCount ();因为Angular Http的可观察对象在请求后完成。
这个简单的类缓存结果,因此您可以多次订阅.value,并且只发出一个请求。你也可以使用.reload()来发出新的请求并发布数据。
你可以这样使用它:
let res = new RestResource(() => this.http.get('inline.bundleo.js'));
res.status.subscribe((loading)=>{
console.log('STATUS=',loading);
});
res.value.subscribe((value) => {
console.log('VALUE=', value);
});
来源是:
export class RestResource {
static readonly LOADING: string = 'RestResource_Loading';
static readonly ERROR: string = 'RestResource_Error';
static readonly IDLE: string = 'RestResource_Idle';
public value: Observable<any>;
public status: Observable<string>;
private loadStatus: Observer<any>;
private reloader: Observable<any>;
private reloadTrigger: Observer<any>;
constructor(requestObservableFn: () => Observable<any>) {
this.status = Observable.create((o) => {
this.loadStatus = o;
});
this.reloader = Observable.create((o: Observer<any>) => {
this.reloadTrigger = o;
});
this.value = this.reloader.startWith(null).switchMap(() => {
if (this.loadStatus) {
this.loadStatus.next(RestResource.LOADING);
}
return requestObservableFn()
.map((res) => {
if (this.loadStatus) {
this.loadStatus.next(RestResource.IDLE);
}
return res;
}).catch((err)=>{
if (this.loadStatus) {
this.loadStatus.next(RestResource.ERROR);
}
return Observable.of(null);
});
}).publishReplay(1).refCount();
}
reload() {
this.reloadTrigger.next(null);
}
}
只需在map之后和任何订阅之前调用share()。
在我的例子中,我有一个通用服务(RestClientService.ts),它进行其余调用,提取数据,检查错误,并将可观察对象返回到具体的实现服务(f.ex)。: ContractClientService.ts),最后这个具体的实现将可观察对象返回给de ContractComponent。Ts,这个订阅来更新视图。
RestClientService.ts:
export abstract class RestClientService<T extends BaseModel> {
public GetAll = (path: string, property: string): Observable<T[]> => {
let fullPath = this.actionUrl + path;
let observable = this._http.get(fullPath).map(res => this.extractData(res, property));
observable = observable.share(); //allows multiple subscribers without making again the http request
observable.subscribe(
(res) => {},
error => this.handleError2(error, "GetAll", fullPath),
() => {}
);
return observable;
}
private extractData(res: Response, property: string) {
...
}
private handleError2(error: any, method: string, path: string) {
...
}
}
ContractService.ts:
export class ContractService extends RestClientService<Contract> {
private GET_ALL_ITEMS_REST_URI_PATH = "search";
private GET_ALL_ITEMS_PROPERTY_PATH = "contract";
public getAllItems(): Observable<Contract[]> {
return this.GetAll(this.GET_ALL_ITEMS_REST_URI_PATH, this.GET_ALL_ITEMS_PROPERTY_PATH);
}
}
ContractComponent.ts:
export class ContractComponent implements OnInit {
getAllItems() {
this.rcService.getAllItems().subscribe((data) => {
this.items = data;
});
}
}
rxjs 5.4.0(2017-05-09)增加了对shareereplay的支持。
为什么使用共享回放?
当您有不希望在多个订阅者之间执行的副作用或繁重的计算时,通常需要使用shareReplay。在您知道流的后期订阅者需要访问先前发出的值的情况下,它可能也很有价值。这种在订阅上重放价值的能力是share和shareereplay的区别所在。
你可以很容易地修改一个angular服务来使用它,并返回一个带有缓存结果的可观察对象,它只会进行一次http调用(假设第一次调用成功)。
Angular服务示例
这是一个非常简单的客户服务,使用共享回放。
customer.service.ts
import { shareReplay } from 'rxjs/operators';
import { Observable } from 'rxjs';
import { HttpClient } from '@angular/common/http';
@Injectable({providedIn: 'root'})
export class CustomerService {
private readonly _getCustomers: Observable<ICustomer[]>;
constructor(private readonly http: HttpClient) {
this._getCustomers = this.http.get<ICustomer[]>('/api/customers/').pipe(shareReplay());
}
getCustomers() : Observable<ICustomer[]> {
return this._getCustomers;
}
}
export interface ICustomer {
/* ICustomer interface fields defined here */
}
注意,构造函数中的赋值可以移动到getCustomers方法,但由于HttpClient返回的可观察对象是“冷的”,因此在构造函数中这样做是可以接受的,因为http调用只会在订阅的第一个调用中进行。
此外,这里还假设初始返回的数据在应用程序实例的生命周期内不会过时。