通过使用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结果。
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调用只会在订阅的第一个调用中进行。
此外，这里还假设初始返回的数据在应用程序实例的生命周期内不会过时。
这是.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);
}
}
我把问题打了星号，但我会试着试一下。
//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的)订阅者。
你可以简单地使用ngx-cacheable!它更适合你的场景。
使用这个的好处
它只调用rest API一次，缓存响应并为接下来的请求返回相同的响应。
在创建/更新/删除操作后，可以根据需要调用API。
那么，你的服务等级应该是这样的
import { Injectable } from '@angular/core';
import { Cacheable, CacheBuster } from 'ngx-cacheable';
const customerNotifier = new Subject();
@Injectable()
export class customersService {
// relieves all its caches when any new value is emitted in the stream using notifier
@Cacheable({
cacheBusterObserver: customerNotifier,
async: true
})
getCustomer() {
return this.http.get('/someUrl').map(res => res.json());
}
// notifies the observer to refresh the data
@CacheBuster({
cacheBusterNotifier: customerNotifier
})
addCustomer() {
// some code
}
// notifies the observer to refresh the data
@CacheBuster({
cacheBusterNotifier: customerNotifier
})
updateCustomer() {
// some code
}
}
这里有更多的参考链接。
RXJS 5.3.0
我对.map(myFunction).publishReplay(1).refCount()不满意
对于多个订阅者，.map()在某些情况下执行myFunction两次(我希望它只执行一次)。一个修复似乎是publishReplay(1).refCount().take(1)
你可以做的另一件事，就是不使用refCount()，让Observable立即热:
let obs = this.http.get('my/data.json').publishReplay(1);
obs.connect();
return obs;
这将启动HTTP请求，而不考虑订阅者。我不确定在HTTP GET完成之前取消订阅是否会取消它。
