Làm thế nào tôi có thể sửa đổi một bộ sưu tập hàng đợi trong một vòng lặp?

Question

Tôi có một kịch bản mà tôi cần phải xóa một mục cho hàng đợi ngay sau khi được xử lý. Tôi hiểu rằng tôi không thể loại bỏ một mục từ một bộ sưu tập trong khi trong vòng lặp nhưng đã tự hỏi nếu có điều gì có thể được thực hiện với sự Enumerator vv ...

Đây chỉ là một ví dụ cơ bản ném một lỗi "Bộ sưu tập đã được sửa đổi sau khi điều tra đã được khởi tạo. "

Mọi đề xuất? Cảm ơn rất nhiều!!!

Mã là như sau:

 class Program 
     { 
      static void Main() 
      { 

       Queue<Order> queueList = GetQueueList(); 

       foreach (Order orderItem in queueList) 
       { 
        Save(orderItem); 
        Console.WriteLine("Id :{0} Name {1} ", orderItem.Id, orderItem.Name); 
        queueList.Dequeue(); 
       } 
       Console.Read(); 

      } 

      private static void Save(Order orderItem) 
      { 
       //we are pretending to save or do something. 
      } 

      private static Queue<Order>GetQueueList() 
      { 
       Queue<Order> orderQueue = new Queue<Order>(); 
       orderQueue.Enqueue(new Order { Id = 1, Name = "Order 1" }); 
       orderQueue.Enqueue(new Order { Id = 1, Name = "Order 2" }); 
       orderQueue.Enqueue(new Order { Id = 2, Name = "Order 3" }); 
       orderQueue.Enqueue(new Order { Id = 3, Name = "Order 4" }); 
       orderQueue.Enqueue(new Order { Id = 4, Name = "Order 5" }); 
       return orderQueue; 
      } 
     } 

     public class Order 
     { 
      public int Id { get; set; } 
      public string Name { get; set; } 
     } 

Answer 1

Thay đổi foreach của bạn để:

while (queueList.Count > 0) 
{ 
    Order orderItem = queueList.Dequeue(); 
    Save(orderItem); 
    Console.WriteLine("Id :{0} Name {1} ", orderItem.Id, orderItem.Name); 
} 

Edit:

Để tái xử lý nếu tiết kiệm không làm điều gì đó như:

while (queueList.Count > 0) 
{ 
    Order orderItem = queueList.Dequeue(); 

    if (!Save(orderItem)) 
    { 
     queueList.Enqueue(orderItem); // Reprocess the failed save, probably want more logic to prevent infinite loop 
    } 
    else 
    { 
     Console.WriteLine("Successfully saved: {0} Name {1} ", orderItem.Id, orderItem.Name); 
    } 
} 

Chỉnh sửa:

John K đề cập đến an toàn chủ đề là mối quan tâm hợp lệ nếu bạn có nhiều chủ đề truy cập cùng một Queue. Xem http://ccutilities.codeplex.com/SourceControl/changeset/view/40529#678487 cho lớp ThreadSafeQueue bao gồm các vấn đề an toàn chủ đề đơn giản.

---

Edit: Đây là ví dụ an toàn thread Tôi cứ chỉ mọi người :-)

Dưới đây là một ví dụ về vấn đề an toàn chủ đề đề cập. Như được hiển thị, mặc định Queue có thể "bỏ lỡ" các mục trong khi vẫn giảm số lượng.

Cập nhật: Để thể hiện tốt hơn vấn đề. Tôi không bao giờ thêm một mục rỗng vào Queue nhưng tiêu chuẩn Queue.Dequeue() trả về một số giá trị null. Điều này một mình sẽ là tốt nhưng trong khi làm như vậy một mục hợp lệ được lấy ra từ bộ sưu tập nội bộ và giảm Count. Đó là một giả định an toàn, trong ví dụ cụ thể này, rằng mỗi mục null được trả lại từ hoạt động Queue.Dequeue() đại diện cho một mục hợp lệ chưa bao giờ được xử lý.

using System; 
using System.Collections.Generic; 
using System.Threading; 

namespace SO_ThreadSafeQueue 
{ 
    class Program 
    { 
     static int _QueueExceptions; 
     static int _QueueNull; 
     static int _QueueProcessed; 

     static int _ThreadSafeQueueExceptions; 
     static int _ThreadSafeQueueNull; 
     static int _ThreadSafeQueueProcessed; 

     static readonly Queue<Guid?> _Queue = new Queue<Guid?>(); 
     static readonly ThreadSafeQueue<Guid?> _ThreadSafeQueue = new ThreadSafeQueue<Guid?>(); 
     static readonly Random _Random = new Random(); 

     const int Expected = 10000000; 

     static void Main() 
     { 
      Console.Clear(); 
      Console.SetCursorPosition(0, 0); 
      Console.WriteLine("Creating queues..."); 

      for (int i = 0; i < Expected; i++) 
      { 
       Guid guid = Guid.NewGuid(); 
       _Queue.Enqueue(guid); 
       _ThreadSafeQueue.Enqueue(guid); 
      } 

      Console.SetCursorPosition(0, 0); 
      Console.WriteLine("Processing queues..."); 

      for (int i = 0; i < 100; i++) 
      { 
       ThreadPool.QueueUserWorkItem(ProcessQueue); 
       ThreadPool.QueueUserWorkItem(ProcessThreadSafeQueue); 
      } 

      int progress = 0; 

      while (_Queue.Count > 0 || _ThreadSafeQueue.Count > 0) 
      { 
       Console.SetCursorPosition(0, 0); 

       switch (progress) 
       { 
        case 0: 
         { 
          Console.WriteLine("Processing queues... |"); 
          progress = 1; 
          break; 
         } 
        case 1: 
         { 
          Console.WriteLine("Processing queues... /"); 
          progress = 2; 
          break; 
         } 
        case 2: 
         { 
          Console.WriteLine("Processing queues... -"); 
          progress = 3; 
          break; 
         } 
        case 3: 
         { 
          Console.WriteLine("Processing queues... \\"); 
          progress = 0; 
          break; 
         } 
       } 

       Thread.Sleep(200); 
      } 

      Console.SetCursorPosition(0, 0); 
      Console.WriteLine("Finished processing queues..."); 
      Console.WriteLine("\r\nQueue Count:   {0} Processed: {1, " + Expected.ToString().Length + "} Exceptions: {2,4} Null: {3}", _Queue.Count, _QueueProcessed, _QueueExceptions, _QueueNull); 
      Console.WriteLine("ThreadSafeQueue Count: {0} Processed: {1, " + Expected.ToString().Length + "} Exceptions: {2,4} Null: {3}", _ThreadSafeQueue.Count, _ThreadSafeQueueProcessed, _ThreadSafeQueueExceptions, _ThreadSafeQueueNull); 

      Console.WriteLine("\r\nPress any key..."); 
      Console.ReadKey(); 
     } 

     static void ProcessQueue(object nothing) 
     { 
      while (_Queue.Count > 0) 
      { 
       Guid? currentItem = null; 

       try 
       { 
        currentItem = _Queue.Dequeue(); 
       } 
       catch (Exception) 
       { 
        Interlocked.Increment(ref _QueueExceptions); 
       } 

       if (currentItem != null) 
       { 
        Interlocked.Increment(ref _QueueProcessed); 
       } 
       else 
       { 
        Interlocked.Increment(ref _QueueNull); 
       } 

       Thread.Sleep(_Random.Next(1, 10)); // Simulate different workload times 
      } 
     } 

     static void ProcessThreadSafeQueue(object nothing) 
     { 
      while (_ThreadSafeQueue.Count > 0) 
      { 
       Guid? currentItem = null; 

       try 
       { 
        currentItem = _ThreadSafeQueue.Dequeue(); 
       } 
       catch (Exception) 
       { 
        Interlocked.Increment(ref _ThreadSafeQueueExceptions); 
       } 

       if (currentItem != null) 
       { 
        Interlocked.Increment(ref _ThreadSafeQueueProcessed); 
       } 
       else 
       { 
        Interlocked.Increment(ref _ThreadSafeQueueNull); 
       } 

       Thread.Sleep(_Random.Next(1, 10)); // Simulate different workload times 
      } 
     } 

     /// <summary> 
     /// Represents a thread safe <see cref="Queue{T}"/> 
     /// </summary> 
     /// <typeparam name="T"></typeparam> 
     public class ThreadSafeQueue<T> : Queue<T> 
     { 
      #region Private Fields 
      private readonly object _LockObject = new object(); 
      #endregion 

      #region Public Properties 
      /// <summary> 
      /// Gets the number of elements contained in the <see cref="ThreadSafeQueue{T}"/> 
      /// </summary> 
      public new int Count 
      { 
       get 
       { 
        int returnValue; 

        lock (_LockObject) 
        { 
         returnValue = base.Count; 
        } 

        return returnValue; 
       } 
      } 
      #endregion 

      #region Public Methods 
      /// <summary> 
      /// Removes all objects from the <see cref="ThreadSafeQueue{T}"/> 
      /// </summary> 
      public new void Clear() 
      { 
       lock (_LockObject) 
       { 
        base.Clear(); 
       } 
      } 

      /// <summary> 
      /// Removes and returns the object at the beggining of the <see cref="ThreadSafeQueue{T}"/> 
      /// </summary> 
      /// <returns></returns> 
      public new T Dequeue() 
      { 
       T returnValue; 

       lock (_LockObject) 
       { 
        returnValue = base.Dequeue(); 
       } 

       return returnValue; 
      } 

      /// <summary> 
      /// Adds an object to the end of the <see cref="ThreadSafeQueue{T}"/> 
      /// </summary> 
      /// <param name="item">The object to add to the <see cref="ThreadSafeQueue{T}"/></param> 
      public new void Enqueue(T item) 
      { 
       lock (_LockObject) 
       { 
        base.Enqueue(item); 
       } 
      } 

      /// <summary> 
      /// Set the capacity to the actual number of elements in the <see cref="ThreadSafeQueue{T}"/>, if that number is less than 90 percent of current capactity. 
      /// </summary> 
      public new void TrimExcess() 
      { 
       lock (_LockObject) 
       { 
        base.TrimExcess(); 
       } 
      } 
      #endregion 
     } 

    } 
} 

Answer 2

Với tôi, có vẻ như bạn đang cố xử lý phần tử xếp hàng từng người một.

Làm thế nào về gói này trong while lặp và xử lý từng phần tử từ Dequeue, cho đến khi hàng đợi rỗng?

Answer 3

foreach như một cách hợp lý để lặp qua các hàng đợi khi bạn không loại bỏ các mục

Khi bạn muốn loại bỏ và các mặt hàng chế biến, các, cách thích hợp thread-safe chỉ là loại bỏ chúng cùng một thời gian và xử lý chúng sau khi chúng đã được gỡ bỏ.

Một cách là này

// the non-thread safe way 
// 
while (queueList.Count > 0) 
{ 
    Order orderItem = queueList.Dequeue(); 
    Save(orderItem); 
    Console.WriteLine("Id :{0} Name {1} ", orderItem.Id, orderItem.Name); 
} 

Có thể cho số lượng các mục trong hàng đợi để thay đổi giữa queueList.Count và queueList.Dequeue(), vì vậy được chủ đề an toàn, bạn phải chỉ cần sử dụng Dequeue, nhưng Dequeue sẽ ném khi hàng đợi trống, vì vậy bạn phải sử dụng một trình xử lý ngoại lệ.

// the thread safe way. 
// 
while (true) 
{ 
    Order orderItem = NULL; 
    try { orderItem = queueList.Dequeue(); } catch { break; } 
    if (null != OrderItem) 
    { 
     Save(orderItem); 
     Console.WriteLine("Id :{0} Name {1} ", orderItem.Id, orderItem.Name); 
    } 
}