用C＃winform創建zeromq pub/sub的示例

Question

我正嘗試在pub/sub模型中創建一個使用ZeroMQ（clrzmq .net bindings（x86）via nuget）的C＃Winform應用程序。

經過大量搜索之後，我只能找到獨立的C＃示例，其中的代碼使用while語句無限期處理新消息。當我嘗試使用這些示例時，我不知道該把代碼放在哪裏，它只是阻止了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); 
}