C＃winformのzeromq pub/subの例

Question

pub/subモデルでZeroMQ（clrzmq .net bindings（x86）via nuget）を使用するC＃Winformアプリケーションを作成しようとしています。

多くの検索の後で、コードがwhileステートメントを使用して新しいメッセージを無期限に処理するスタンドアロンのC＃のサンプルのみを見つけることができます。これらの例を使用しようとすると、コードをどこに置くべきか分かりませんし、guiやその他すべてをブロックするだけです。

別のスレッドを使用せずに行うことは不可能かどうかは分かりませんが、私はZeroMQの非同期動作が余分なスレッドをコーディングせずに機能するという印象を受けました。たぶん私はzeromqコードをどこに置くか分からないかもしれませんが、おそらく本当に別のスレッドが必要です。

誰かが単純なpub/subサンプルを提供して、デフォルトのC＃winformアプリケーションにコードを実際に挿入する場所を指示できたら、非常に感謝しています。

Answer 1

あなたのプロジェクトでclrzmq ZeroMq wrapperを使用していることを前提としています。私が知っている限り、clrzmqを使用して単純なループでメッセージ非ブロックを受信することはできませんが、特定の時間（受信メソッドにタイムアウトを指定することによって）またはメッセージを受信するまで無期限にブロックされます。

しかし、ソケットを定期的にポーリングし、着信メッセージをQueueにプッシュするようにスレッドを設定することはかなり簡単です。たとえば、単純なWinForms Timerを使用して、保留中のメッセージを定期的にその（共有）Queueからデキューできます。ここでは、ねじ加入者の作業例です：

public class ZeroMqSubscriber 
{ 
    private readonly ZmqContext _zmqContext; 
    private readonly ZmqSocket _zmqSocket; 
    private readonly Thread _workerThread; 
    private readonly ManualResetEvent _stopEvent = new ManualResetEvent(false); 
    private readonly object _locker = new object(); 
    private readonly Queue<string> _queue = new Queue<string>(); 

    public ZeroMqSubscriber(string endPoint) 
    { 
     _zmqContext = ZmqContext.Create(); 
     _zmqSocket = _zmqContext.CreateSocket(SocketType.SUB); 
     _zmqSocket.Connect(endPoint); 
     _zmqSocket.SubscribeAll(); 

     _workerThread = new Thread(ReceiveData); 
     _workerThread.Start(); 
    } 

    public string[] GetMessages() 
    { 
     lock (_locker) 
     { 
      var messages = _queue.ToArray(); 
      _queue.Clear(); 
      return messages; 
     } 
    } 

    public void Stop() 
    { 
     _stopEvent.Set(); 
     _workerThread.Join(); 
    } 

    private void ReceiveData() 
    { 
     try 
     { 
      while (!_stopEvent.WaitOne(0)) 
      { 
       var message = _zmqSocket.Receive(Encoding.UTF8, 
           new TimeSpan(0, 0, 0, 1)); 
       if (string.IsNullOrEmpty(message)) 
        continue; 

       lock (_locker) 
        _queue.Enqueue(message); 
      } 
     } 
     finally 
     { 
      _zmqSocket.Dispose(); 
      _zmqContext.Dispose(); 
     } 
    } 
} 

Formあなたは、単に（この例では、Forms Timerを使用し、単にTextboxにメッセージデータを追加）、定期的にキューをポーリングから：

private readonly ZeroMqSubscriber _zeroMqSubscriber = 
     new ZeroMqSubscriber("tcp://127.0.0.1:5000"); 

void ReceiveTimerTick(object sender, EventArgs e) 
{ 
    var messages = _zeroMqSubscriber.GetMessages(); 
    foreach (var message in messages) 
     _textbox.AppendText(message + Environment.NewLine); 
}