网络上的NetMQ

Question

我试图在本地网络上的不同计算机上进行程序间的消息传递。该设计基本上是一个“服务器”，“多个”客户端，服务器向客户端发送消息，从客户端接收消息，客户端从服务器接收消息，彼此之间发送/接收客户端消息。

我正在尝试使用NetMQ来实现这一点，但我在网络上使用它时遇到了困难。我可以在一台机器上很好地运行服务器和客户端程序，但我不知道如何让它在网络上工作，而且我找不到任何NetMQ示例来显示这一点--为了明确的是，我可以找到的每个示例都使用"localhost“或其他IP地址---- "publisher”和"subscriber“(我使用的是Pub模式，但不管在示例中使用的模式是什么)。

一些使用客户端客户端框架的示例代码(这需要一个XSub框架，因为有多个发布者和订阅者)：

NetMQPoller poller = new NetMQPoller();
PublisherSocket clientPub = new PublisherSocket("tcp://*:4444");
XPublisherSocket xPub = new XPublisherSocket("tcp://*:5555");
XSubscriberSocket xSub = new XSubscriberSocket("tcp://*:4444");
Proxy proxy = new proxy(xSub, xPub, poller: poller);
proxy.Start();
SubscriberSocket clientSub = new SubscriberSocket("tcp://*:5555");
poller.Add(clientSub);
poller.Add(clientPub);
poller.RunAsync();

如果运行此操作，将得到“应用程序处于中断模式”，并出现以下错误：

NetMQ.Core.Transports.Tcp.TcpConnector.OutCompleted(SocketError HResult=0x80131500 Source=NetMQ StackTrace:在NetMQ.NetMQException socketError，Int32 bytesTransferred( NetMQ.Core.IOObject.OutCompleted(SocketError socketError，Int32 bytesTransferred) at NetMQ.Core.Utils.Proactor.Loop() at System.Threading.ExecutionContext.RunInternal(ExecutionContext Int32)，socketError回调，Object state，Boolean NetMQ.Core.Utils.Proactor.Loop)在en3 20#，回调，Object状态，在System.Threading.ExecutionContext.Run(ExecutionContext executionContext，ContextCallback回调，Object preserveSyncCtx)在System.Threading.ThreadHelper.ThreadStart()

如果我将clientSub的地址更改为"tcp://127.0.0.1:5555“，它将很好地启动。但是，由于订阅者需要从网络上的任何机器接收消息，所以无法工作。

我如何在NetMQ中做到这一点？有人能提供这方面的工作例子吗？

Answer 1

默认情况下，当使用套接字的构造函数是绑定或连接时，取决于套接字类型。发布服务器绑定(侦听)和订阅服务器连接。因此，将*用于订阅服务器是有意义的。

您正在使用tcp传输，因此在连接时必须指定远程对等端的ip地址。如果要连接到多个发布者，可以调用connect或为字符串提供由逗号分隔的多个地址。

底线，而不是*，localhost或127.0.0.1使用发行者机器的ip地址。

您可以使用@或>前缀覆盖构造函数的默认值。@绑定和>连接。

NetMQPoller poller = new NetMQPoller();
PublisherSocket clientPub = new PublisherSocket(">tcp://192.168.0.15:4444");
XPublisherSocket xPub = new XPublisherSocket("@tcp://*:5555");
XSubscriberSocket xSub = new XSubscriberSocket("@tcp://*:4444");
Proxy proxy = new proxy(xSub, xPub, poller: poller);
proxy.Start();
SubscriberSocket clientSub = new SubscriberSocket(">tcp://192.168.0.15:5555");
poller.Add(clientSub);
poller.Add(clientPub);
poller.RunAsync();

192.168.0.15是我的ip地址。你得把它改成你的ip。

Answer 2

我有一个简单的例子。我的“服务器”代码运行在亚马逊的EC2实例上。它监听端口11000并响应单个消息。

using System;
using NetMQ;
using NetMQ.Sockets;

class Program
{
    private static void Main()
    { 
        using (var server = new ResponseSocket("@tcp://*:11000"))
        {
            // the server receives, then sends
            Console.WriteLine("From Client: {0}", server.ReceiveFrameString());
            server.SendFrame("Hi Back");
            Console.ReadKey();
        }
    }
}

我的客户端代码在我的电脑上运行：

using System;
using NetMQ;
using NetMQ.Sockets;

class Program
{
    private static void Main()
    {
        using (var client = new RequestSocket(">tcp://3.13.186.250:11000"))
        {
            client.SendFrame("Hello");
            Console.WriteLine("From Server: {0}", client.ReceiveFrameString());
            Console.ReadKey();
        }
    }
}