Parallel.ForEach wielokrotne powtarzanie elementów w kolekcji

Question

Mam Parallel.ForEach działający wewnątrz zadania. Iteruje on nad zbiorem adresów e-mail i wysyła wiadomość MailMessage do kolejki SMTP, po jej wysłaniu aktualizuje tabelę w bazie danych z wynikiem.

Widzę w DB, że wysyła wiadomość MailMessage do kolejki wiele razy, czasami nawet do 6 razy. Oto mój uproszczony kod, czy ktoś może polecić lepsze podejście?

Na kliknięcia przycisku, tworzę nowe zadanie ...

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."); 
     } 

Stwarza to zadanie, które w zasadzie dostaje ConcurrentBag abonentów (niestandardowe klasy), iteracje nad kolekcji i wysyła wiadomość .. .

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 
} 

ja skłonni uwierzyć, że ConcurrentBag są thread-safe, więc nie jestem pewien, dlaczego byłoby iteracji nad nieco w kolekcji wiele razy. Spośród tysiąca w kolejce pojawią się co najmniej 2 wiadomości dla 10% kolekcji.

Dzięki,

Greg.

Answer 1

Jestem przekonany, że ConcurrentBag są bezpieczne dla wątków, więc nie jestem pewien, dlaczego iteracja nad niektórymi w kolekcji wiele razy.

Twoje założenie tutaj jest prawdziwe. W rzeczywistości metoda ConcurrentBag<T> (GetEnumerator<T>) (w celu wyliczenia kolekcji) faktycznie tworzy w tym miejscu całą kopię kolekcji wewnętrznej, więc trwa iteracja nad kopią kolekcji.

Jeśli widzisz kolejkę jest nazywany wielokrotnie dla pojedynczego abonenta, to znaczy, że dodał, że abonent do ConcurrentBag<T> wiele razy, czy jest jakiś inny problem dzieje ...

---

Oddzielnie, użycie zadania tutaj naprawdę nie jest konieczne. Dodaje tylko narzut (tworzy dedykowany wątek w tym przypadku, a następnie natychmiast blokuje i czeka na niego). Byłoby o wiele lepiej po prostu przepisać to wywołać metodę, jak tak:

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

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

Tworzenie zadania tylko natychmiast czekać na nim (Task.WaitAny) nie jest pomocne w ogóle. Ponadto, zamiast używać Task.WaitAny(...), jeśli chcesz, aby zadanie było w jakimś innym celu, możesz po prostu zadzwonić pod numer broadcastResult = task.Result;, ponieważ będzie to blokowane do czasu zakończenia zadania.