C# xếp hàng các nhiệm vụ phụ thuộc để được xử lý bởi một nhóm hồ sơ

Question

Tôi muốn xếp hàng các tác vụ phụ thuộc vào một số luồng cần được xử lý theo thứ tự (trong mỗi luồng). Các luồng có thể được xử lý song song.

Để cụ thể, giả sử tôi cần hai hàng đợi và tôi muốn các tác vụ trong mỗi hàng đợi được xử lý theo thứ tự. Đây là giả mẫu để minh họa cho hành vi mong muốn:

Queue1_WorkItem wi1a=...; 

enqueue wi1a; 

... time passes ... 

Queue1_WorkItem wi1b=...; 

enqueue wi1b; // This must be processed after processing of item wi1a is complete 

... time passes ... 

Queue2_WorkItem wi2a=...; 

enqueue wi2a; // This can be processed concurrently with the wi1a/wi1b 

... time passes ... 

Queue1_WorkItem wi1c=...; 

enqueue wi1c; // This must be processed after processing of item wi1b is complete 

Dưới đây là một sơ đồ với các mũi tên minh họa phụ thuộc giữa các hạng mục công trình:

Câu hỏi đặt ra là làm thế nào để tôi làm điều này bằng C# 4.0/.NET 4.0? Ngay bây giờ tôi có hai chuỗi công nhân, mỗi luồng một hàng và tôi sử dụng một BlockingCollection<> cho mỗi hàng đợi. Thay vào đó, tôi muốn sử dụng nhóm chủ đề .NET và có các luồng công việc xử lý đồng thời các mục (qua các luồng), nhưng trong một luồng. Nói cách khác tôi muốn để có thể chỉ ra rằng ví dụ wi1b phụ thuộc vào hoàn thành wi1a, mà không cần phải theo dõi hoàn thành và nhớ wi1a, khi wi1b đến. Nói cách khác, tôi chỉ muốn nói, "Tôi muốn gửi một mục công việc cho queue1, được xử lý serially với các mục khác mà tôi đã gửi cho queue1, nhưng có thể song song với các mục công việc được gửi tới các hàng đợi khác".

Tôi hy vọng mô tả này có ý nghĩa. Nếu không xin vui lòng đặt câu hỏi trong phần bình luận và tôi sẽ cập nhật câu hỏi này cho phù hợp.

Cảm ơn bạn đã đọc.

Cập nhật:

Để tóm tắt "sai lầm" giải pháp cho đến nay, đây là những giải pháp từ phần câu trả lời mà tôi không thể sử dụng và lý do (s) tại sao tôi không thể sử dụng chúng:

TPL các nhiệm vụ yêu cầu xác định tác vụ tiền định trước cho ContinueWith(). Tôi không muốn duy trì kiến ​​thức về nhiệm vụ tiền nhiệm của mỗi hàng đợi khi gửi một nhiệm vụ mới.

TDF ActionBlocks trông đầy hứa hẹn, nhưng có vẻ như các mục được đăng lên một ActionBlock được xử lý song song. Tôi cần cho các mục cho một hàng đợi cụ thể được xử lý serially.

Cập nhật 2:

RE: ActionBlocks

Nó sẽ xuất hiện mà thiết lập các tùy chọn MaxDegreeOfParallelism một ngăn xử lý song song của hạng mục công trình nộp cho một đơn ActionBlock. Do đó có vẻ như có một ActionBlock mỗi hàng đợi giải quyết vấn đề của tôi với bất lợi duy nhất là điều này đòi hỏi việc cài đặt và triển khai thư viện TDF của Microsoft và tôi đã hy vọng cho một giải pháp .NET 4.0 thuần túy. Cho đến nay, đây là ứng cử viên chấp nhận câu trả lời, trừ khi ai đó có thể tìm ra cách để làm điều này với một giải pháp .NET 4.0 thuần túy không làm thoái hóa chuỗi công việc trên mỗi hàng đợi (mà tôi đã sử dụng).

Answer 1

Tôi hiểu rằng bạn có nhiều hàng đợi và không muốn liên kết chuỗi. Bạn có thể có một ActionBlock mỗi hàng đợi. ActionBlock tự động hóa hầu hết những gì bạn cần: Nó xử lý các mục công việc serially, và chỉ bắt đầu một Task khi công việc đang chờ xử lý. Khi không có tác phẩm nào đang chờ xử lý, không có Tác vụ/Chủ đề nào bị chặn.

Answer 2

Cách tốt nhất là sử dụng Task Parallel Library (TPL) và Continuations. Việc tiếp tục không chỉ cho phép bạn tạo luồng công việc mà còn xử lý các ngoại lệ của bạn. Đây là great introduction cho TPL. Nhưng để cung cấp cho bạn một số ý tưởng ...

Bạn có thể bắt đầu một nhiệm vụ TPL sử dụng

Task task = Task.Factory.StartNew(() => 
{ 
    // Do some work here... 
}); 

Bây giờ để bắt đầu một nhiệm vụ thứ hai khi một kết thúc nhiệm vụ tiền đề (do lỗi hoặc thành công) bạn có thể sử dụng phương pháp ContinueWith

Task task1 = Task.Factory.StartNew(() => Console.WriteLine("Antecedant Task")); 
Task task2 = task1.ContinueWith(antTask => Console.WriteLine("Continuation...")); 

Vì vậy, khi ngay sau khi task1 hoàn thành, không thành công hoặc bị hủy task2 'kích hoạt' và bắt đầu chạy. Lưu ý rằng nếu task1 đã hoàn tất trước khi đến dòng mã thứ hai task2 sẽ được lập lịch để thực thi ngay lập tức. Đối số antTask được truyền cho lambda thứ hai là tham chiếu đến tác vụ tiền định trước. Xem this link cho các ví dụ chi tiết hơn ...

Bạn cũng có thể vượt qua continuations kết quả từ nhiệm vụ tiền đề

Task.Factory.StartNew<int>(() => 1) 
    .ContinueWith(antTask => antTask.Result * 4) 
    .ContinueWith(antTask => antTask.Result * 4) 
    .ContinueWith(antTask =>Console.WriteLine(antTask.Result * 4)); // Prints 64. 

Note. Hãy chắc chắn để đọc lên trên xử lý ngoại lệ trong liên kết đầu tiên được cung cấp vì điều này có thể dẫn một người mới đến TPL lạc lối.

Một điều cuối cùng cần xem xét đặc biệt cho những gì bạn muốn là nhiệm vụ của trẻ. Nhiệm vụ con là những nhiệm vụ được tạo thành AttachedToParent. Trong trường hợp này, việc tiếp tục sẽ không chạy cho đến khi tất cả các công việc con đã hoàn thành

TaskCreationOptions atp = TaskCreationOptions.AttachedToParent; 
Task.Factory.StartNew(() => 
{ 
    Task.Factory.StartNew(() => { SomeMethod() }, atp); 
    Task.Factory.StartNew(() => { SomeOtherMethod() }, atp); 
}).ContinueWith(cont => { Console.WriteLine("Finished!") }); 

Tôi hy vọng điều này sẽ hữu ích.

Chỉnh sửa: Bạn đã xem ConcurrentCollections cụ thể là BlockngCollection<T>. Vì vậy, trong trường hợp của bạn, bạn có thể sử dụng một cái gì đó như

public class TaskQueue : IDisposable 
{ 
    BlockingCollection<Action> taskX = new BlockingCollection<Action>(); 

    public TaskQueue(int taskCount) 
    { 
     // Create and start new Task for each consumer. 
     for (int i = 0; i < taskCount; i++) 
      Task.Factory.StartNew(Consumer); 
    } 

    public void Dispose() { taskX.CompleteAdding(); } 

    public void EnqueueTask (Action action) { taskX.Add(Action); } 

    void Consumer() 
    { 
     // This seq. that we are enumerating will BLOCK when no elements 
     // are avalible and will end when CompleteAdding is called. 
     foreach (Action action in taskX.GetConsumingEnumerable()) 
      action(); // Perform your task. 
    } 
} 

Answer 3

Dường như thiết kế bạn đã có tốt và hoạt động. Chuỗi công việc của bạn (một trên mỗi hàng đợi) là dài hạn vì vậy nếu bạn muốn sử dụng Tác vụ thay vào đó, hãy chỉ định TaskCreationOptions.LongRunning để bạn nhận được chuỗi công việc chuyên dụng.

Nhưng thực sự không cần sử dụng ThreadPool tại đây. Nó không mang lại nhiều lợi ích cho công việc lâu dài.

Answer 4

Một giải pháp .NET 4.0 dựa trên TPL là có thể, trong khi ẩn đi thực tế là nó cần lưu trữ nhiệm vụ cha mẹ ở đâu đó. Ví dụ:

class QueuePool 
{ 
    private readonly Task[] _queues; 

    public QueuePool(int queueCount) 
    { _queues = new Task[queueCount]; } 

    public void Enqueue(int queueIndex, Action action) 
    { 
     lock (_queues) 
     { 
      var parent = _queue[queueIndex]; 
      if (parent == null) 
       _queues[queueIndex] = Task.Factory.StartNew(action); 
      else 
       _queues[queueIndex] = parent.ContinueWith(_ => action()); 
     } 
    } 
} 

Điều này đang sử dụng một khóa duy nhất cho tất cả các hàng đợi để minh họa ý tưởng. Tuy nhiên, trong mã sản xuất, tôi sẽ sử dụng khóa trên mỗi hàng đợi để giảm ganh đua.