网络上的 NetMQ

Question

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

我正在尝试为此使用 NetMQ，但我无法让它在网络上运行。我可以在一台机器上完美地运行服务器和客户端程序，但我不确定如何让它在网络上工作，而且我找不到任何显示这一点的 NetMQ 示例——要清楚，我能找到的每个示例都使用“localhost”或其他一些 IP 地址作为 “发布者”和“订阅者”（我使用的是 Pub-Sub 模式，但我'无论示例中使用何种模式，都看到了相同的情况）。

一些使用客户端-客户端框架的示例代码（这需要一个 XPub-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.NetMQException HResult=0x80131500 源=NetMQ StackTrace： 在 NetMQ.Core.Transports.Tcp.TcpConnector.OutCompleted（SocketError socketError, Int32 bytesTransferred) at NetMQ.Core.IOObject.OutCompleted(SocketError socketError, Int32 bytesTransferred) 在 NetMQ.Core.Utils.Proactor.Loop() 在 System.Threading.ThreadHelper.ThreadStart_Context（对象状态）在 System.Threading.ExecutionContext.RunInternal（ExecutionContext executionContext、ContextCallback 回调、对象状态、布尔值 preserveSyncCtx) 在 System.Threading.ExecutionContext.Run（ExecutionContext executionContext、ContextCallback 回调、对象状态、布尔值 preserveSyncCtx) 在 System.Threading.ExecutionContext.Run（ExecutionContext executionContext、ContextCallback 回调、对象状态）在 System.Threading.ThreadHelper.ThreadStart()

如果我将 clientSub 的地址更改为“tcp://127.0.0.1:5555”，它开始正常。但是，这行不通，因为订阅者需要从网络上的任何机器接收消息。

如何使用 NetMQ 做到这一点？谁能提供一个可行的例子？

Answer 1

默认情况下，当你使用socket的构造函数是bind或者connect，取决于socket的类型。发布者绑定（监听）和订阅者连接。所以对订阅者使用 * 是有意义的。

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

使用发布者机器的 ip 地址代替 *、localhost 或 127.0.0.1。

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

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

我有一个简单的例子。我的“服务器”代码在 Amazon 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();
        }
    }
}

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

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();
        }
    }
}