ZeroMQ/0MQプッシュ/プルメモリとルーティングの問題

Question

私はZeroMQでしばらくプレイして、私が思いついた質問/問題をいくつか持っています。 ZeroMQの貢献者が誰でもチャイムすることができれば、またはライブラリを使用しているか現在使用している人なら誰でも感謝します。

*私は1台のルータ/フォワーダと2種類のクライアント（c1、c2）を持っているとします。私はルーティングデバイスを介してクライアント1からクライアント2にメッセージをプッシュしたい。ルータは、クライアント（ここではclient1）からメッセージを引き出し、それらを任意のサブスクライブされたクライアント（ここではclient2）にパブリッシュします。しかし、私は現在、そのようなメッセージを適切なクライアントにルーティングする唯一の方法は、pub/subを使用することです。a）メッセージボディとともにroutingToタグを送信することによって実行時にルーティングする方法を決定したい、b）push /私はc1とc2を押して正確に1つのポートに接続し、サブスクライブするために1つのポートに接続したいと思っています。私は何とかルータ側でpub/subを使用する必要がないために変更を加えることができますか、pub/subはメッセージが転送されるはずのルーティング側で知っていてもクライアントにルーティングする唯一の方法ですか？キューサイズが私が望んでいないhwmを超えると、そのpub/sub dropsメッセージを読んでいます。また、私は返答が必要ないので、不必要なオーバーヘッドを追加するので、リクエスト/返信のパターンを実装したくありません。コードの下に実行した後

* （プッシュ/プル - >パブ/サブ）は、すべてのメッセージを送信し、すべてのメッセージはまだメッセージを押し出す、明らかに巨大なメモリ・フットプリントがあります表示するクライアントを受け取った確認を受けているとプッシュソケットのキューにまだ膨大な量のメッセージが残っています。それはなぜですか、それを修正するために私は何ができますか？ここで

は私のコードです：

ROUTER：

class Program 
{ 
    static void Main(string[] args) 
    { 
     using (var context = new Context(1)) 
     { 
      using (Socket socketIn = context.Socket(SocketType.PULL), socketOut = context.Socket(SocketType.XPUB)) 
      { 
       socketIn.HWM = 10000; 
       socketOut.Bind("tcp://*:5560"); //forwards on this port 
       socketIn.Bind("tcp://*:5559"); //listens on this port 

       Console.WriteLine("Router started and running..."); 

       while (true) 
       { 
        //Receive Message 
        byte[] address = socketIn.Recv(); 
        byte[] body = socketIn.Recv(); 

        //Forward Message 
        socketOut.SendMore(address); 
        socketOut.Send(body); 
       } 
      } 
     } 
    } 
} 

CLIENT1：

class Program 
{ 
    static void Main(string[] args) 
    { 
     using (var context = new Context(1)) 
     { 
      using (Socket socketIn = context.Socket(SocketType.SUB), socketOut= context.Socket(SocketType.PUSH)) 
      { 
       byte[] iAM = Encoding.Unicode.GetBytes("Client1"); 
       byte[] youAre = Encoding.Unicode.GetBytes("Client2"); 
       byte[] msgBody = new byte[16]; 

       socketOut.HWM = 10000; 
       socketOut.Connect("tcp://localhost:5559"); 
       socketIn.Connect("tcp://localhost:5560"); 
       socketIn.Subscribe(iAM); 

       Console.WriteLine("Press key to kick off Test Client1 Sending Routine"); 
       Console.ReadLine(); 

       for (int counter = 1; counter <= 10000000; counter++) 
       { 
        //Send Message 
        socketOut.SendMore(youAre); 
        socketOut.Send(msgBody); 
       } 

       Console.WriteLine("Client1: Finished Sending"); 
       Console.ReadLine(); 
      } 
     } 
    } 
} 

がCLIENT2：

class Program 
{ 
    public static int msgCounter; 

    static void Main(string[] args) 
    { 
     msgCounter = 0; 

     using (var context = new Context(1)) 
     { 
      using (Socket socketIn = context.Socket(SocketType.SUB), socketOut = context.Socket(SocketType.PUSH)) 
      { 
       byte[] iAM = Encoding.Unicode.GetBytes("Client2"); 

       socketOut.Connect("tcp://localhost:5559"); 
       socketIn.Connect("tcp://localhost:5560"); 
       socketIn.Subscribe(iAM); 

       Console.WriteLine("Client2: Started Listening"); 

       //Receive First Message 
       byte[] address = socketIn.Recv(); 
       byte[] body = socketIn.Recv(); 
       msgCounter += 1; 

       Console.WriteLine("Received first message"); 

       Stopwatch watch = new Stopwatch(); 
       watch.Start(); 

       while (msgCounter < 10000000) 
       { 
        //Receive Message 
        address = socketIn.Recv(); 
        body = socketIn.Recv(); 
        msgCounter += 1; 
       } 

       watch.Stop(); 
       Console.WriteLine("Elapsed Time: " + watch.ElapsedMilliseconds + "ms"); 
       Console.ReadLine(); 
      } 
     } 
    } 
} 

Answer 1

私はあなたのアーキテクチャがかもしれないことを示唆するつもりですここから少し離れていてください。

1）正確に1つのPUSHと1つのPULLが必要な場合は、デバイスを真ん中から取り外します。ノードを追加するたびにプロデューサを更新する必要がないように、複数のコンシューマを管理するために、デバイスがアーキテクチャに明示的に追加されます。いつ/複数のコンシューマーやプロデューサーが必要なのかを知るには、デバイス上の各ノードに接続する必要があります。この場合、デバイスがあなたのソリューションを過度に複雑にしているように見えます。

2）「ルート」タグを持つというアイデアは、本当に私の心を揺さぶる。おそらく、他の統合オプションよりもメッセージングを選択する最大の理由は、どちらの側も（ブローカレス設計の場合にメッセージを送信する場所以外の）他の側について何も知る必要がないように、プロデューサとコンシューマを切り離すことです。ロジックに直接ルーティング情報を追加すると、これが破られます。

オーバーヘッドに関しては、私はこれを経験したことがありません。しかし、私はZeroMQのために.Netドライバを使ったことは一度もありませんでした。無神経な推測は.Netドライバ自体を見ることです。