异步队列处理与活性扩展 [英] Async Queue Processing With Reactive Extensions

问题描述

有一对夫妇在这个文章，我有这个工作......但我想知道如何设置工作线程我观测订阅的最大数量的一次。

There are a couple of articles on this, and I have this working...but I want to know how to set a max number of Task threads for my Observable subscriptions at once.

我有以下并行异步保存日志条目：

I have the following to parallelize async saving of log entries:

private BlockingCollection<ILogEntry> logEntryQueue;

和

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

要安排我的储蓄...但我怎么指定最大线程调度使用一次？

To schedule my saving...but how do I specify the max threads for the scheduler to use at once?

推荐答案

这是不是可观测的功能，但是调度的功能。可观察到定义的什么的和调度程序定义的其中的。

This is not a function of the Observable, but a function of the Scheduler. The Observable defines what and the scheduler defines where.

您就需要传递一个自定义的调度。一个简单的方法，这样做将是子类的TaskScheduler并重写了MaximumConcurrencyLevel属性。

You'd need to pass in a custom scheduler. A simple way to do this would be to subclass TaskScheduler and override the "MaximumConcurrencyLevel" property.

<一个href="http://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler.maximumconcurrencylevel.aspx">http://msdn.microsoft.com/en-us/library/system.threading.tasks.taskscheduler.maximumconcurrencylevel.aspx

我居然发现这个样本MSDN上：

I actually found a sample of this on MSDN:

<一个href="http://msdn.microsoft.com/en-us/library/ee789351.aspx">http://msdn.microsoft.com/en-us/library/ee789351.aspx

编辑：：您问到如何去从的TaskScheduler到IScheduler。另一个开发商只给我的信息是一点：

You asked about how to go from TaskScheduler to IScheduler. Another developer just gave me that little bit of info:

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

这篇关于异步队列处理与活性扩展的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！