ZeroMQ для C++: высокопроизводительные сообщения
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() {