pensamiento de cola de mensajes

Question

Hemos implementado una cola de mensajes utilizando C# Queue. Sabemos que solo tenemos un consumidor para sacar el mensaje disponible de la cola para su procesamiento con un bucle while. También sabemos que solo hay UN productor para poner un mensaje en la cola.

Tenemos una cola de mensajes en la parte superior de lock para asegurarnos de que el consumidor y el productor no puedan acceder a la cola simultáneamente.

Mi pregunta es ¿es lock necesario? Si el Queue aumenta su propiedad CountDESPUÉS, un artículo se agrega realmente y si el consumidor marca el Count antes de recuperar, el consumidor debe obtener un elemento de mensaje completo incluso si no tenemos ese lock. ¿Derecha? Por lo tanto, no enfrentaremos un problema de elemento de mensaje parcial. Entonces podemos deshacernos de ese lock?

Eso lock ralentizará el sistema y ocasionalmente podemos ver que el hilo de recuperación está bloqueado por un tiempo porque tenemos un productor muy pesado.

EDIT:

Desafortunadamente estamos utilizando .Net 3.5.

Answer 1

El problema real es que las estructuras de datos internas de la cola pueden estar mutando mientras se produce una cola o dequeue, y durante este período la estructura de datos está en un estado indeterminado desde el punto de vista de otro hilo.

Por ejemplo, un enqueue puede requerir estructuras internas de datos para expandirse, donde se crea una nueva estructura, y los elementos viejos se copian de una estructura antigua a una nueva estructura. Habrá muchos pasos involucrados en este proceso donde en cualquier momento sería peligroso que otro hilo acceda a la cola ya que las operaciones no están completas.

Por lo tanto, durante enqueue \ dequeue debe bloquear para que estas operaciones parezcan lógicamente atómicas.

Puede probar la nueva clase ConcurrentQueue en .Net 4.0 ya que posiblemente tenga mejores características de rendimiento, ya que utiliza un algoritmo sin bloqueo.

Answer 2

Ver ConcurrentQueue.

Answer 3

El bloqueo es necesario si está utilizando Queue<T>. Puede eliminar fácilmente esto reemplazando eso con ConcurrentQueue<T>; sin embargo, es posible que desee considerar la simplificación de este código reemplazándolo por un BlockingCollection<T>.

Esto le permitiría a su consumidor eliminar el bloqueo y durante la comprobación, y simplemente hacer un solo foreach en collection.GetConsumingEnumerable(). El productor puede eliminar el bloqueo y agregar elementos según sea necesario. También le permitiría ampliar el uso de múltiples productores, ya que mencionó que tiene un "productor muy pesado" en este momento.

Answer 4

Debe estar bloqueado; la clase no es threadsafe Si está utilizando el Queue en System.Collections, existe un hilo seguro Queue útil (System.Collections.Queue.Synchronized() devuelve un Queue).De lo contrario, asegúrese de utilizar el objeto proporcionado Queue<T>.SyncRoot para sincronizar:

using System.Collections.Generic; 
public static class Q_Example 
{ 
    private readonly Queue<int> q = new Queue<int>(); 
    public void Method1(int val) 
    { 
     lock(q.SyncRoot) 
     { 
      q.EnQueue(val); 
     } 
    } 

    public int Method2() 
    { 
     lock(q.SyncRoot) 
     { 
      return q.Dequeue(); 
     } 
    } 
} 

Answer 5

ConcurrentQueue está disponible incluso si está utilizando .NET 3.5. El Reactive Extensions incluye lo que solía ser las Extensiones Paralelas de .NET 3.5, el precursor de la Biblioteca de tareas paralelas que se incluye en .NET 4.0.

Answer 6

No, esto no funcionará constantemente ... ¿por qué?

Vamos a desmontar los dos métodos que vamos a estar llamando al mismo tiempo a partir de dos hilos (uno lee & un escritor):

public T Dequeue() 
{ 
    if (this._size == 0) 
    { 
     ThrowHelper.ThrowInvalidOperationException(ExceptionResource.InvalidOperation_EmptyQueue); 
    } 
    T local = this._array[this._head]; 
    this._array[this._head] = default(T); 
    this._head = (this._head + 1) % this._array.Length; 
    this._size--; 
    this._version++; 
    return local; 
} 

public void Enqueue(T item) 
{ 
    if (this._size == this._array.Length) 
    { 
     int capacity = (int) ((this._array.Length * 200L)/100L); 
     if (capacity < (this._array.Length + 4)) 
     { 
      capacity = this._array.Length + 4; 
     } 
     this.SetCapacity(capacity); 
    } 
    this._array[this._tail] = item; 
    this._tail = (this._tail + 1) % this._array.Length; 
    this._size++; 
    this._version++; 
} 

Teniendo en cuenta el código de seguridad de tres de las variables son seguros si (y sólo si) hay capacidad adecuada en la cola. los campos para _array, _head y _tail no están modificados o modificados en solo uno de los dos métodos anteriores.

El motivo por el que no se puede eliminar el bloqueo() es que ambos métodos modifican _size y _version. Aunque podría decirse que la colisión en _version podría ser ignorada, la colisión en _size causaría un comportamiento no deseado e impredecible.

Answer 7

BTW, las colas sin bloqueo para un solo lector y un solo escritor son bastante fáciles de escribir. Este es un ejemplo muy primitivo del concepto, pero hace el trabajo:

class LocklessQueue<T> 
{ 
    class Item 
    { 
     public Item Next; 
     bool _valid; 
     T _value; 
     public Item(bool valid, T value) 
     { 
      _valid = valid; 
      _value = value; 
      Next = null; 
     } 
     public bool IsValid { get { return _valid; } } 
     public T TakeValue() 
     { 
      T value = _value; 
      _valid = false; 
      _value = default(T); 
      return value; 
     } 
    } 

    Item _first; 
    Item _last; 

    public LocklessQueue() 
    { 
     _first = _last = new Item(false, default(T)); 
    } 

    public bool IsEmpty 
    { 
     get 
     { 
      while (!_first.IsValid && _first.Next != null) 
       _first = _first.Next; 
      return false == _first.IsValid; 
     } 
    } 

    public void Enqueue(T value) 
    { 
     Item i = new Item(true, value); 
     _last.Next = i; 
     _last = i; 
    } 

    public T Dequeue() 
    { 
     while (!_first.IsValid && _first.Next != null) 
      _first = _first.Next; 

     if (IsEmpty) 
      throw new InvalidOperationException();//queue is empty 

     return _first.TakeValue(); 
    } 
}