01. ZeroMQ - "сокеты на стероидах"

ZeroMQ (ZMQ) - библиотека обмена сообщениями. 

Быстрая, компактная, удобная. Бесплатная.

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

ZMQ – кроссплатформенная библиотека. Реализована для множества операционных систем.
ZMQ написана на «чистом С». И именно "сишный" интерфейс  API позволил реализовать связь с ней практически из любого языка программировании. Ссылка на список: http://zeromq.org/bindings:_start

ZMQ - это не просто транспорт. Это - инструмент для создания сетевой архитектуры.
ZMQ позволяет в рамках одного приложения создать и отладить модель сложной сетевой архитектуры. Реализуется это очень просто: с помощью многонитевого приложения, когда отдельная нить имитирует отдельный процесс.
В ZMQ встроены средства буферизации сообщений. ZMQ, будучи реализована асинхронной, может работать как в синхронном, так и в асинхронном режиме.
У ZeroMQ пропускная способность выше, чем у TCP/IP, хотя ZeroMQ работает over TCP/IP...Поискал - "каким же образом?" ...нашел кое-какое описание, вроде специальной упаковки сообщений ZMQ в пакеты tcp.



Чтобы много не разговаривать, напишем клиент-сервное приложение.

Сервер получает от клиентов целое число, возводит его в квадрат и возвращает обратно клиенту. Клиент, соответственно,  в цикле отправляет серверу число, получает результат в виде квадрата числа. То есть, сетевая конфигурация имеет вот такую схему:

Поехали.
Код сервера:

program zmq_hw_Server;

{$APPTYPE CONSOLE}

uses
  SysUtils, ZMQ_h;
var
  fContext: Pointer;
  fResponder: Pointer;
  fStatus: Cardinal;
  fInValue: Cardinal;
  fOutValue: UInt64;

begin
  fContext := zmq_ctx_new();
  fResponder := zmq_socket(fContext, ZMQ_REP);
  fStatus := zmq_bind(fResponder, 'tcp://*:5555');
  assert(fStatus = 0);
  Writeln('Starting...');
  while (True) do begin
    zmq_recv(fResponder, @fInValue, SizeOf(fInValue), 0);
    Writeln('Received: ', fInValue);
    fOutValue := Int64(fInValue) * Int64(fInValue);
    zmq_send(fResponder, @fOutValue, SizeOf(fOutValue), 0);
  end;

end.

Код клиента:

program zmq_hw_Client;

{$APPTYPE CONSOLE}

uses
  SysUtils, ZMQ_h;

const
  c_iter = 999999;
var
  context: Pointer;
  fRequester: Pointer;
  i: Integer;
  f_ScrValue: Cardinal;
  f_qValue: UInt64;

begin
  Writeln('Client starting...');
  Randomize();

  context := zmq_ctx_new();
  fRequester := zmq_socket(context, ZMQ_REQ);
  zmq_connect(fRequester, 'tcp://localhost:5555');

  for i := 0 to c_iter do begin
    f_ScrValue := Random(High(Integer));
    Writeln('Sending ', f_ScrValue);
    zmq_send(fRequester, @f_ScrValue, SizeOf(f_ScrValue), 0);
    zmq_recv(fRequester, @f_qValue, SizeOf(f_qValue), 0);
    Writeln('Received ', f_qValue);
  end;

  zmq_close(fRequester);
  zmq_ctx_destroy(context);
end.

Для компиляции требуется подключить файл zmq_h.pas. Для запуска приложения требуется libzmq.dll. Эти файлы можно скачать здесь: http://delphi-and-zeromq.blogspot.ru/2014/10/pas-dll.html

Запуск. 
Запускаем одного или несколько клиентов. Если посмотреть в отладчике, клиент выполнил коннект, отправил сообщение и ждет ответа:

    zmq_recv(fResponder, @fInValue, SizeOf(fInValue), 0);

Затем запускаем сервер. Все сразу ожило, и клиенты и сервер!

Если сначала запустить сервер, а потом - клиентов, то тоже все будет работать.
Видно, что скорость работы очень высока и фактически ограничена скоростью оператора Writeln(). Я убрал вывод (чтобы не тормозило) и добавил измерение времени. При параллельной работе тридцати клиентов сервер обрабатывает более 40 000 запросов в секунду. (испытания проводились на одной машине с 4х - ядерным процессором) .

Такой режим работы называется "Запрос-Ответ". Он определяется значениями ZMQ_REQ (запрос) и ZMQ_REP (ответ), которые указываются при создании сокетов ZMQ.

fResponder := zmq_socket(fContext, ZMQ_REP);
...
fRequester := zmq_socket(context, ZMQ_REQ);
...

Стоит отметить, что запускать несколько серверов смысла нет: работать будет только самый первый из них, остальные будут ждать, когда освободится порт tcp:5555.

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

 zmq_connect(requester, 'tcp://Host1:5555'); // Сервера на разных доменах
 zmq_connect(requester, 'tcp://Host2:5555');

или

 zmq_connect(requester, 'tcp://Host1:5555'); // Сервера на разных портах
 zmq_connect(requester, 'tcp://Host1:5556');
 

  - и все. Запросы будут поочередно отсылаться разным серверам.

Интересно, что длина сообщений может быть любой (почти: Integer), просто короткие сообщения доставляются прямо в теле сообщения, а для длинных специально автоматически выделяется память и в сообщении - адрес начала блока.
После чтения следующего сообщения предыдущее "пропадает".

Таким образом, только что был рассмотрен режим обмена сообщениями по схеме "Запрос-Ответ" (Request-Reply).

Схема работы "Запрос-Ответ": сервер слушает сокет, принимает сообщения от всех желающих, и отвечает на запросы.

...
~~~~~~~~~~~~~~~~~~~~~~~~

Бывает, что нужно по-другому. Сервер должен оповестить клиентов о каком-то событии.
Клиенты отправляют серверу заявку, что они готовы получать сообщения о наступившем событии ("подписываются" на событие).
Одних клиентов могут интересовать одни события, других - другие.
А сервера вообще не волнует, кому из клиентов что нужно. Сервер - это как бы радиоприемник, вещающий в эфир. Кто слушает - молодец. А кто не слушает - тот ССЗБ.

Такая схема называется "Издатель - Подписчик" (Publisher-Subscriber).

Пример.

Автоматическая метеостанция измеряет температуру, атмосферное давление и скорость ветра. Результаты измерений время от времени (например, после завершения цикла измерений) передаются "всем заинтересованным лицам".
Кого-то интересует всё, кому-то нужна температура, кого-то волнует только скорость ветра.

Пишем код метеостанции, то есть, сервер-издатель: (продолжение)