Aggregate large number of observables into new observable
I have, let's say, a 1000 observables. Now I want to aggregate all the events into a new observable that fires OnNext 开发者_开发问答once all the others have sent an event. Whats the best way to do that using Rx?
Update: Some great feedback over at the Rx forum, especially by Dave Sexton. He showed how to create a Zip extension method that takes multiple observables: http://social.msdn.microsoft.com/Forums/en-US/rx/thread/daaa84db-b560-4eda-871e-e523098db20c/
There is a MailboxProcessor in F#... I would use a SynchronizationContext in C# for the same purpose. Give me a few minutes and I will write up an example.
Aside: Here's my code in F# that does something similar... It will be considerably more effort, but still doable in C# with Rx.
open System.Diagnostics
let numWorkers = 20
let asyncDelay = 100
type MessageForMailbox =
| DataMessage of AsyncReplyChannel<unit>
| GetSummary of AsyncReplyChannel<unit>
let main =
let actor =
MailboxProcessor.Start( fun inbox ->
let rec loop acc =
async {
let! message = inbox.Receive()
match message with
| DataMessage replyChannel -> replyChannel.Reply(); return! loop acc
| GetSummary replyChannel -> replyChannel.Reply(); return! loop acc
}
loop 0 // seed for acc
)
let codeBlocks = [for i in 1..numWorkers ->
async {
do! Async.Sleep asyncDelay
return! actor.PostAndAsyncReply DataMessage
} ]
while true do
printfn "Concurrent started..."
let sw = new Stopwatch()
sw.Start()
codeBlocks |> Async.Parallel |> Async.RunSynchronously |> ignore
actor.PostAndReply GetSummary
sw.Stop()
printfn "Concurrent in %d millisec" sw.ElapsedMilliseconds
printfn "efficiency: %d%%" (int64 (asyncDelay * 100) / sw.ElapsedMilliseconds)
printfn "Synchronous started..."
let sw = new Stopwatch()
sw.Start()
for codeBlock in codeBlocks do codeBlock |> Async.RunSynchronously |> ignore
sw.Stop()
printfn "Synchronous in %d millisec" sw.ElapsedMilliseconds
printfn "efficiency: %d%%" (int64 (asyncDelay * numWorkers * 100) / sw.ElapsedMilliseconds)
main
精彩评论