c# - ZeroMQ 性能问题

Question

我对 ZeroMQ 有疑问，我认为这是因为我对它不是很熟悉。

我正在尝试构建一个非常简单的服务，其中多个客户端连接到服务器并发送查询。服务器响应此查询。

当我使用 REQ-REP 套接字组合时(客户端使用 REQ，服务器绑定(bind)到 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());
            }
        }
    }
}

Answer 1

您能否发布一些源代码或至少对您的测试用例进行更详细的解释？通常，构 build 计的方法是一次进行一项更改，并在每次更改时进行测量。您始终可以从已知的工作设计逐步过渡到更复杂的设计。