.NET Rx - ReplaySubject緩衝區大小不起作用

我一直在使用.NET Reactive Extensions來觀察日誌事件進來。我目前使用派生自IObservable的類並使用ReplaySubject來存儲日誌，這樣我就可以過濾並重播日誌（例如：顯示所有錯誤日誌，或顯示所有詳細日誌），而不會丟失我已緩衝的日誌。.NET Rx - ReplaySubject緩衝區大小不起作用

的問題是，即使我已經設定關於這一主題的緩衝區大小：

this.subject = new ReplaySubject<LogEvent>(10);

我的程序的內存佔用穿過屋頂，當我使用OnNext添加到觀察集合上無限循環：

internal void WatchForNewEvents() 
     { 
      Task.Factory.StartNew(() => 
       { 
        while (true) 
        { 
         dynamic parameters = new ExpandoObject(); 
         // TODO: Add parameters for getting specific log events 

         if (this.logEventRepository.GetManyHasNewResults(parameters)) 
         { 
          foreach (var recentEvent in this.logEventRepository.GetMany(parameters)) 
          { 
           this.subject.OnNext(recentEvent); 
          } 
         } 

         // Commented this out for now to really see the memory go up 
         // Thread.Sleep(1000); 
        } 
       }); 
     }

ReplaySubject上的緩衝區大小不起作用嗎？當達到緩衝區大小時，似乎沒有清除緩衝區。任何幫助非常感謝！

UPDATE：

我添加的用戶像這樣（這是錯的？）：

public IDisposable Subscribe(IObserver<LogEvent> observer) 
     { 
      return this.subject.Subscribe(observer); 
     }

...這就是所謂像：

// Inserts into UI ListView 
    this.logEventObservable.Subscribe(evt => this.InsertNewLogEvent(evt));

來源

2012-01-31 Nick Ramirez

如果將訂閱者添加到ReplaySubject會發生什麼？我認爲它不應該像這樣泄漏，但我仍然很好奇，看看會發生什麼 – 2012-01-31 05:34:29

謝謝@保羅，我已經添加了更多關於如何添加訂閱者的信息。 – 2012-01-31 16:23:10

我已經隔離了這個問題，但我還不知道解決方案。在循環中調用OnNext之前訂閱主題可防止內存泄漏。訂閱AFTER導致它。 – 2012-01-31 16:47:55

我不知道，如果這是一個明確的答案，但我懷疑是因爲你使用的調度程序的併發性而遇到問題。你在ReplaySubject調用構造函數如下：

public ReplaySubject(int bufferSize) 
    : this(bufferSize, TimeSpan.MaxValue, Scheduler.CurrentThread) 
{ }

的Scheduler.CurrentThread讓我擔心。嘗試將其更改爲Scheduler.ThreadPool，看看是否有幫助。

另外，作爲一個附註，你似乎將Rx與TPL和老式線程混合睡眠混合在一起。通常最好避免這樣做。你可以改變你WatchForNewEvents代碼看起來像這樣：

dynamic parameters = new ExpandoObject(); 

var newEvents = 
    from n in Observable.Interval(TimeSpan.FromSeconds(1.0)) 
    where this.logEventRepository.GetManyHasNewResults(parameters) 
    from recentEvent in 
     this.logEventRepository.GetMany(parameters).ToObservable() 
    select recentEvent; 

newEvents.Subscribe(this.subject);

這就是做事的一個很好的緊湊的Rx-Y方式。

來源

2012-01-31 07:02:58 Enigmativity

感謝您的建議，但更改調度程序沒有效果。我不知道我的IObservable的實現是否不正確？ – 2012-01-31 16:22:36

@NickRamirez - 對不起，我的預感沒有奏效。我建議不要試圖實現'IObservable' - 這通常會更復雜，而且它很容易出錯。改爲使用內置的運算符。 – Enigmativity 2012-01-31 23:36:36

最後，按照你的建議，使用Observable.Interval代替TPL修復內存泄漏。謝謝！ – 2012-02-01 01:15:09

.NET Rx - ReplaySubject緩衝區大小不起作用

回答

相關問題