Wie kann ich von einem Router-Socket in ZeroMQ oder NetMQ sowohl senden als auch empfangen?

Question

Ich habe einen Händler < -> Router-Setup in NetMQ v4 wo ich kann asynchron senden und empfangen Nachrichten in jeder Richtung ohne Probleme.

Ich möchte nun, dass in eine Abstraktion formalisieren, wo der Server (Router) für jede eingehende Nachricht hört, aber es müssen auch Sendungen Nachfrage Nachrichten an jedem der angeschlossenen Clients auf (Händler).

Ich versuche, Pub < -> Sub-Sockets zu vermeiden, da ich die Abonnenten auch Nachrichten an den Server senden müssen. Das nächste Muster zu dem, was ich versuche zu erreichen, ist eine WebSocket-Client-Server-Kommunikation.

using (var server = new RouterSocket("@tcp://*:80")) 
{ 
    var addresses = new HashSet<string>(); 
    while (true) 
    { 
     var msg = server.ReceiveMultipartMessage(); 

     var address = Encoding.UTF8.GetString(msg[0].Buffer); 
     var payload = Encoding.UTF8.GetString(msg[2].Buffer); 
     Console.WriteLine("[Server] - Client: {0} Says: {1}", address, payload); 

     var contains = addresses.Contains(address); 
     if (!contains) { addresses.Add(address); }    

     msg.Clear(); 
     msg.Append(address); 
     msg.AppendEmptyFrame(); 
     msg.Append("Reply for: " + address); 
     server.SendMultipartMessage(msg); 
    } 
} 

Nun da die Steckdosen Thread-sicher nicht, ich auf der Suche nach einem Weg, steckte Nachrichten senden (aus Richtung:

Der erste Teil der Client-Nachrichten zu hören sind, in so etwas wie getan ein anderer Thread auf Anfrage) an alle Kunden.

Ich kann wahrscheinlich verwenden Sie die TryReceiveMultipartMessage Methode in der Schleife statt mit einem festgelegten Timeout, nach dem ich eine Warteschlange für alle Broadcast-Nachrichten überprüfen und dann durchlaufen jeden Client Senden solcher Nachricht. Etwas wie:

using (var server = new RouterSocket("@tcp://*:80")) 
{ 
    var addresses = new HashSet<string>(); 

    var msg = new NetMQMessage(); 
    while (true) 
    { 
     var clientHasMsg = server.TryReceiveMultipartMessage(TimeSpan.FromSeconds(1), ref msg); 
     if (!clientHasMsg) 
     { 
      // Check any incoming broacast then loop through all the clients 
      // sending each the brodcast msg 
      var broadMsg = new NetMQMessage(); 
      foreach (var item in addresses) 
      { 
       broadMsg.Append(item); 
       broadMsg.AppendEmptyFrame(); 
       broadMsg.Append("This is a broadcast"); 
       server.SendMultipartMessage(broadMsg); 
       broadMsg.Clear(); 
      } 

      // Go back into the loop waiting for client messages 
      continue; 
     } 

     var address = Encoding.UTF8.GetString(msg[0].Buffer); 
     var payload = Encoding.UTF8.GetString(msg[2].Buffer); 
     Console.WriteLine("[Server] - Client: {0} Says: {1}", address, payload); 

     var contains = addresses.Contains(address); 
     if (!contains) { addresses.Add(address); } 

     msg.Clear(); 
     msg.Append(address); 
     msg.AppendEmptyFrame(); 
     msg.Append("Reply for: " + address); 
     server.SendMultipartMessage(msg); 
    } 
} 

die irgendwie fühlt sich nicht richtig in erster Linie auf:

• Welcher Wert für die Zeitüberschreitung ist ein guter Wert? 1 Sekunde, 100 ms usw .;
• Ist dies die effizienteste/leistungsstärkste Lösung, da dieses Programm verwendet wird, um 100k + Clients zu verbinden, die jeweils Tausende von Nachrichten pro Sekunde senden.

Irgendwelche Hinweise darauf, was der beste Ansatz dazu ist, wird sehr geschätzt.

Answer 1

Sie können Netmqqueue, es Multi-Produzent einzigen Verbraucher Warteschlange verwenden. Sie können es zu NetMQPoller hinzufügen und aus mehreren Threads ohne Sperre in die Warteschlange einreihen.

Answer 2

Ich denke, PUB/SUB ist ein richtiger Ansatz für Ihre Anforderungen von 100k + Clients. Dies bedeutet jedoch nicht, dass Sie nicht zurück zum Server kommunizieren können: Verwenden Sie den DEALER/ROUTER. Warum denken Sie, dass diese Lösung nicht akzeptabel ist?