Wie .Net Task Parallel Library Datenfluss senden Nachricht an viele Clients mit nicht nur einem Cache?

Question

Es gibt einen Server-Task, der TPL Dataflow verwendet, um Nachrichten an viele Clients zu senden.

• Client stellt eine Verbindung mit dem Server zufällig
• Client kann Nachricht an Server und Client senden Nachricht vom Server empfangen kann

Server eine BufferBlock<string> verwenden Nachricht an Client zu senden, wenn der Client mit dem Server verbindet Es empfängt eine Nachricht von diesem BufferBlock<string>.

Aber diese BufferBlock<string> kann nur eine Nachricht zwischenspeichern, und der Client kann nicht mehr als eine Nachricht vom Server anfordern, und der Client kann keine Bedingung für die Auswahl der zu empfangenden Nachricht festlegen.

Ich möchte einen Blocktyp, der mehrere Nachrichten zwischenspeichern kann; und der Client kann nicht nur eine Nachricht von diesem Blocktyp lesen, und der Client kann auswählen, welche Nachricht er empfangen soll;

Ich versuchte andere TPL Dataflow Blocktypen, aber niemand hat solche Fähigkeit, ist TPL Dataflow Block nicht für solche Anforderungen geeignet?

Die Bedingung ist einfach, jede Nachricht hat einen Zeitstempel, der Client sendet nur einen Zeitstempel an den Server, dann gibt der Server Nachrichten zurück, die nach dem Zeitstempel gesendet wurden.

using System; 
using System.Web; 
using System.Net; 
using System.Threading.Tasks; 
using System.Text; 
using SimpleJSON; 
using System.Collections.Generic; 
using System.Threading.Tasks.Dataflow; 

namespace TestHttp 
{ 
    public class HttpServer 
    { 
     private HttpListener httpListener; 
     public Task task; 
     public HttpServer() 
     { 
      var ta = Task.Factory.StartNew(RunHttp); 
      task = ta.Result; 
     } 

     private async Task RunHttp() 
     { 
      var httpPort = 9090; 
      httpListener = new HttpListener(); 
      httpListener.Prefixes.Add("http://*:"+httpPort+"/"); 
      httpListener.Start(); 
      while (httpListener.IsListening) 
      { 
       var context = await httpListener.GetContextAsync(); 
       var req = context.Request; 
       Handle(context, req); 
      } 

      httpListener.Stop(); 
      httpListener.Close(); 
     } 

     private async Task Handle(HttpListenerContext context, HttpListenerRequest req) 
     { 
      Console.WriteLine(req.RawUrl); 

      var resp = await HandleGet(req); 
      var buf = Encoding.UTF8.GetBytes(resp); 
      context.Response.AddHeader("Content-Encoding", "utf-8"); 
      context.Response.ContentEncoding = Encoding.UTF8; 
      context.Response.ContentLength64 = buf.Length; 
      try 
      { 
       context.Response.OutputStream.Write(buf, 0, buf.Length); 
      } 
      catch (Exception exp) 
      { 
       Console.WriteLine(exp.ToString()); 
      } 
      finally 
      { 
       context.Response.OutputStream.Close(); 
      } 

     } 

     private BufferBlock<string> messages = new BufferBlock<string>(); 

     private async Task<string> HandleGet(HttpListenerRequest req) 
     { 
      var r = req.RawUrl.Split('?'); 
      if (r[0] == "/send") 
      { 
       await messages.SendAsync(r[1]); 
       return "Suc"; 
      } 
      else if(r[0] == "/receive"){ 
       var timestamp = Convert.ToInt32(r[1]); 
       var ret = await messages.ReceiveAsync(); 
       return ret; 
      } 
      //Console.WriteLine(r[0]); 
      return "Error"; 

     } 
    } 
} 

Answer 1

Warum sagen Sie, dass die BufferBlock nur einen Wert enthalten kann? Es ist nicht wahr, es kann so viele Nachrichten enthalten, wie Sie möchten, basierend auf den Blockerstellungsoptionen und speziell BoundedCapacity Option. Der Standardwert für diesen Parameter ist -1, was für unbegrenzte Kapazität steht.

In dem Moment, in dem der Client eine Verbindung herstellt, können Sie alle Nachrichten, die nach dem Zeitstempel gefiltert sind, einfach abrufen und an den Client zurücksenden. Dies führt möglicherweise dazu, dass die Signatur des Ergebniswerts für die Anforderung des Clients geändert wird, da Sie den Parameter TimeStamp angeben und die Möglichkeit bereitstellen müssen, die List<T> der Nachrichten und nicht die einzige zurückzugeben. Ohne Code können wir für dieses Problem nicht mehr sagen.

Answer 2

Ich denke, TPL-Datenflussblock kann diese Anforderung nicht erfüllen.

Ich benutze nur eine Liste, um alle Nachrichten zu halten List<Message> messages;

struct Message { 
    int id; 
    string msg; 
} 

und ich brauche eine Sperre oder Schauspieler Modell ähnlich Mailbox zu verwenden, mein Mandant ist die Anforderung für die List<Message> zu behandeln.