ZeroMQ 메시지브로커 C++ 예제

ZeroMQ 메시지브로커 C++ 예제

ZeroMQ 는 메시지브로커를 구현한 라이브러리 이다. ZeroMQ 는 C, C++, Javascript 등, 다양한 버전으로 배포된다. 임베디드 개발 과정에서 간단한 메시지브로커 라이브러리가 필요할때 추천되는 라이브러리 이다. 이글은 C++ 버전의 ZeroMQ를 이용한 메시지발행 (Publish) 와 수신 (Subscribe) 의 예제를 정리한다.

ZeroMQ 메시지브로커

본 예제는 hmartiro 의 오픈소스를 참고하였습니다.

ZeroMQ C++ pub / sub 샘플코드

메시지 발행하기

socket 인스턴스로 바인딩하여 메시지를 전송한다.

/**
* Example of ZeroMQ pub/sub usage for C++11.
*/

#include <zmqpp/zmqpp.hpp>
#include <iostream>
#include <chrono>
#include <thread>

using namespace std;

static const string PUBLISH_ENDPOINT = "tcp://*:4242";

int main(int argc, char *argv[]) {

  // Create a publisher socket
  zmqpp::context context;
  zmqpp::socket_type type = zmqpp::socket_type::publish;
  zmqpp::socket socket (context, type);

  // Open the connection
  cout << "Binding to " << PUBLISH_ENDPOINT << "..." << endl;
  socket.bind(PUBLISH_ENDPOINT);

  // Pause to connect
  this_thread::sleep_for(chrono::milliseconds(1000));

  while(true) {

    // Current time in ms
    unsigned long ms = chrono::system_clock::now().time_since_epoch() /
        chrono::milliseconds(1);

    string text = "Hello " + to_string(ms);

    // Create a message and feed data into it
    zmqpp::message message;
    message << text;

    // Send it off to any subscribers
    socket.send(message);
    cout << "[SENT] at " << ms << ": " << text << endl;

    this_thread::sleep_for(chrono::microseconds(1000));
  }

  // Unreachable, but for good measure
  socket.disconnect(PUBLISH_ENDPOINT);
  return 0;
}

1. 메시지 발행을 위한 socket 인스턴스를 생성한다.
2. 메시지 발행을 위해 bind 를 시도한다.
3. (binding 이 되면) 생성된 socket 인스턴스를 통해서 메시지를 1초 간격으로 보낸다.

메시지 구독하기

socket 인스턴스를 생성하여 발행자의 socket 에 connect 하고 메시지를 수신한다.

#include <zmqpp/zmqpp.hpp>
#include <iostream>
#include <chrono>

using namespace std;

static const string PUBLISHER_ENDPOINT = "tcp://localhost:4242";

int main(int argc, char *argv[]) {

  // Create a subscriber socket
  zmqpp::context context;
  zmqpp::socket_type type = zmqpp::socket_type::subscribe;
  zmqpp::socket socket(context, type);

  // Subscribe to the default channel
  socket.subscribe("");

  // Connect to the publisher
  cout << "Connecting to " << PUBLISHER_ENDPOINT << "..." << endl;
  socket.connect(PUBLISHER_ENDPOINT);

  while(true) {

    // Receive (blocking call)
    zmqpp::message message;
    socket.receive(message);

    // Read as a string
    string text;
    message >> text;

    unsigned long ms = std::chrono::system_clock::now().time_since_epoch() /
            std::chrono::milliseconds(1);

    cout << "[RECV] at " << ms << ": \"" << text << "\"" << endl;
  }

  // Unreachable, but for good measure
  socket.disconnect(PUBLISHER_ENDPOINT);
  return 0;
}

1. socket 인스턴스를 생성하고 발행자 socket 과 connect 한다.
2. connect 가 되면, 메시지 수신을 반복한다.

makepluscode 의 github 에서 전체코드를 확인할 수 있다.

https://github.com/makepluscode/message-broker-examples/tree/main/001-zeromq-pubsub-cpp

GitHub - makepluscode/message-broker-examples

참고자료

이 글에서 참조한 hmartiro 의 예제 코드를 참고하자.

https://gist.github.com/hmartiro/df1eb214f77f549b3691

ZeroMQ pub/sub usage and latency timing for C++11

ZeroMQ 에 대해 자세히 알고 싶다면,  ZeroMQ 공식홈페이지를 살펴본다.

https://zeromq.org/

ZeroMQ