我正在使用反应式扩展将数据聚合到100毫秒的缓冲区中：
this.subscription = this.dataService
.Where(x => !string.Equals("FOO", x.Key.Source))
.Buffer(TimeSpan.FromMilliseconds(100))
.ObserveOn(this.dispatcherService)
.Where(x => x.Count != 0)
.Subscribe(this.OnBufferReceived);
这个代码可以正常工作。然而，我想要的行为略有不同于Buffer操作提供的行为。实际上，我希望在接收到其他数据项时重置计时器。只有当整个100ms内没有收到任何数据时才处理它。这打开了永远不处理数据的可能性，因此我还应该能够指定最大计数。我想象中的代码应该是这样的：
.SlidingBuffer(TimeSpan.FromMilliseconds(100), 10000)
我已经查过了，但是在Rx中没有找到类似的东西？有人可以证实/否认这一点吗？
通过组合内置的Window和Throttle方法可以实现。这两个方法都在Observable类中定义。首先，让我们解决简单问题，即忽略最大计数条件：
public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay)
{
var closes = stream.Throttle(delay);
return stream.Window(() => closes).SelectMany(window => window.ToList());
}
强大的Window方法完成了繁重的工作。现在很容易看到如何添加最大计数：
public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? max=null)
{
var closes = stream.Throttle(delay);
if (max != null)
{
var overflows = stream.Where((x,index) => index+1>=max);
closes = closes.Merge(overflows);
}
return stream.Window(() => closes).SelectMany(window => window.ToList());
}
我将在我的博客上写一篇文章来解释这个问题。 https://gist.github.com/2244036
Window方法的文档：
我编写了一个扩展程序，可以完成大部分你想要的功能 - BufferWithInactivity。
这是它的代码：
public static IObservable<IEnumerable<T>> BufferWithInactivity<T>(
this IObservable<T> source,
TimeSpan inactivity,
int maximumBufferSize)
{
return Observable.Create<IEnumerable<T>>(o =>
{
var gate = new object();
var buffer = new List<T>();
var mutable = new SerialDisposable();
var subscription = (IDisposable)null;
var scheduler = Scheduler.ThreadPool;
Action dump = () =>
{
var bts = buffer.ToArray();
buffer = new List<T>();
if (o != null)
{
o.OnNext(bts);
}
};
Action dispose = () =>
{
if (subscription != null)
{
subscription.Dispose();
}
mutable.Dispose();
};
Action<Action<IObserver<IEnumerable<T>>>> onErrorOrCompleted =
onAction =>
{
lock (gate)
{
dispose();
dump();
if (o != null)
{
onAction(o);
}
}
};
Action<Exception> onError = ex =>
onErrorOrCompleted(x => x.OnError(ex));
Action onCompleted = () => onErrorOrCompleted(x => x.OnCompleted());
Action<T> onNext = t =>
{
lock (gate)
{
buffer.Add(t);
if (buffer.Count == maximumBufferSize)
{
dump();
mutable.Disposable = Disposable.Empty;
}
else
{
mutable.Disposable = scheduler.Schedule(inactivity, () =>
{
lock (gate)
{
dump();
}
});
}
}
};
subscription =
source
.ObserveOn(scheduler)
.Subscribe(onNext, onError, onCompleted);
return () =>
{
lock (gate)
{
o = null;
dispose();
}
};
});
}
使用 Rx Extensions 2.0，您可以通过接受超时和大小的新缓冲区过载来满足这两个要求：
this.subscription = this.dataService
.Where(x => !string.Equals("FOO", x.Key.Source))
.Buffer(TimeSpan.FromMilliseconds(100), 1)
.ObserveOn(this.dispatcherService)
.Where(x => x.Count != 0)
.Subscribe(this.OnBufferReceived);
Observable.Throttle。你所要求的更加复杂，但我猜可以用 Observable.Window 来实现。无论如何，除非我漏掉了什么，我的答案与此问题上被接受的答案完全相同。 - Sébastien Lorion正如 Rohit Sharma 在 Colonel Panic 的 solution 中所提到的，存在一个问题，即项目将被缓冲而不会被推送给订阅者，除非生成一个项目。
正如在这个 comment 中所描述的那样，问题在于 p.Window(() => closes)，因为它打开了一个间隙，在这个间隙中可能会错过事件。
每次处理窗口后，该 lambda 将被调用。 Window 运算符将在每次调用 Subscribe 时调用 lambda 返回的内容，因为就其而言，你每次都可能从 lambda 返回完全不同的 IObservable。
由于现在始终使用相同的 lambda，我们需要调整 maxCount。如果不进行更改，则 maxCount 永远不会被重置，一旦它达到最大值，每个新事件都将超过 maxCount。
public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? maxCount = null)
{
var publish = stream.Publish(p =>
{
var closes = p.Throttle(delay);
if (maxCount != null)
{
Int32 i = 0;
var overflows = p.Where(x =>
{
++i;
if (i >= maxCount)
{
i = 0;
return true;
}
return false;
});
closes = closes.Merge(overflows);
}
return p.Window(closes).SelectMany(window => window.ToList());
});
return publish;
}
更新：
经过进一步的测试，我发现仍然有些情况下，项目将不能正确地推送给订阅者。
这里是我们已经使用4个月而没有任何问题的解决方法。
解决方法是添加 .Delay(...) 与任何 TimeSpan。
public static IObservable<IList<T>> BufferUntilInactive<T>(this IObservable<T> stream, TimeSpan delay, Int32? maxCount = null)
{
var publish = stream.Publish(p =>
{
var closes = p.Throttle(delay);
if (maxCount != null)
{
var overflows = stream.Where((x, index) => index + 1 >= maxCount);
closes = closes.Merge(overflows);
}
return p.Window(() => closes).SelectMany(window => window.ToList()).Delay(TimeSpan.Zero);
});
return publish;
}
我想这可以通过如下所示的Buffer方法来实现：
public static IObservable<IList<T>> SlidingBuffer<T>(this IObservable<T> obs, TimeSpan span, int max)
{
return Observable.CreateWithDisposable<IList<T>>(cl =>
{
var acc = new List<T>();
return obs.Buffer(span)
.Subscribe(next =>
{
if (next.Count == 0) //no activity in time span
{
cl.OnNext(acc);
acc.Clear();
}
else
{
acc.AddRange(next);
if (acc.Count >= max) //max items collected
{
cl.OnNext(acc);
acc.Clear();
}
}
}, err => cl.OnError(err), () => { cl.OnNext(acc); cl.OnCompleted(); });
});
}
Colonel Panic的solution几乎完美。唯一缺少的是一个Publish组件，以便使解决方案也适用于冷序列。
/// <summary>
/// Projects each element of an observable sequence into a buffer that's sent out
/// when either a given inactivity timespan has elapsed, or it's full,
/// using the specified scheduler to run timers.
/// </summary>
public static IObservable<IList<T>> BufferUntilInactive<T>(
this IObservable<T> source, TimeSpan dueTime, int maxCount,
IScheduler scheduler = default)
{
if (maxCount < 1) throw new ArgumentOutOfRangeException(nameof(maxCount));
scheduler ??= Scheduler.Default;
return source.Publish(published =>
{
var combinedBoundaries = Observable.Merge
(
published.Throttle(dueTime, scheduler),
published.Skip(maxCount - 1)
);
return published
.Window(() => combinedBoundaries)
.SelectMany(window => window.ToList());
});
}