C# .NET - Messages de tampon avec temporisateur

Question

J'ai besoin de mettre en œuvre un système de tamponnage de message qui est également basé sur la temporisation. Ce que j'ai besoin de faire est de stocker des instances de ma classe, puis de les envoyer quand j'atteins 100 instances ou quand 1 minute s'est écoulée.

En gros:

List<Message> messages; 

public void GotNewMessage(Message msg) 
{ 
    messages.add(msg); 

    if (messages.count() == 100 || timer.elapsed(1 minute)) 
    { 
     SendMessages(messages); 
     messages.clear() 
    } 
} 

Je ne peux pas sembler comprendre comment implémenter cela sans une utilisation excessive de serrures qui ralentiront considérablement le processus. Est-ce que quelqu'un sait d'un bon moyen de mettre en œuvre un tel système? Merci d'avance.

Answer 1

Il existe une bibliothèque fantastique pour ce genre d'exigences (combiner le temps avec des séquences), ce sont les extensions réactives. Voir https://github.com/Reactive-Extensions/Rx.NET

Vous pouvez écrire quelque chose comme

void Main() 
{ 
    messages 
     .Buffer(TimeSpan.FromMinutes(1), 100) // Buffer until 100 items or 1 minute has elapsed, whatever comes first. 
     .Subscribe(msgs => SendMessages(msgs));  
} 

Subject<Message> messages = new Subject<Message>(); 

public void GotNewMessage(Message msg) 
{ 
    messages.OnNext(msg); 
} 

Note: ce n'est pas prêt pour la production, mais il montre la base de la façon de le faire. En fonction de l'endroit où vous avez trouvé les messages à partir de là, il existe de meilleurs moyens de créer un observable auquel vous pouvez vous abonner.

Plus de références:

• http://www.introtorx.com/
• https://msdn.microsoft.com/en-us/library/hh242985(v=vs.103).aspx

Si votre message est reçu à l'aide d'un événement, vous pouvez lier l'événement à un flux RX, voir https://msdn.microsoft.com/en-us/library/hh242978(v=vs.103).aspx et https://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.fromeventpattern(v=vs.103).aspx

Answer 2

D'abord, vous devriez envisager d'utiliser un ConcurrentQueue <> insted d'une Liste <>. ConcurrentQueue <> est complètement sécurisé et ne nécessite aucun verrou supplémentaire. Avec cela, vous vous êtes déjà épargné un verrou pour la file d'attente des messages. Interlocked fournit l'atomicité, quand elle n'est pas disponible.

Selon le C# language specification, lit indépendant/écriture sont atomiques (mais seulement pour certains types de données et à long est pas toujours atomique - c'est pourquoi je changeai le DateTime.Now.Ticks pour obtenir un int32 sans perdre de bits qui influencera la durée écoulée time) et read-modify-write (par exemple ++ i) n'est jamais atomique.

Le décalage (par exemple, < <) est indépendant et ne nécessite aucun verrouillage supplémentaire.

private ConcurrentQueue<Message> Queue = new ConcurrentQueue<Message>(); 
private int QueueSize = 0; 
private int LastSend = (int)(DateTime.Now.Ticks >> 23); 
private int LastMessage = (int)(DateTime.Now.Ticks >> 23); 

public void GotNewMessage(Message Message) 
{ 
    Queue.Enqueue(Message); 

    Interlocked.Increment(ref QueueSize); 
    Interlocked.Exchange(ref LastMessage, (int)(DateTime.Now.Ticks >> 23)); 

    if (Interlocked.CompareExchange(ref QueueSize, 0, 100) >= 100 || 
     LastMessage - LastSend >= 60) 
    { 
     Message Dummy; 
     while (!Queue.IsEmpty) 
      if (Queue.TryDequeue(out Dummy)) 
       SendMessage(Dummy); 

     Interlocked.Exchange(ref LastSend, (int)(DateTime.Now.Ticks >> 23)); 
    } 
} 

public void SendMessage(Message Message) 
{ 
    // ... 
} 

Edit: Il peut se produire, qui sont envoyés plus de 100 messages. Si vous souhaitez envoyer uniquement 100 messages, vous pouvez implémenter une autre incrémentation atomique dans le cycle.