Xử lý hàng đợi không đồng bộ với phần mở rộng phản ứng

Question

Có một vài bài viết về điều này, và tôi làm việc này ... nhưng tôi muốn biết cách đặt số lượng chủ đề Task tối đa cho các đăng ký quan sát cùng một lúc.

Tôi có sau đây để parallelize tiết kiệm async các mục nhật ký:

private BlockingCollection<ILogEntry> logEntryQueue; 

và

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

Để lập kế hoạch tiết kiệm của tôi ... nhưng làm thế nào để xác định các chủ đề tối đa cho scheduler để sử dụng cùng một lúc?

Answer 1

Đây không phải là chức năng của Quan sát, mà là chức năng của Trình lập lịch biểu. Đài quan sát xác định những gì và trình lên lịch xác định trong đó.

Bạn cần chuyển qua trình lên lịch tùy chỉnh. Một cách đơn giản để làm điều này sẽ là phân lớp TaskScheduler và ghi đè thuộc tính "MaximumConcurrencyLevel".

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

Tôi thực sự tìm thấy một mẫu xe này trên MSDN:

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

Edit: Bạn hỏi về làm thế nào để đi từ TaskScheduler để IScheduler. Một nhà phát triển chỉ cho tôi rằng một chút thông tin:

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

Answer 2

Nếu bạn tạo "tác phẩm" của bạn như IObservable<T> với thực hiện chậm (. tức là họ muốn làm bất cứ điều gì cho đến khi đăng ký đến), bạn có thể sử dụng quá tải Merge chấp nhận một số đăng ký đồng thời tối đa:

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());