Parallel.ForEach lặp lại các mục trong bộ sưu tập nhiều lần

Question

Tôi có một Parallel.ForEach đang chạy bên trong một Tác vụ. Nó lặp qua một tập hợp các địa chỉ email và gửi một MailMessage đến hàng đợi SMTP, một khi nó được gửi nó cập nhật một bảng trong DB với kết quả.

Tôi có thể thấy trong DB rằng nó gửi MailMessage đến hàng đợi nhiều lần, đôi khi lên đến 6 lần. Đây là mã đơn giản của tôi, bất cứ ai có thể giới thiệu một cách tiếp cận tốt hơn?

Mở nút nhấp chuột, tôi tạo ra một nhiệm vụ mới ...

CampaignManager.Broadcast.BroadcastService broadcastService = new CampaignManager.Broadcast.BroadcastService(); 

     var task = Task<CampaignManager.Broadcast.Results.Broadcast>.Factory.StartNew(() => { 
      return broadcastService.BroadcastCampaign(); 
     }, TaskCreationOptions.LongRunning); 

     Task.WaitAny(task); 

     if (task.Result != null) 
     { 
      Broadcast.Results.Broadcast broadcastResult = task.Result; 
      MessageBox.Show(broadcastResult.BroadcastSent.GroupName + " completed. " + broadcastResult.NumberSuccessful + " sent."); 
     } 

Điều này tạo ra một nhiệm vụ, mà về cơ bản được một ConcurrentBag thuê bao (lớp tùy chỉnh), lặp trên việc thu thập và gửi một thông điệp ..

public Results.Broadcast BroadcastCampaign() 
{ 
// Get ConcurrentBag of subscribers 
subscribers = broadcast.GetSubscribers(); 

// Iterate through subscribers and send them a message 
Parallel.ForEach(subscribers, subscriber => 
{ 
    // do some work, send to SMTP queue 

    // Add to DB log 
}); 

// return result 
} 

Tôi được cho rằng ConcurrentBag là an toàn chỉ, vì vậy tôi không chắc tại sao nó sẽ lặp lại một số trong bộ sưu tập nhiều lần. Trong số một nghìn, nó sẽ xếp ít nhất 2 tin nhắn cho 10% bộ sưu tập.

Xin cảm ơn,

Greg.

Answer 1

Tôi được cho rằng ConcurrentBag là an toàn chỉ, vì vậy tôi không chắc tại sao nó sẽ lặp lại một số trong bộ sưu tập nhiều lần.

Giả định của bạn ở đây là đúng. Trong thực tế, phương pháp của GetEnumerator<T> (để liệt kê bộ sưu tập) thực sự tạo một bản sao toàn bộ bộ sưu tập nội bộ tại thời điểm đó, vì vậy bạn đang lặp qua một bản sao của bộ sưu tập.

Nếu bạn thấy hàng đợi được gọi nhiều lần cho một thuê bao duy nhất, nó có nghĩa là bạn nói thêm rằng thuê bao đến ConcurrentBag<T> nhiều lần, hoặc có một số vấn đề khác xảy ra ...

---

Trên một ghi chú riêng biệt, việc sử dụng Tác vụ ở đây thực sự là không cần thiết. Nó chỉ cho biết thêm chi phí (tạo ra một thread chuyên dụng trong trường hợp này, sau đó ngay lập tức khối và chờ đợi trên nó). Sẽ tốt hơn nhiều để chỉ viết lại này để gọi phương thức của bạn, như vậy:

CampaignManager.Broadcast.BroadcastService broadcastService = new CampaignManager.Broadcast.BroadcastService(); 

Broadcast.Results.Broadcast broadcastResult = broadcastService.BroadcastCampaign(); 
MessageBox.Show(broadcastResult.BroadcastSent.GroupName + " completed. " + broadcastResult.NumberSuccessful + " sent."); 

Tạo một nhiệm vụ chỉ để chờ đợi vào nó ngay lập tức (Task.WaitAny) là không hữu ích gì cả. Ngoài ra, thay vì sử dụng Task.WaitAny(...), nếu bạn muốn nhiệm vụ cho một số mục đích khác, bạn chỉ có thể gọi broadcastResult = task.Result;, vì điều đó sẽ chặn cho đến khi tác vụ hoàn tất.