Use ZeroMQ

이전 포스팅들에서 메시징에 대해 계속 적고 있다보니, ZeroMQ (ØMQ) 에 대해 짧게 어떤 물건인지 보려고 합니다.

 

ZeroMQ

공식 문서에 적혀있는 내용을 그대로 가져와보자면, ZeroMQ 는 고성능의 비동기 메세징 라이브러리이고, 분산 어플리케이션 또는

동시성 어플리케이션에서 사용하는 것에 초점이 맞춰져있습니다. 메세지 큐를 제공하긴 하지만, 다른 메세지 지향 미들웨어와 달리

전용 메세지 브로커 없이 ZeroMQ 시스템을 사용할 수 있습니다.

 

ZeroMQ 는 TCP, in-process, inter-process, multicast, WebSocket 등의 다양한 전송 방식을 통해 pub/sub, request/reply,
client/server 등의 일반적인 메세징 패턴을 지원합니다. 프로세스간 메세징을 스레드 간 메세징처럼 단순하게 만들어주고,

이것은 깔끔하면서 모듈화하기 좋고 확장하기 간편합니다.

 

이름 앞에 Zero(Ø) 가 붙은 이유는 위에 적은 내용 중에 있듯이 브로커가 없는 것 외에도 대기시간, 비용, 관리 등이 없음을 나타낸다고
합니다. 아래에서 짧게 보겠지만 사용법이 굉장히 간단한데,  프로젝트에서 P2P 방식의 메세지 교환이 필요하다면 ZeroMQ 는
최적의 솔루션이 될 수 있을 것 같습니다.

 

 

위의 공식 사이트를 들어가면 위처럼 상당히 많은 언어를 ZeroMQ 는 라이브러리로 지원하고 있습니다.

여기에는 Libzmq 라는 애가 엮여있는데, 이 라이브러리는 다양한 언어에 바인딩 되어 있는 저수준 라이브러리입니다.

Libzmq 는 C++ 로 구현되어 있고 C-API 를 제공하는데, Libzmq 를 직접 다룰 일은 잘 없겠지만 그 내부를 까서 파악하고 싶다면

보는 것이 좋겠지만 우리는 prebuilt 되어 있는 각 언어의 라이브러리를 사용함으로써 쉽게 ZeroMQ 를 사용할 수 있습니다.

 

저는 Node.JS 에서 ZeroMQ 를 Fastify 와 묶어서 pub/sub 패턴에 대해 간단하게 살펴보려고 합니다.

 

npm install fastify zeromq@5
npm install @types/zeromq --save-dev

 

node 16 버전에서 진행했습니다. 현재 Node.JS 에서 제공되는 zeromq 는 6.0.0 버전이 나와있는데, 베타 버전이라고 써있기도 하지만

제 mac 환경에서 빌드가 실패하는데 간단히 보려는 과정에서 이것까지 해결하기엔 넌센스라 zeromq 5 버전을 사용했습니다.

fastify 모듈까지 설치해주고, TS 에서 사용할거라 @types 모듈까지 설치했습니다.

 

// zmq-server.ts
import * as zmq from 'zeromq';
import * as fastify from 'fastify';

const app = fastify();
const pubSocket = zmq.socket('pub');

app.post('/', (request, reply) => {
  pubSocket.send(`zmq ${JSON.stringify(request.body)}`);
  return reply.send('send message by zmq');
});

const main = async () => {
  pubSocket.bind('tcp://127.0.0.1:5000');
  await app.listen({ port: 3000 });
};

main();

 

zeromq 의 서버측, 그러니까 publish 를 하는 부분의 코드입니다.

fastify 를 통해 생성한 앱서버에서 POST 요청을 받았을 때 그 body 를 pub socket 을 통해 게시합니다.

 

// zmq-client.ts
import * as zmq from 'zeromq';

const subSocket = zmq.socket('sub');

subSocket.connect('tcp://127.0.0.1:5000');
subSocket.subscribe('zmq');

subSocket.on('message', msg => {
  console.info(msg.toString());
});

 

게시된 메세지를 수신할 수 있는 client 측 코드입니다.

서버측에서 생성한 pub socket 에 연결을 하고, zmq 로 시작하는 메세지에 대해 구독을 생성합니다.

전 많이 쓰던 redis 의 pub/sub 을 생각했다보니 이게 좀 특이하다고 생각한 부분인데, 게시하는 쪽에서 채널을 명시적으로 선언하는 것도
아닌데 subscirbe 가 왜 필요한거지? 했는데 pub socket 에서 보내는 메세지의 문자열의 앞부분이 일종의 채널과 같은 역할을 합니다.

 

subscirbe('zmq') 로 구독을 하는 것은 zmq 로 시작되는 메세지만 수신하겠다는 의미이므로, zmq 로 시작하는 메세지가 아니거나 혹은

subscribe 함수를 호출하는 부분을 제거하면 pub socket 으로부터 아무런 메세지도 받지 못하게 됩니다.

아무튼 여기까지 하면, 이벤트 핸들러를 추가해 message 이벤트에 대해 메세지를 수신할 수 있게 됩니다.

이때 받는 메세지는 Buffer 타입이라서 toString 메서드를 호출해 눈으로 이해할 수 있도록 변경해줬습니다.

 

이제 server, client 양쪽 코드를 각각 실행하고 3000 port 로 request body 는 { id: 1, name: "Tom" } 과 같이 POST 요청을

보내면 zmq {"id":1,"name":"Tom"} 이런 메세지를 수신하는 것을 확인할 수 있습니다.

zmq 가 앞에 포함된 문자열만 받을 수 있는 건 맞는데, 이런식이면 메세지를 받을 때마다 앞의 문자열을 걸러줘야하는게 좀 귀찮습니다.

그래서 앞의 topic 은 제외하고 메세지를 받을 수 있도록 약간 수정을 해보겠습니다.

 

// zmq-server.ts

...

app.post('/', (request, reply) => {
  pubSocket.send(['zmq', `${JSON.stringify(request.body)}`]);
  return reply.send('send message by zmq');
});

...

 

서버측에선 pub socket 을 통해 메세지를 보낼 때 배열로 보내도록 수정합니다. 첫번째 인자가 topic, 두번째 인자가 메세지를
의미하게 됩니다.

 

// zmq-client.ts

...

subSocket.on('message', (topic, message) => {
  console.info('worker', message.toString());
});

 

이제 client 측에선 topic, message 두개의 인자로 메세지를 수신할 수 있게 됩니다.

이렇게 수정하고 이전과 동일하게 POST 요청을 만들어서 게시를 하면, worker {"id":1,"name":"Tom"} 이런 메세지를 받습니다.

 

아주 짧은 코드였지만 ZeroMQ 를 사용하면서 느낀 것은 간단한것이야 뭐 하는 일이 별로 없으니 당연한 것 같기도 합니다만,

뭐 특별한 에러를 뱉는 경우도 거의 없습니다. 예를 들면 아직 서버측 코드를 실행하지 않았거나 그냥 서버 프로세스를 종료하는 경우
5000 포트에 pub socket 이 바인딩되지 않았지만 sub socket 가 pub socket 에 연결을 하는 부분에서 에러가 발생하지 않습니다.

 

이런 부분은 ZeroMQ 가 자체적으로 일정 시간 간격 동안 연결을 자동으로 시도하는 retry 메커니즘을 갖고 있기 때문입니다.

라이브에서 운영하는 가정을 하면 노드가 다운되거나 재시작되는 상황은 꽤 있을 수 있는데, 이런 환경에서 상당히 유용할 것 같습니다.

하나의 커다란 관리 포인트가 줄어들 수 있는 장점까지 가져갈 수 있으니까요.

 

메세지 브로커 없이 P2P 방식으로 메세지를 전달해야하만 하는 상황에선 별도의 설치 필요없이 라이브러리에 prebuilt 된 ZeroMQ 를

사용하는 것은 고려할 수 있는 충분히 좋은 옵션인 것 같습니다.