我对 ZeroMQ 有疑问，我相信这是因为我对它不是很熟悉。
我正在尝试构建一个非常简单的服务，其中多个客户端连接到服务器并发送查询。服务器响应此查询。
当我使用 REQ-REP 套接字组合（客户端使用 REQ，服务器绑定到 REP 套接字）时，我能够在服务器端每秒获得接近 60,000 条消息（当客户端和服务器在同一台机器上时）。当分布在机器上时，不同机器上的每个新客户端实例都会线性增加服务器每秒的消息数，并且在有足够的客户端实例的情况下很容易达到 40,000+。
现在 REP 套接字阻塞了，所以我遵循 ZeroMQ 指南并使用了 rrbroker 模式（http://zguide.zeromq.org/cs:rrbroker）：
REQ (client) <----> [server ROUTER -- DEALER --- REP (workers running on different threads)]
但是，这完全破坏了性能。跨机器运行时，我在服务器上每秒只能收到大约 4000 条消息。不仅如此，在不同机器上启动的每个新客户端都会降低其他所有客户端的吞吐量。
我很确定我在做一些愚蠢的事情。我想知道这里的 ZeroMQ 专家是否可以指出任何明显的错误。谢谢！
编辑：根据建议添加代码。我正在使用 clrzmq nuget 包（https://www.nuget.org/packages/clrzmq-x64/）
这是客户端代码。计时器计算每秒收到多少响应。
for (int i = 0; i < numTasks; i++) { Task.Factory.StartNew(() => Client(), TaskCreationOptions.LongRunning); }
void Client()
{
using (var ctx = new Context())
{
Socket socket = ctx.Socket(SocketType.REQ);
socket.Connect("tcp://192.168.1.10:1234");
while (true)
{
socket.Send("ping", Encoding.Unicode);
string res = socket.Recv(Encoding.Unicode);
}
}
}
服务器 - 案例 1：服务器跟踪每秒收到的请求数
using (var zmqContext = new Context())
{
Socket socket = zmqContext.Socket(SocketType.REP);
socket.Bind("tcp://*:1234");
while (true)
{
string q = socket.Recv(Encoding.Unicode);
if (q.CompareTo("ping") == 0) {
socket.Send("pong", Encoding.Unicode);
}
}
}
使用此设置，在服务器端，我可以看到每秒收到大约 60,000 个请求（当客户端在同一台机器上时）。当在不同的机器上时，每个新客户端都会按预期增加服务器收到的请求数。
服务器案例 2：这本质上是来自 ZMQ 指南的 rrbroker。
void ReceiveMessages(Context zmqContext, string zmqConnectionString, int numWorkers)
{
List<PollItem> pollItemsList = new List<PollItem>();
routerSocket = zmqContext.Socket(SocketType.ROUTER);
try
{
routerSocket.Bind(zmqConnectionString);
PollItem pollItem = routerSocket.CreatePollItem(IOMultiPlex.POLLIN);
pollItem.PollInHandler += RouterSocket_PollInHandler;
pollItemsList.Add(pollItem);
}
catch (ZMQ.Exception ze)
{
Console.WriteLine("{0}", ze.Message);
return;
}
dealerSocket = zmqContext.Socket(SocketType.DEALER);
try
{
dealerSocket.Bind("inproc://workers");
PollItem pollItem = dealerSocket.CreatePollItem(IOMultiPlex.POLLIN);
pollItem.PollInHandler += DealerSocket_PollInHandler;
pollItemsList.Add(pollItem);
}
catch (ZMQ.Exception ze)
{
Console.WriteLine("{0}", ze.Message);
return;
}
// Start the worker pool; cant connect
// to inproc socket before binding.
workerPool.Start(numWorkers);
while (true)
{
zmqContext.Poll(pollItemsList.ToArray());
}
}
void RouterSocket_PollInHandler(Socket socket, IOMultiPlex revents)
{
RelayMessage(routerSocket, dealerSocket);
}
void DealerSocket_PollInHandler(Socket socket, IOMultiPlex revents)
{
RelayMessage(dealerSocket, routerSocket);
}
void RelayMessage(Socket source, Socket destination)
{
bool hasMore = true;
while (hasMore)
{
byte[] message = source.Recv();
hasMore = source.RcvMore;
destination.Send(message, message.Length, hasMore ? SendRecvOpt.SNDMORE : SendRecvOpt.NONE);
}
}
工作池的启动方法是：
public void Start(int numWorkerTasks=8)
{
for (int i = 0; i < numWorkerTasks; i++)
{
QueryWorker worker = new QueryWorker(this.zmqContext);
Task task = Task.Factory.StartNew(() =>
worker.Start(),
TaskCreationOptions.LongRunning);
}
Console.WriteLine("Started {0} with {1} workers.", this.GetType().Name, numWorkerTasks);
}
public class QueryWorker
{
Context zmqContext;
public QueryWorker(Context zmqContext)
{
this.zmqContext = zmqContext;
}
public void Start()
{
Socket socket = this.zmqContext.Socket(SocketType.REP);
try
{
socket.Connect("inproc://workers");
}
catch (ZMQ.Exception ze)
{
Console.WriteLine("Could not create worker, error: {0}", ze.Message);
return;
}
while (true)
{
try
{
string message = socket.Recv(Encoding.Unicode);
if (message.CompareTo("ping") == 0)
{
socket.Send("pong", Encoding.Unicode);
}
}
catch (ZMQ.Exception ze)
{
Console.WriteLine("Could not receive message, error: " + ze.ToString());
}
}
}
}