cổ điển mô hình sản xuất tiêu dùng sử dụng blockingcollection và nhiệm vụ net 4 TPL

Question

Xin vui lòng xem bên dưới mã giả

//Single or multiple Producers produce using below method 
    void Produce(object itemToQueue) 
    { 
     concurrentQueue.enqueue(itemToQueue); 
     consumerSignal.set; 
    } 

    //somewhere else we have started a consumer like this 
    //we have only one consumer 
    void StartConsumer() 
    { 
     while (!concurrentQueue.IsEmpty()) 
     { 
      if (concurrentQueue.TrydeQueue(out item)) 
      { 
       //long running processing of item 
      } 
     } 
     consumerSignal.WaitOne(); 
    } 

Làm thế nào để cổng mô hình này tôi đã sử dụng từ thời xa xưa để sử dụng nhiệm vụ tạo taskfactory và các tính năng truyền tín hiệu mới net 4. Nói cách khác, nếu ai đó viết mẫu này bằng cách sử dụng net 4 thì nó trông như thế nào? Mã giả là tốt. Iam đã sử dụng .net 4 concurrentQueue như bạn có thể thấy. Làm cách nào để sử dụng tác vụ và có thể sử dụng một số cơ chế báo hiệu mới hơn nếu có thể. cảm ơn

Giải pháp cho vấn đề của tôi bên dưới nhờ Jon/Dan. Ngọt. Không nhãn hiệu hoặc while (true) hoặc while (itemstoProcess) gõ vòng như ngày xưa

//Single or multiple Producers produce using below method 
void Produce(object itemToQueue) 
{ 
    blockingCollection.add(item); 
} 

//somewhere else we have started a consumer like this 
//this supports multiple consumers ! 
task(StartConsuming()).Start; 

void StartConsuming() 
{ 
    foreach (object item in blockingCollection.GetConsumingEnumerable()) 
    { 
       //long running processing of item 
    } 
} 

cancellations are handled using cancel tokens 

Answer 1

Bạn sẽ sử dụng BlockingCollection<T>. Có một ví dụ trong tài liệu.

Lớp học đó được thiết kế đặc biệt để làm cho điều này tầm thường.

Answer 2

Tôi đã sử dụng một mô hình trước đó tạo ra một loại 'theo yêu cầu' hàng đợi của người tiêu dùng (dựa trên tiêu thụ từ ConcurrentQueue):

 private void FireAndForget(Action fire) 
     { 
      _firedEvents.Enqueue(fire); 
      lock (_taskLock) 
      { 
       if (_launcherTask == null) 
       { 
        _launcherTask = new Task(LaunchEvents); 
        _launcherTask.ContinueWith(EventsComplete); 
        _launcherTask.Start(); 
       } 
      } 
     } 

     private void LaunchEvents() 
     { 
      Action nextEvent; 

      while (_firedEvents.TryDequeue(out nextEvent)) 
      { 
       if (_synchronized) 
       { 
        var syncEvent = nextEvent; 
        _mediator._syncContext.Send(state => syncEvent(), null); 
       } 
       else 
       { 
        nextEvent();       
       } 

       lock (_taskLock) 
       { 
        if (_firedEvents.Count == 0) 
        { 
         _launcherTask = null; 
         break; 
        } 
       } 
      } 
     } 

     private void EventsComplete(Task task) 
     { 
      if (task.IsFaulted && task.Exception != null) 
      { 
       // Do something with task Exception here 
      } 
     } 

Answer 3

Khối mã thứ hai của bạn trông đẹp hơn. Nhưng, bắt đầu một Task và sau đó ngay lập tức chờ đợi trên đó là vô nghĩa. Chỉ cần gọi Take và sau đó xử lý mục được trả lại trực tiếp trên chuỗi tiêu thụ. Đó là cách mô hình người tiêu dùng sản xuất được thực hiện. Nếu bạn nghĩ rằng việc xử lý các mục công việc là đủ sâu để đảm bảo cho người tiêu dùng nhiều hơn thì bằng mọi cách bắt đầu nhiều người tiêu dùng hơn. BlockingCollection là nhiều nhà sản xuất an toàn và nhiều người tiêu dùng.

public class YourCode 
{ 
    private BlockingCollection<object> queue = new BlockingCollection<object>(); 

    public YourCode() 
    { 
    var thread = new Thread(StartConsuming); 
    thread.IsBackground = true; 
    thread.Start(); 
    } 

    public void Produce(object item) 
    { 
    queue.Add(item); 
    } 

    private void StartConsuming() 
    { 
    while (true) 
    { 
     object item = queue.Take(); 
     // Add your code to process the item here. 
     // Do not start another task or thread. 
    } 
    } 
}