示例 Plunkr: https://plnkr.co/edit/NZwb3ol8CbZFtSc6Q9zm?p=preview
我知道 RxJS（5.0 beta.4）有这三种节流方法：
auditTime()、throttleTime() 和 debounceTime()
我要寻找的行为与 lodash 在默认情况下执行的 throttle 相同：
理论上应该看起来像：
inputObservable
.do(() => cancelPreviousRequest())
.throttleTime(500)
.subscribe((value) => doNextRequest(value))
但是
throttleTime如果在节流超时期间被发出，不会给我最后一个值debounceTime不会立即触发auditTime不会立即触发我能否组合任何RxJS方法来实现所描述的行为？
对于在2018年之后寻找此功能的任何人：此功能已经添加了一年多，但由于某种原因，文档尚未更新。
您可以将配置对象直接传递给throttleTime。默认值为{ leading: true, trailing: false }。要实现此处讨论的行为，只需将trailing设置为true：{ leading: true, trailing: true }
编辑：
为了完整起见，这里是一个有效的片段：
import { asyncScheduler } from 'rxjs'
import { throttleTime } from 'rxjs/operators'
...
observable.pipe(
throttleTime(100, asyncScheduler, { leading: true, trailing: true })
)
对于旧版本的RxJs，我编写了一个名为concatLatest的操作符，它可以完成大部分你所需的操作。使用它，您可以通过以下代码获得您所需的节流行为：
const delay = Rx.Observable.empty().delay(500);
inputObservable
.map(value => Rx.Observable.of(value).concat(delay))
.concatLatest()
.subscribe(...);
这里是操作员。我尝试更新它以与RxJS5一起使用：
Rx.Observable.prototype.concatLatest = function () {
/// <summary>
/// Concatenates an observable sequence of observable sequences, skipping sequences that arrive while the current sequence is being observed.
/// If N new observables arrive while the current observable is being observed, the first N-1 new observables will be thrown
/// away and only the Nth will be observed.
/// </summary>
/// <returns type="Rx.Observable"></returns>
var source = this;
return Rx.Observable.create(function (observer) {
var latest,
isStopped,
isBusy,
outerSubscription,
innerSubscription,
subscriptions = new Rx.Subscription(function () {
if (outerSubscription) {
outerSubscription.unsubscribe();
}
if (innerSubscription) {
innerSubscription.unsubscribe();
}
}),
onError = observer.error.bind(observer),
onNext = observer.next.bind(observer),
innerOnComplete = function () {
var inner = latest;
if (inner) {
latest = undefined;
if (innerSubscription) {
innerSubscription.unsubscribe();
}
innerSubscription = inner.subscribe(onNext, onError, innerOnComplete);
}
else {
isBusy = false;
if (isStopped) {
observer.complete();
}
}
};
outerSubscription = source.subscribe(function (newInner) {
if (isBusy) {
latest = newInner;
}
else {
isBusy = true;
if (innerSubscription) {
innerSubscription.unsubscribe();
}
innerSubscription = newInner.subscribe(onNext, onError, innerOnComplete);
}
}, onError, function () {
isStopped = true;
if (!isBusy) {
observer.complete();
}
});
return subscriptions;
});
};
新的plunker: https://plnkr.co/edit/4NkXsOeJOSrLUP9WEtp0?p=preview
原始链接:
更改内容:
从(auditTime):
protected _next(value: T): void {
this.value = value;
this.hasValue = true;
if (!this.throttled) {
this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this));
}
}
clearThrottle(): void {
const { value, hasValue, throttled } = this;
if (throttled) {
this.remove(throttled);
this.throttled = null;
throttled.unsubscribe();
}
if (hasValue) {
this.value = null;
this.hasValue = false;
this.destination.next(value);
}
}
to (auditTimeImmediate):
protected _next(value: T): void {
this.value = value;
this.hasValue = true;
if (!this.throttled) {
// change 1:
this.clearThrottle();
}
}
clearThrottle(): void {
const { value, hasValue, throttled } = this;
if (throttled) {
this.remove(throttled);
this.throttled = null;
throttled.unsubscribe();
}
if (hasValue) {
this.value = null;
this.hasValue = false;
this.destination.next(value);
// change 2:
this.add(this.throttled = this.scheduler.schedule(dispatchNext, this.duration, this));
}
}
我在next操作后开始计时。
用法：
inputObservable
.do(() => cancelPreviousRequest())
.auditTimeImmediate(500)
.subscribe((value) => doNextRequest(value))
throttle(() => Promise.resolve(...), { leading: true, trailing:true })- 我发现这对于立即运行ajax放置并等待完成，然后在第一个完成时运行最终的放置系列特别有用。 - Josh Mc