다수의 관측 값을 새로운 관측 값으로 집계

Question

나는 1000 개의 관측 값을 가지고 있다고 가정 해 보겠습니다. 이제 모든 이벤트가 OnNext를 트리거 한 새로운 관찰 가능으로 집계하여 다른 이벤트를 보냈 으면합니다. Rx를 사용하여 가장 좋은 방법은 무엇입니까?

업데이트 : Rx 포럼에서 특히 Dave Sexton의 위대한 피드백. 그는 여러 관측 값을 취하는 Zip 확장 메서드를 만드는 방법을 보여주었습니다. http://social.msdn.microsoft.com/Forums/en-US/rx/thread/daaa84db-b560-4eda-871e-e523098db20c/

Answer 1

F #에 MailboxProcessor가 있습니다. 동일한 목적으로 C#에서 SynchronizationContext를 사용합니다. 몇 분만 기다려주세요. 예를 쓸 것입니다.

참고 사항 : 여기에 F #의 코드가 있습니다. 비슷한 점이 있습니다 ... C#에서 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