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完成之前取消订阅是否会取消它。

根据@Cristian的建议，这是一种适用于HTTP可观察对象的方法，它只发射一次，然后就完成了:

getCustomer() {
    return this.http.get('/someUrl')
        .map(res => res.json()).publishLast().refCount();
}

你可以简单地使用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
    }
}

这里有更多的参考链接。

我把问题打了星号，但我会试着试一下。

//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的)订阅者。

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调用只会在订阅的第一个调用中进行。

此外，这里还假设初始返回的数据在应用程序实例的生命周期内不会过时。

您选择的实现将取决于是否希望unsubscribe()取消您的HTTP请求。

在任何情况下，TypeScript装饰器都是标准化行为的好方法。这是我写的:

  @CacheObservableArgsKey
  getMyThing(id: string): Observable<any> {
    return this.http.get('things/'+id);
  }

装饰器定义:

/**
 * Decorator that replays and connects to the Observable returned from the function.
 * Caches the result using all arguments to form a key.
 * @param target
 * @param name
 * @param descriptor
 * @returns {PropertyDescriptor}
 */
export function CacheObservableArgsKey(target: Object, name: string, descriptor: PropertyDescriptor) {
  const originalFunc = descriptor.value;
  const cacheMap = new Map<string, any>();
  descriptor.value = function(this: any, ...args: any[]): any {
    const key = args.join('::');

    let returnValue = cacheMap.get(key);
    if (returnValue !== undefined) {
      console.log(`${name} cache-hit ${key}`, returnValue);
      return returnValue;
    }

    returnValue = originalFunc.apply(this, args);
    console.log(`${name} cache-miss ${key} new`, returnValue);
    if (returnValue instanceof Observable) {
      returnValue = returnValue.publishReplay(1);
      returnValue.connect();
    }
    else {
      console.warn('CacheHttpArgsKey: value not an Observable cannot publishReplay and connect', returnValue);
    }
    cacheMap.set(key, returnValue);
    return returnValue;
  };

  return descriptor;
}