У меня проблема с ZeroMQ, и я считаю, что я не очень хорошо знаком с ней.ZeroMQ performance issue
Я пытаюсь создать очень простой сервис, когда несколько клиентов подключаются к серверу и отправляют запрос. Сервер отвечает на этот запрос.
Когда я использую комбинацию сокетов REQ-REP (клиент с использованием REQ, привязка сервера к сокету REP), я могу получить около 60 000 сообщений в секунду на стороне сервера (когда клиент и сервер находятся на одной машине) , При распределении между машинами каждый новый экземпляр клиента на другой машине линейно увеличивает количество сообщений в секунду на сервере и легко достигает 40 000+ с достаточным количеством экземпляров клиента.
Теперь РЭП сокет блокирует, таким образом, я следовал 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: Это, по сути, rrbroker от руководства ZMQ.
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());
}
}
}
}
Pieter, спасибо за ответ. Я добавил код. – Andy