Observable 对不同线程上更改的阻塞集合没有反应

Question

我有以下代码：

class Program
{
    static void Main(string[] args)
    {
        var watcher = new SNotifier(DumpToConsole);
        watcher.StartQueue();

        Console.ReadLine();
    }

    private static void DumpToConsole(IList<Timestamped<int>> currentCol)
    {
        Console.WriteLine("buffer time elapsed, current collection contents is: {0} items.", currentCol.Count);
        Console.WriteLine("holder has: {0}", currentCol.Count);
    }
}

SNotifier：

public class SNotifier
{
    private BlockingCollection<int> _holderQueue;
    private readonly Action<IList<Timestamped<int>>> _dumpAction;

    public SNotifier(Action<IList<Timestamped<int>>> dumpAction)
    {
        PopulateListWithStartValues();
        _dumpAction = dumpAction;
    }

    public void StartQueue()
    {
        PopulateQueueOnDiffThread();

        var observableCollection = _holderQueue.ToObservable();

        var myCollectionTimestamped = observableCollection.Timestamp();
        var bufferedTimestampedCollection = myCollectionTimestamped.Buffer(TimeSpan.FromSeconds(3), TimeSpan.FromSeconds(3));

        using (bufferedTimestampedCollection.Subscribe(_dumpAction))
        {
            Console.WriteLine("started observing collection");
        }
    }

    private void PopulateQueueOnDiffThread()
    {
        Action addToCollectionAction = AddToCollection;
        var t = new TaskFactory();
        t.StartNew(addToCollectionAction);

    }

    private static IEnumerable<int> GetInitialElements()
    {
        var random = new Random();
        var items = new List<int>();
        for (int i = 0; i < 10; i++)
            items.Add(random.Next(1, 10));

        return items;
    }

    private void AddToCollection()
    {
        while (true)
        {
            var newElement = new Random().Next(1, 10);
            _holderQueue.Add(newElement);
            Console.WriteLine("added {0}", newElement);
            Console.WriteLine("holder has: {0}", _holderQueue.Count);
            Thread.Sleep(1000);
        }
    }

    private void PopulateListWithStartValues()
    {
        _holderQueue = new BlockingCollection<int>();
        var elements = GetInitialElements();
        foreach (var i in elements)
            _holderQueue.Add(i);
    }
}

我需要运行 DumpToConsole() 方法以每 3 秒 显示一次集合计数，而此集合的内容正在另一个线程上更改。我的问题是 DumpToConsole() 只调用一次。这是为什么？！我已经花了一整天的时间在这上面。由于我已经使用我的 dump 方法订阅了 observable，它应该“观察”集合的变化并每 3 秒重新调用 DumpToConsole() 方法；这就是我需要的。

想法？谢谢

（PS 传递给 SNotifier 类的操作是我在 SNotifier 中删除控制台相关内容的方法，我需要更好地重构它，它可以忽略，因为它与问题本身无关）

Answer 1

您正在通过您的BlockingCollection<int> 呼叫ToObservable()。这个扩展方法只需要集合上的IEnumerable<int> 接口并将其转换为IObservable<int>。这具有在订阅点获取集合内容列表并通过 Observable 流转储出来的效果。

它不会在添加项目时继续枚举它们。

在ToObservable() 前面使用GetConsumingEnumerable() 可以解决这个问题。

但是，需要小心，因为这也会从集合中删除项目，这可能是不可取的。

如果这是可以接受的，您可能希望在有多个订阅者的情况下发布结果 observable 以避免造成严重破坏。

如果您只是添加，您可以考虑扭转整个局面 - 使用主题支持“添加”方法并让一个订阅者填写列表（或 BlockingCollection，如果需要）来跟踪集合，以及然后第二个订阅者可以报告进度。

另一种方法是使用ObservableCollection 并订阅其事件。

在最后两个建议中，您需要使“添加”线程安全，因为 Subject<T> 和 ObservableCollection<T> 本身都不是线程安全的。

附录

Brandon 关于您在 StartQueue 中处理订阅的评论让我意识到另一个问题 - StartQueue 永远不会回来！这是因为在 IEnumerable 的 ToObservable() 转换上对 Subscribe 的调用在枚举完成之前不会返回 - 因此它也支持处置（因为 IDisposable 是 Subscribe 的返回值) 这就是为什么我也没有注意到using@Brandon 指出的原因！

结合以上两点，您需要进行以下附加更改。首先，删除订阅周围的using 语句，隐式处置会取消订阅。当我们解决阻塞的订阅调用时，这将导致订阅立即被取消。如果您确实需要在某些时候明确取消订阅，则应保留 IDisposable 句柄。

其次，在ToObservable() 之后立即添加对SubscribeOn(Scheduler.Default) 的调用，以防止Subscribe 调用阻塞。