在不跳過值的情況下節流接收Rx.Observable

Question

Throttle如果其他人跟隨得太快，則會跳過可觀察序列的值。但我需要一種方法來延遲它們。也就是說，我需要設置項目之間的最小延遲，而不會跳過任何。

實際例子：有一個Web服務可以接受請求不會比每秒一次更快;有一個用戶可以添加請求，單個或批量。沒有Rx，我會創建一個列表和一個計時器。當用戶添加請求時，我會將它們添加到列表中。在計時器事件中，我會檢查列表是否爲空。如果不是，我會發送請求並刪除相應的項目。用鎖和所有的東西。現在，通過Rx，我可以創建Subject，當用戶添加請求時添加項目。但我需要一種方法來確保Web服務不會因應用延遲而被淹沒。

我是Rx的新手，所以也許我錯過了一些明顯的東西。

Answer 1

有一個相當簡單的方法做你想要使用EventLoopScheduler什麼。

我從一個可觀測值開始，每0到3秒會隨機產生一次值。

var rnd = new Random(); 

var xs = 
    Observable 
     .Generate(
      0, 
      x => x < 20, 
      x => x + 1, 
      x => x, 
      x => TimeSpan.FromSeconds(rnd.NextDouble() * 3.0)); 

現在，爲了使這個輸出值立即除非最後一個值是內一秒鐘前我這樣做：

var ys = 
    Observable.Create<int>(o => 
    { 
     var els = new EventLoopScheduler(); 
     return xs 
      .ObserveOn(els) 
      .Do(x => els.Schedule(() => Thread.Sleep(1000))) 
      .Subscribe(o); 
    }); 

這有效地觀察到在EventLoopScheduler源，然後把它睡在每個OnNext後1秒，以便它在喚醒後才能開始下一個OnNext。

我測試了一下它的工作與此代碼：

ys 
    .Timestamp() 
    .Select(x => x.Timestamp.Second + (double)x.Timestamp.Millisecond/1000.0) 
    .Subscribe(x => Console.WriteLine(x)); 

我希望這有助於。

Answer 2

如何使用可觀察計時器從阻塞隊列中取出？下面的代碼是未經測試，但應該給你的我的意思的想法...

//assuming somewhere there is 
BlockingCollection<MyWebServiceRequestData> workQueue = ... 

Observable 
    .Timer(new TimeSpan(0,0,1), new EventLoopScheduler()) 
    .Do(i => myWebService.Send(workQueue.Take())); 

// Then just add items to the queue using workQueue.Add(...) 

Answer 3

考慮使用緩衝區。

http://msdn.microsoft.com/en-us/library/hh212130(v=vs.103).aspx

正如其名稱所示，它緩衝事件的觀察到的數據流中的元素沒有「滴」任何人。

-G

Answer 4

怎麼樣一個簡單的擴展方法：

public static IObservable<T> StepInterval<T>(this IObservable<T> source, TimeSpan minDelay) 
{ 
    return source.Select(x => 
     Observable.Empty<T>() 
      .Delay(minDelay) 
      .StartWith(x) 
    ).Concat(); 
} 

用法：

var bufferedSource = source.StepInterval(TimeSpan.FromSeconds(1)); 

Answer 5

.Buffer(TimeSpan.FromSeconds(0.2)).Where(i => i.Any()) 
.Subscribe(buffer => 
{ 
    foreach(var item in buffer) Console.WriteLine(item) 
}); 

Answer 6

我想建議使用Observable.Zip的方法：

// Incoming requests 
var requests = new[] {1, 2, 3, 4, 5}.ToObservable(); 

// defines the frequency of the incoming requests 
// This is the way to emulate flood of incoming requests. 
// Which, by the way, uses the same approach that will be used in the solution 
var requestsTimer = Observable.Interval(TimeSpan.FromSeconds(0.1)); 
var incomingRequests = Observable.Zip(requests, requestsTimer, (number, time) => {return number;}); 
incomingRequests.Subscribe((number) => 
{ 
    Console.WriteLine($"Request received: {number}"); 
}); 

// This the minimum interval at which we want to process the incoming requests 
var processingTimeInterval = Observable.Interval(TimeSpan.FromSeconds(1)); 

// Zipping incoming requests with the interval 
var requestsToProcess = Observable.Zip(incomingRequests, processingTimeInterval, (data, time) => {return data;}); 

requestsToProcess.Subscribe((number) => 
{ 
    Console.WriteLine($"Request processed: {number}"); 
}); 

Answer 7

我正在玩這個，發現。郵編（如前所述）是最簡單的方法：

var stream = "ThisFastObservable".ToObservable(); 
var slowStream = 
    stream.Zip(
     Observable.Interval(TimeSpan.FromSeconds(1)), //Time delay 
     (x, y) => x); // We just care about the original stream value (x), not the interval ticks (y) 

slowStream.TimeInterval().Subscribe(x => Console.WriteLine($"{x.Value} arrived after {x.Interval}")); 

輸出：

T arrived after 00:00:01.0393840 
h arrived after 00:00:00.9787150 
i arrived after 00:00:01.0080400 
s arrived after 00:00:00.9963000 
F arrived after 00:00:01.0002530 
a arrived after 00:00:01.0003770 
s arrived after 00:00:00.9963710 
t arrived after 00:00:01.0026450 
O arrived after 00:00:00.9995360 
b arrived after 00:00:01.0014620 
s arrived after 00:00:00.9993100 
e arrived after 00:00:00.9972710 
r arrived after 00:00:01.0001240 
v arrived after 00:00:01.0016600 
a arrived after 00:00:00.9981140 
b arrived after 00:00:01.0033980 
l arrived after 00:00:00.9992570 
e arrived after 00:00:01.0003520