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

我有以下代码:

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() 只调用一次。这是为什么？!我已经在这上面花了一整天了。因为我已经用我的转储方法订阅了可观察对象，它应该“观察”集合变化并每 3 秒重新调用一次 DumpToConsole() 方法；这就是我需要的。

想法？谢谢

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

请您参考如下方法:

您正在调用 ToObservable()在你的 BlockingCollection<int> .此扩展方法仅采用 IEnumerable<int>集合上的接口(interface)并将其转换为 IObservable<int> .这具有在订阅点获取集合内容列表的效果，并通过 Observable 流将它们转储出来。

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

使用 GetConsumingEnumerable() 在 ToObservable() 前会解决这个问题。

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

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

如果你只是添加，你可以考虑扭转整个局面——使用一个 Subject 来支持一个“Add”方法，并让一个订阅者填写一个 List(或者 BlockingCollection，如果你需要的话)来跟踪集合，以及一个然后，第二个订阅者可以报告进度。

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

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

附录

Brandon 评论说您正在处理 StartQueue 中的订阅让我意识到另一个问题 - StartQueue永远不会回来!这是因为调用 Subscribe在 ToObservable() 上制作IEnumerable 的转换在枚举完成之前不会返回 - 因此它也阻止处理(因为 IDisposable 是 Subscribe 的返回值)这就是为什么我没有注意到 using @Brandon 也指出了这一点!

有了以上两点，还需要做如下额外的改动。首先，删除 using围绕订阅声明，隐式处置将取消订阅。当我们解决阻塞的 Subscribe 调用时，这将导致订阅立即被取消。你应该保留 IDisposable如果您确实需要在某个时候明确取消订阅，请处理。

其次，添加对SubscribeOn(Scheduler.Default)的调用紧接在 ToObservable() 之后防止Subscribe来电拦截。