zeromq сообщение шаблон для многих для многих запрос/ответ

Question

Я очень новичок в ZeroMQ. Я прочитал руководство и в настоящее время просматриваю примеры, а также просматриваю другую соответствующую информацию в Интернете. У меня есть нерешительность с тем, какие шаблоны сообщений использовать или если я должен использовать комбинацию из двух шаблонов.

У меня есть существующее прикладное программное обеспечение, в котором есть система обмена сообщениями для домашних пользователей, которая нуждается в замене. У меня есть довольно простая архитектура:

|Client|<----->|driver1| 
      | 
      |---|driverN| 

только один «клиент» подключается к водителю в то время в настоящее время, и может быть много водителей.

(в действительности, клиент, в данном случае, это действительно не мое клиентское приложение, но посредник сортов Для этой дискуссии, он может рассматриваться в качестве клиента.)

Сообщения:

1. Клиент выдает команды водителю.
2. Драйверы возвращают информацию о состоянии/состоянии в ответ на команды.
3. Драйверы формируют элементы данных (то есть не информацию о состоянии/состоянии)
4. Некоторые сообщения клиентов отправляются на все подключенные устройства, некоторые из них направлены только на один драйвер.

Драйверы могут находиться в одной системе или удаленно в локальной сети. Это не общедоступная сеть.

В настоящее время я думаю, что у меня будет паб и вспомогательный сокет на каждом драйвере и на паб/паб на клиенте. Сообщения не должны удаляться после подключения. Я предполагаю, что клиент будет подписываться на разные типы данных драйверов, а затем драйвер подписывается на сообщения команд клиентов.

Важные соображения: низкая латентность, наименьшая возможная пропускная способность.

Буду признателен за любые предложения или рекомендации! Заранее спасибо!

Answer 1

Вы выбрали отличное учебное упражнение, это точно!

Прочитайте их, они предоставляют базовую реализацию для запроса/ответа с использованием настраиваемого прокси-маршрутизатора-маршрутизатора с опросом и должны обращаться к вашей проблеме «клиент-устройство».

• https://github.com/imatix/zguide/blob/master/examples/C/lbbroker.c
  
• https://github.com/imatix/zguide/blob/master/examples/C/lbbroker2.c

Решение является синхронным, поэтому любой запрос, отправленный из клиентских блоков до тех пор, пока не получит ответ. Лично я хотел бы использовать async для запроса и ответа для полной гибкости, но это решение является более сложным. Однако есть примеры в книге под названием Freelance и Dealer/Router, которые иллюстрируют запрос/ответ асинхронного запроса.

Вот пример синхронного запроса/ответа «многие-ко-многим». Вы ДОЛЖНЫ знать, как обертывание ZeroMq работает, чтобы полностью понять механику этого подхода; см. пример lbbroker1.

Client

Установить личность клиента с setIdentity(); важно для маршрутизации ответа.
Клиент отправляет запросы на device1, device2 и т. Д. В петлю; если устройство существует, сообщения статуса, возвращенные с определенного устройства, в противном случае «нет устройства» возвращается клиенту.

Socket client = context.socket(ZMQ.REQ); 
    client.setIdentity("client1".getBytes()); 
    client.connect("tcp://localhost:5550"); 

    for(int i = 0; i < 5; i++){ 
     client.send("device" + i); 
     String reply = client.recvStr(); 
     log("Received message: " + reply); 
     Thread.currentThread().sleep(1000); 
} 

Устройство

Устройство устанавливает идентификатор клиента, как и для уникальной маршрутизации.
Устройство отправляет device.send("DEVICEREADY") на сервер, чтобы сигнализировать о доступности в сети.
Устройство делает recvStr() три раза, чтобы прочитать полный конверт с сервера.

String deviceId = "device1" 
Socket device = context.socket(ZMQ.REQ); 
device.setIdentity(deviceId.getBytes()); 

device.connect("tcp://localhost:5560"); 
device.send("DEVICEREADY"); 

while (!Thread.currentThread().isInterrupted()) { 
    String clientAddress = device.recvStr();    
    String empty = device.recvStr(); 
    String clientRequest = device.recvStr(); 

    //create envelope to send reply to same client who made request   
    device.sendMore(clientAddress); 
    device.sendMore(""); 
    device.send("stauts on " + deviceId + " is ok"); 
} 

Сервер (маршрутизатор/маршрутизатор)

Пользовательский прокси с использованием маршрутизатора гнезда; клиенты подключаются к внешнему разъему ROUTER, а устройства подключаются к серверному маршрутизатору. Опрос серверов в обоих сокетах для сообщений.

Context context = ZMQ.context(1);  
Socket frontend = context.socket(ZMQ.ROUTER); 
Socket backend = context.socket(ZMQ.ROUTER); 
frontend.bind("tcp://localhost:5550"); 
backend.bind( "tcp://localhost:5560");   

Poller poller = new Poller(2); 
poller.register(frontend, Poller.POLLIN); 
poller.register(backend, Poller.POLLIN); 

while (!Thread.currentThread().isInterrupted()) { 
    poller.poll(); 

    //frontend poller 
    if (poller.pollin(0)) { 
     String clientId = frontend.recvStr(); 
     String empty = frontend.recvStr(); //empty frame 
     String deviceId = frontend.recvStr(); 

     //if client is requesting to talk to nonexistent deviceId, 
     //return message "no device", otherwise, create envelope and send 
     //request on backend router to device. 
     if(deviceMap.get(deviceId) == null){ 
     frontend.sendMore(clientId); 
     frontend.sendMore(""); 
     frontend.send("no deviceId: " + deviceId); 
     } else { 
     //request envelope addressed to specific device 
     backend.sendMore(deviceId); 
     backend.sendMore(""); 
     backend.sendMore(clientId); 
     backend.sendMore(""); 
     backend.send("hello from " + clientId); 
     } 
    } 

    //backend poller 
    if(poller.pollin(1)){ 
     String deviceId = backend.recvStr(); 
     String empty = backend.recvStr(); 
     String clientId = backend.recvStr(); 

     //device signaling it's ready 
     //store deviceId in map, don't send a response 
     if(clientId.equals("DEVICEREADY")) 
      deviceMap.put(deviceId, deviceId); 

     else { 
      //the device is sending a response to a client 
      //create envelope addressed to client, send on frontend socket 
      empty = backend.recvStr(); 
      String reply = backend.recvStr(); 
      frontend.sendMore(clientId); 
      frontend.sendMore(""); 
      frontend.send(reply); 
     } 
    } 
}