リアクティブエクステンションを使用した非同期キュー処理

Question

これに関するいくつかの記事がありますが、これは機能していますが、Observableサブスクリプションの最大数のタスクスレッドを一度に設定する方法を知りたいと思います。

私は、ログエントリの非同期省並列化するために、次のしている：私の節約をスケジュールするには

private BlockingCollection<ILogEntry> logEntryQueue; 

と

logEntryQueue = new BlockingCollection<ILogEntry>(); 
logEntryQueue.GetConsumingEnumerable().ToObservable(Scheduler.TaskPool).Subscribe(SaveLogEntry); 

を...しかし、どのように私はへのスケジューラの最大スレッドを指定してください一度に使用しますか？

Answer 1

これはObservableの機能ではなく、Schedulerの機能です。 Observableはを定義し、とスケジューラはを定義します。です。

カスタムスケジューラを渡す必要があります。これを行う簡単な方法は、TaskSchedulerをサブクラス化し、 "MaximumConcurrencyLevel"プロパティをオーバーライドすることです。

http://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler.maximumconcurrencylevel.aspx

私は実際にはMSDNでこのサンプルが見つかりました：

http://msdn.microsoft.com/en-us/library/ee789351.aspx

編集：あなたはTaskSchedulerからISchedulerに行く方法についての質問を。あなたは（。すなわち、それらはに加入するまで何もしたい）繰延実行をIObservable<T>としてあなたの「仕事」を作成した場合

var ischedulerForRx = new TaskPoolScheduler 
(
    new TaskFactory 
    (
     //This is your custom scheduler 
     new LimitedConcurrencyLevelTaskScheduler(1) 
    ) 
); 

Answer 2

、あなたが受け入れるMergeオーバーロードを使用することができます別の開発者は、単に情報の少しことを私に与えました

ISubject<QueueItem> synchronizedQueue = new Subject<QueueItem>().Synchronize(); 

queue 
    .Select(item => StartWork(item)) 
    .Merge(maxConcurrent: 5) // C# 4 syntax for illustrative purposes 
    .Subscribe(); 

// To enqueue: 
synchronizedQueue.OnNext(new QueueItem());