BullMQ에서의 Pub/Sub

 

BullMQ에서는

 

job을 Queue에 추가 → BullMQ가 Redis에 저장 → Worker가 Job을 꺼내 실행 → 상태 관리까지 자동처리

위 프로세스로 동작하게 된다.

 

여기서 어떻게 worker가 job이 쌓인걸 알게 되는지 문득 궁금해졌다.

BullMQ에서 Worker는 Pub/Sub 메커니즘을 통해 새로운 Job이 Queue에 추가된 것을 알게 된다.

 

동작 과정을 보게 되면

1. Job 추가 (Producer): Producer(Job을 생성하는 주체)가 queue.add() 메서드를 호출하여 새로운 Job을 Queue에 추가한다.
2. Redis에 저장 및 알림 (BullMQ)
   1. 이 때 BullMQ는 이 Job을 JSON 형태로 직렬화하여 Redis의 리스트(list)에 RPUSH 명령어를 사용하여 추가한다. 큐 이름이 my_queue라면 Redis에는 bull:my_queue:wait라는 리스트가 생성되고 여기에 Job들이 차곡차곡 쌓이게 된다.
   2. Job이 리스트에 성공적으로 추가되면, BullMQ는 PUBLISH 명령어를 사용하여 Redis의 Pub/Sub 채널에 새로운 Job이 추가되었다는 알림 메세지를 발행(Publish) 한다. 이 알림은 newJob과 같은 이벤트 메시지이다. 이 메시지는 bull:my-queue:new-job과 같은 채널에 발행된다.
3. Worker 대기 및 수신 (Consumer)
   1. Worker는 SUBSCRIBE 명령어를 사용하여 Redis의 bull:my_queue:new-job 채널을 구독하고 있다.
   2. Worker는 Queue에 새로운 Job이 있는지 주기적으로 확인하는 (Polling) 대신, 이 알림 메시지를 실시간으로 받기 위해 대기한다.
4. Job 처리 (Worker)
   1. Worker가 Pub/Sub 채널로부터 newJob 알림을 받으면, Redis의 리스트에서 BLPOP 명령어를 사용하여 가장 먼저 들어온 Job을 꺼내온다. (BullMQ v5+)
      • BLPOP은 Blocking Left POP의 약자로 이 명령어는 큐의 왼쪽 끝에서 요소를 꺼내오고 큐가 비어 있으면 새로운 요소가 추가될 때까지 기다리는 Blocking 기능을 수행한다. 즉 선입선출을 위해 Job을 큐의 오른쪽 끝에 추가하고(RPUSH), 왼쪽에서 꺼내는 (BLPOP) 방식을 사용했다.(이전 버전에서는 BRPOPLPUSH 방식을 사용했다고 함)
   2. 이 명령은 Job을 꺼내는 동시에 다른 대기열(예: active 상태의 Job을 담는 대기열)로 옮겨 놓아, 여러 worker가 같은 Job을 가져가서 중복 처리하는 것을 방지한다.

이러한 이벤트 기반(Event-driven) 방식은 Worker가 Job이 있는지 계속해서 확인하는 폴링(Polling) 방식에 비해 훨씬 효율적이다. 폴링은 불필요한 네트워크 트래픽과 CPU 자원을 소모하지만, Pub/Sub 방식은 새로운 Job이 발생했을 때만 Worker를 깨워 작업을 처리하게 한다. 이는 자원 낭비를 최소화하고 시스템의 응답성을 높이는 핵심적인 메커니즘이다.

여기서 핵심이 되는 Pub/Sub에 대해 좀 더 알아보자면 Pub/Sub은 중앙에 메시즈 브로커(Message Broker) 또는 channel이라는 중간 지점을 두고 동작한다.

 

1. 발행자(Publisher) : 메시지를 발행하는 주체이다. 발행자는 메시지를 특정 토픽이나 채널에 보낸다. 발행자는 누가 메시지를 수신하는지 알 필요가 없으며, 오직 브로커에 메시지를 보내는 역할만 한다.
2. 구독자(Subscriber) : 메시지를 수신하는 주체이다. 구독자는 자신이 관심 있는 토픽을 구독(Subscribe)한다. 메시즈 브로커는 해당 토픽에 새로운 메시지가 발행되면, 이를 구독하고 있는 모든 구독자에게 메시지를 전달한다.
3. 메시지 브로커 : 발행자와 구독자 사이에서 메시지를 중개하는 역할을 한다. 브로커는 발행자가 보낸 메시지를 받아 해당 토픽을 구독하는 모든 구독자에게 메시지를 전달한다.

Pub/Sub의 장점으로는

• 결합도 낯춤 : 발행자와 구독자가 서로 직접적으로 연결되지 않고 브로커를 통해 통신하기 때문에, 시스템의 결합도가 낮아진다. 한쪽이 장애가 발생해도 다른 쪽은 영향을 받지 않는다.
• 확장성 : 새로운 구독자를 추가하거나 기존의 구독자를 삭제하더라도 발행자의 코드를 수정할 필요가 없다. 이는 시스템을 유연하게 확장하는데 큰 도움이 된다.
• 비동기 처리 : 메시지 발행 시점에 구독자가 온라인 상태가 아니더라도, 메시지 브로커가 메시지를 보관했다가 구독자가 다시 연결되면 전달할 수 있다. 이는 실시간 처리가 필수가 아닌 경우 효율적이다.

주의점

1. 최대 연결 개수

redis 서버는 동시에 처리할 수 있는 클라이언트 연결 수에 제한이 있다. 기본 설정값인 maxclients는 보통 10,000이다. 운영체제 자체의 파일 디스크립터 제한도 영향제한도 영향을 미친다.

BullMQ는 일반적으로 하나의 Worker가 여러 개의 Redis 연결을 사용한다. Job을 가져오기 위한 연결, Pub/Sub 알림을 위한 연결, 상태 업데이트를 위한 연결 등이다. 따라서 Worker 수를 늘리면 Redis 서버에 대한 전체 연결 수가 빠르게 증가한다. 이러한 제한은 시스템의 규모와 설계에 따라 조정해야 하는데, 많은 수의 Worker를 운영할 경우 Redis 서버의 maxclients 값을 늘리거나, 클러스터링(Clustering)을 통해 부하를 분산시켜야 한다.