1. Что это и зачем нужна

ZeroMQ (также известная как ØMQ, 0MQ или zmq) — это высокопроизводительная асинхронная библиотека обмена сообщениями, предназначенная для использования в распределенных и параллельных приложениях. В отличие от традиционных брокеров сообщений (таких как RabbitMQ или Kafka), ZeroMQ не имеет центрального сервера. Вместо этого она предоставляет сокеты, которые могут соединяться напрямую друг с другом, образуя произвольные топологии (звезда, шина, кластер и т.д.). Библиотека реализует различные паттерны обмена: "запрос-ответ" (REQ/REP), "издатель-подписчик" (PUB/SUB), "конвейер" (PUSH/PULL) и другие.

Зачем нужна ZeroMQ? В современном мире программное обеспечение редко работает в изоляции. Микросервисы, распределенные вычисления, системы реального времени, чаты, финансовые трейдинговые платформы — все они требуют быстрого и надежного обмена данными между компонентами. ZeroMQ решает эту задачу "из коробки": она берет на себя всю сложность управления сетевыми соединениями, буферизацией, повторными попытками отправки и фрагментацией сообщений. Разработчику нужно лишь выбрать подходящий паттерн обмена и вызвать несколько функций.

Ключевые преимущества ZeroMQ: невероятная скорость (библиотека написана на C++ и использует эффективные алгоритмы), минимальное потребление памяти, поддержка множества языков (включая C++, Python, Java, C#, Go и другие), а также встроенная поддержка шифрования (через расширение CurveZMQ) и прокси-устройств.

2. Установка

Для C++ нет единого менеджера пакетов, как pip для Python. Однако ZeroMQ можно установить несколькими способами:

Способ 1: Системный пакетный менеджер (Linux/macOS)

# Ubuntu/Debian
sudo apt-get install libzmq3-dev

# Fedora/RHEL
sudo dnf install zeromq-devel

# macOS (Homebrew)
brew install zeromq

Способ 2: Сборка из исходников

Скачайте последнюю версию с GitHub и соберите:

git clone https://github.com/zeromq/libzmq.git
cd libzmq
mkdir build && cd build
cmake ..
make -j4
sudo make install

Способ 3: vcpkg (Windows/Linux/macOS)

vcpkg install zeromq

После установки не забудьте подключить заголовочный файл zmq.hpp (C++ обертка) и линковаться с библиотекой libzmq (или zmq.lib на Windows).

3. Быстрый старт — минимальный рабочий пример

Самый простой пример — пара "сервер-клиент" с использованием паттерна REQ/REP. Клиент отправляет "Hello", сервер отвечает "World".

Сервер (server.cpp)

#include <zmq.hpp>
#include <string>
#include <iostream>

int main() {
    // Создаем контекст (один на процесс)
    zmq::context_t context(1);

    // Создаем сокет типа REP (ответчик)
    zmq::socket_t socket(context, zmq::socket_type::rep);
    socket.bind("tcp://*:5555");

    while (true) {
        zmq::message_t request;
        // Ждем запрос от клиента
        socket.recv(request, zmq::recv_flags::none);
        std::string msg(static_cast<char*>(request.data()), request.size());
        std::cout << "Received: " << msg << std::endl;

        // Отправляем ответ
        zmq::message_t reply(5);
        memcpy(reply.data(), "World", 5);
        socket.send(reply, zmq::send_flags::none);
    }
    return 0;
}

Клиент (client.cpp)

#include <zmq.hpp>
#include <string>
#include <iostream>

int main() {
    zmq::context_t context(1);
    zmq::socket_t socket(context, zmq::socket_type::req);
    socket.connect("tcp://localhost:5555");

    for (int i = 0; i < 5; ++i) {
        zmq::message_t request(5);
        memcpy(request.data(), "Hello", 5);
        std::cout << "Sending Hello " << i+1 << std::endl;
        socket.send(request, zmq::send_flags::none);

        zmq::message_t reply;
        socket.recv(reply, zmq::recv_flags::none);
        std::string msg(static_cast<char*>(reply.data()), reply.size());
        std::cout << "Received " << msg << std::endl;
    }
    return 0;
}

Сборка:

g++ -std=c++11 server.cpp -o server -lzmq
g++ -std=c++11 client.cpp -o client -lzmq

Запустите сначала сервер, затем клиент. Вы увидите, как они обмениваются сообщениями.

4. Основные методы/функции/классы

В C++ обертке ZeroMQ (zmq.hpp) основными классами являются context_t, socket_t и message_t. Рассмотрим 8 ключевых методов.

4.1. zmq::context_t::context_t(int io_threads)

Сигнатура: context_t(int io_threads = 1)

Что делает: Создает контекст ZeroMQ. Контекст управляет всеми сокетами в одном процессе. Параметр io_threads задает количество потоков ввода-вывода.

Пример:

zmq::context_t context(2); // 2 потока для обработки I/O

4.2. zmq::socket_t::socket_t(context_t& context, socket_type type)

Сигнатура: socket_t(context_t& context, socket_type type)

Что делает: Создает сокет заданного типа. Типы: req, rep, pub, sub, push, pull, pair, dealer, router и другие.

Пример:

zmq::socket_t pub_socket(context, zmq::socket_type::pub);
zmq::socket_t sub_socket(context, zmq::socket_type::sub);

4.3. void bind(const std::string& endpoint)

Сигнатура: void bind(const std::string& endpoint)

Что делает: Привязывает сокет к определенному сетевому адресу (endpoint). Формат: tcp://*:5555, ipc:///tmp/feeds/0, inproc://my-pipe.

Пример:

socket.bind("tcp://*:5555");

4.4. void connect(const std::string& endpoint)

Сигнатура: void connect(const std::string& endpoint)

Что делает: Подключает сокет к удаленному адресу. В отличие от TCP, порядок bind/connect не важен — ZeroMQ сама установит соединение, когда обе стороны будут готовы.

Пример:

socket.connect("tcp://localhost:5555");

4.5. void send(message_t& msg, send_flags flags)

Сигнатура: void send(message_t& msg, send_flags flags = send_flags::none)

Что делает: Отправляет сообщение. Флаги: send_flags::none, send_flags::dontwait (неблокирующая отправка), send_flags::sndmore (если сообщение состоит из нескольких частей).

Пример:

zmq::message_t msg(10);
memcpy(msg.data(), "Hello", 5);
socket.send(msg, zmq::send_flags::dontwait); // Неблокирующая отправка

4.6. void recv(message_t& msg, recv_flags flags)

Сигнатура: void recv(message_t& msg, recv_flags flags = recv_flags::none)

Что делает: Получает сообщение. Флаги: recv_flags::none, recv_flags::dontwait (неблокирующий прием).

Пример:

zmq::message_t msg;
if (socket.recv(msg, zmq::recv_flags::dontwait)) {
    // Обработка полученного сообщения
}

4.7. void setsockopt(int option, const void* optval, size_t optvallen)

Сигнатура: void setsockopt(int option, const void* optval, size_t optvallen)

Что делает: Устанавливает опции сокета. Например, ZMQ_SUBSCRIBE для подписки на топики, ZMQ_RCVHWM для ограничения размера очереди приема.

Пример:

// Подписка на все топики (для SUB сокета)
socket.setsockopt(ZMQ_SUBSCRIBE, "", 0);

// Установка максимального размера очереди отправки
int hwm = 1000;
socket.setsockopt(ZMQ_SNDHWM, &hwm, sizeof(hwm));

4.8. zmq::message_t::message_t(size_t size)

Сигнатура: message_t(size_t size = 0)

Что делает: Создает пустое сообщение заданного размера. Данные можно записать через memcpy или data().

Пример:

zmq::message_t msg(1024); // Сообщение на 1024 байта
char* data = static_cast<char*>(msg.data());
strcpy(data, "Hello, world!");

5. Реальный практический пример (100-200 строк кода)

Рассмотрим более сложный пример: распределенный сборщик логов. У нас будет несколько "издателей" (PUB), которые отправляют логи, и один "подписчик" (SUB), который их собирает и выводит в консоль. Для демонстрации также добавим прокси-устройство для масштабирования.

Прокси-устройство (proxy.cpp)

#include <zmq.hpp>
#include <string>
#include <iostream>

int main() {
    zmq::context_t context(1);

    // Фронтенд: подключаемся к издателям
    zmq::socket_t frontend(context, zmq::socket_type::xsub);
    frontend.bind("tcp://*:5559");

    // Бэкенд: подключаемся к подписчикам
    zmq::socket_t backend(context, zmq::socket_type::xpub);
    backend.bind("tcp://*:5560");

    // Запускаем прокси
    zmq::proxy(frontend, backend, nullptr);
    return 0;
}

Издатель (publisher.cpp)

#include <zmq.hpp>
#include <string>
#include <chrono>
#include <thread>
#include <iostream>

int main() {
    zmq::context_t context(1);
    zmq::socket_t publisher(context, zmq::socket_type::pub);
    publisher.connect("tcp://localhost:5559");

    int id = 1;
    while (true) {
        // Формируем сообщение: "топик:данные"
        std::string topic = "log";
        std::string data = "Log message #" + std::to_string(id++);
        std::string message = topic + ":" + data;

        zmq::message_t msg(message.size());
        memcpy(msg.data(), message.c_str(), message.size());
        publisher.send(msg, zmq::send_flags::none);

        std::cout << "Sent: " << message << std::endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    return 0;
}

Подписчик (subscriber.cpp)

#include <zmq.hpp>
#include <string>
#include <iostream>

int main() {