RX.net 的 DropQueue 机制答案

【问题标题】：DropQueue mechanism for RX.netRX.net 的 DropQueue 机制
【发布时间】：2017-10-28 15:41:56
【问题描述】：

我遇到了 RX.net 的背压问题，我找不到解决方案。我有一个可观察的实时日志消息流。

var logObservable = /* Observable stream of log messages */

我想通过 TCP 接口公开它，该接口在通过网络发送来自logObservable 的实时日志消息之前对其进行序列化。所以我做了以下事情：

foreach (var message in logObservable.ToEnumerable())
{
   // 1. Serialize message
   // 2. Send it over the wire. 
}

如果发生背压情况，.ToEnumerable() 就会出现问题，例如如果另一端的客户端暂停流。问题是.ToEnumerable() 缓存了导致大量内存使用的项目。我正在寻找一种类似于 DropQueue 的机制，它只缓冲最后 10 条消息，例如

var observableStream = logObservable.DropQueue(10).ToEnumerable();

这是解决此问题的正确方法吗？你知道要实现这样的机制来避免可能出现的背压问题吗？

【问题讨论】：

.take(10).toenumerable()? 会工作，不是吗？
我希望通过网络获得连续的日志消息流。如果我按照您的建议做，它不仅需要 10 条日志消息，然后完成可观察流吗？我要解决的问题是，如果客户端太慢而无法检索日志消息或暂停它应该只缓存的流，例如10 个项目，而不是无限数量的项目。
.Throttle(...) 或 .Sample(..) 怎么样？
@Enigmativity：如果客户能跟上，我想避免.Throttle() 或.Sample()。我试图涵盖的情况是，如果客户决定，例如暂停流然后它不应该缓存项目。我找到了运算符.Next() 和.Latest()，它们可能可以用来代替.ToEnumerable()，但我找不到足够的文档。
经过一番研究，我在github.com/ReactiveX/RxJava/issues/59上找到了一个弹珠图，看起来可以用。 Latest() 类似于 .DropQueue(1)。他们写道：“对于Latest，Iterator的next会检查是否有缓存值，如果有，返回缓存值，并删除缓存值。如果没有缓存值，它将阻塞直到下一个value 从 Observable 发出，并返回它。

标签： c# c#-4.0 system.reactive reactive-programming reactive

【解决方案1】：

我的DropQueue 实现：

    public static IEnumerable<TSource> ToDropQueue<TSource>(
        this IObservable<TSource> source,
        int queueSize,
        Action backPressureNotification = null,
        CancellationToken token = default(CancellationToken))
    {
        var queue = new BlockingCollection<TSource>(new ConcurrentQueue<TSource>(), queueSize);
        var isBackPressureNotified = false;

        var subscription = source.Subscribe(
            item =>
            {
                var isBackPressure = queue.Count == queue.BoundedCapacity;

                if (isBackPressure)
                {
                    queue.Take(); // Dequeue an item to make space for the next one

                    // Fire back-pressure notification if defined
                    if (!isBackPressureNotified && backPressureNotification != null)
                    {
                        backPressureNotification();
                        isBackPressureNotified = true;
                    }
                }
                else
                {
                    isBackPressureNotified = false;
                }

                queue.Add(item);
            },
            exception => queue.CompleteAdding(),
            () => queue.CompleteAdding());

        token.Register(() => { subscription.Dispose(); });

        using (new CompositeDisposable(subscription, queue))
        {
            foreach (var item in queue.GetConsumingEnumerable())
            {
                yield return item;
            }
        }
    }

【讨论】：