Chờ chặn thu (hàng đợi) để giảm kích thước trong C#

Question

tôi đang làm việc trên dự án với các công việc sau:

Phần thứ nhất:

• tổ chức sự kiện đến không đồng bộ và được xếp hàng đợi trong việc ngăn chặn hàng đợi, chúng tôi sẽ gọi đó là Q1
• chủ đề nhặt item tiếp theo có sẵn từ mà đợi
• mục kết thúc lên chạy {N} số nhiệm vụ song song
• Mỗi tác vụ xếp hàng kết quả của nó trên hàng đợi thứ hai, chúng tôi sẽ gọi số Q2.
• Khi xử lý mục hoàn thành, mục tiếp theo sẽ được đọc khỏi hàng đợi.

Phần thứ hai:

• chủ đề Một lần đọc tắt của Q2 một đối tượng tại một thời điểm và hoạt động trên kết quả

Vì vậy, vấn đề ở đây là, tất cả các mục trên hàng đợi đầu tiên kết thúc lên chạy một số lượng lớn các nhiệm vụ song song, và mỗi công việc xếp hàng đợi kết quả của nó. Hàng đợi thứ hai phải được xử lý serially, một mục tại một thời điểm mặc dù, và nó đang bị ngập lụt.

---

Câu hỏi của tôi

Tôi cần một cơ chế đó sẽ làm cho việc xử lý chủ đề Q1 chờ đợi cho đến khi số lượng các mục trong Q2 dưới một ngưỡng nhất định. Cách tốt nhất để đạt được điều này là gì? Có cách nào để có giải pháp hướng sự kiện thay vì giải pháp bỏ phiếu không?

Answer 1

Thay vì sử dụng Queue<T>, bạn có thể sử dụng BlockingCollection<T> cho quý 2. Nếu bạn đặt số BoundedCapacity, các cuộc gọi đến Q2.Add() sẽ chặn khi đạt đến dung lượng. Điều này sẽ tự động điều chỉnh quá trình xử lý Q1, vì nhiệm vụ N sẽ bắt đầu chặn nếu chúng không thể thêm vào hàng đợi cuối cùng.

Answer 2

Tôi giả định rằng bạn nhận dữ liệu trong các trận lụt không thường xuyên, với hạn hán kéo dài trong đó quý 2 có thể bắt kịp. Bạn đã xem xét chỉ đơn giản là hạn chế số lượng các chủ đề đồng thời sinh ra từ Q1 bằng cách sử dụng một hồ bơi thread giới hạn cho các nhiệm vụ này?

Tôi nghi ngờ bạn có thể hưởng lợi từ nhiều nhóm hồ sơ, nếu kích thước công việc có thể dễ dàng xác định khi đến. Bạn có thể có một số lượng nhỏ các luồng để xử lý các công việc lớn và một số lượng lớn các luồng đã sẵn sàng để xử lý các công việc nhỏ. Ngay cả hàng đợi trung gian thứ 3 cũng có thể mang lại lợi ích.

Answer 3

Sự cố của bạn có vẻ như một ví dụ hoàn hảo cần được giải quyết bằng thư viện TPL Dataflow.Nếu bạn sẵn sàng để thử nó, dưới đây là cách nó có thể làm việc (đây là ví dụ rất đơn giản tất nhiên):

TransformBlock<int, bool> transform = new TransformBlock<int, bool>(i => i > 5 ? true : false, 
      new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 }); 
ActionBlock<bool> resultBlock = new ActionBlock<bool>(b => Console.WriteLine("My result is : " + b), 
      new ExecutionDataflowBlockOptions { BoundedCapacity = 10 }); 
transform.LinkTo(resultBlock); 

Bạn đang xác định một biến đổi khối mà sẽ làm cho chuyển đổi của bạn (chỉ hoạt động này như Q1 của bạn), bạn có thể đặt mức độ song song của nó thành số lượng tác vụ bạn muốn sử dụng.

Sau đó, bạn đang tạo khối thứ hai (hoạt động như Q2), sẽ có BoundedCapacity thiết lập và xử lý mọi thư một cách đồng bộ, gọi hành động cho từng phần tử. Khối này có thể được thay thế bằng bất kỳ khối nào khác, chẳng hạn như BufferBlock sẽ cho phép bạn thăm dò ý kiến ​​từ nó theo yêu cầu.