.net - 使用响应式(Reactive)扩展的异步队列处理

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。另一位开发人员刚刚给了我一点信息:

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